我遇到了一个问题，gearman工人在多个服务器上运行，我似乎无法解决。
当一个工作服务器脱机，而不是工作进程被取消时，问题发生，并导致所有其他工作进程出错并失败。
仅1个客户端和2个工作程序的示例-
客户:$client = new GearmanClient ();
$client->addServer ('192.168.1.200');
$client->addServer ('192.168.1.201');
$job = $client->do ('generate_tile', serialize ($arrData));
工作人员:
$worker = new GearmanWorker ();
$worker->addServer ('192.168.1.200');
$worker->addServer ('192.168.1.201');
$worker->addFunction ('generate_tile', 'generate_tile');
while (1)
{
if (!$worker->work ())
{
switch ($worker->returnCode ())
{
default:
echo "Error: " . $worker->returnCode () . ': ' . $worker->error () . "n";
break;
}
}
}
function generate_tile ($job) { ... }
工作代码正在2个不同的服务器上运行。当每个服务器都启动并运行时，两个worker按预期执行作业。当一个工作进程被取消时，另一个工作进程按预期执行所有作业。
但是，当取消工作进程的服务器关闭并完全脱机时，对客户端脚本的请求会挂起，剩余的工作进程不会拾取任何作业。
我从剩余的工作进程中得到以下错误集:
Error: 46: gearman_con_wait:timeout reached
Error: 46: gearman_con_wait:timeout reached
Error: 4: gearman_con_flush:write:110
Error: 46: gearman_con_wait:timeout reached
Error: 4: gearman_con_flush:write:113
Error: 4: gearman_con_flush:write:113
Error: 4: gearman_con_flush:write:113
....
当我启动另一个服务器时，没有启动它的工作进程，剩余的工作进程立即跳入生活并执行任何剩余的作业。
对我来说似乎很清楚，我需要在工作进程中的一些代码来处理任何可能离线的服务器，但是我看不出如何做到这一点。
许多谢谢,安迪我们对多个gearman服务器的测试表明，如果列表中的最后一个服务器(在您的示例中为192.168.1.201)被关闭，那么worker将停止按照您描述的方式执行。(此外，工人从最后一个服务器获取工作。只有在。201上没有作业时，它们才会处理。200上的作业)。
似乎这是gearman服务器中的链表的错误，据报道已经修复了多次，但在所有可用的gearman版本中，这个错误仍然存在。对不起，我知道这不是一个解决方案，但我们有同样的问题，没有找到解决方案。(如果有人能提供有效的解决这个问题的方法，我同意给予大量的赏金)
关于@Darhazer上面的评论。我们也发现了这一点，并像这样解决了:-
// Gearman workers show a strong preference for servers at the end of a list so randomize the order
$worker = new GearmanWorker();
$s2 = explode(",", Configure::read('workers.servers'));
shuffle($s2);
$servers = implode(",", $s2);
$worker->addServers($servers);
我们每次运行6到10个worker，并在它们完成x个请求后过期。
我使用这个类，它跟踪哪些作业在哪些服务器上运行。它还没有经过彻底的测试，只是现在写的。我已经粘贴了一个编辑过的版本，所以可能有拼写错误之类的，但其他方面似乎解决了问题。
<?
class MyGearmanClient {
static $server = "server1,server2,server3";
static $server_array = false;
static $workingServers = false;
static $gmclient = false;
static $timeout = 5000;
static $defaultTimeout = 5000;
static function randomServer() {
return self::$server_array[rand(0, count(self::$server_array) -1)];
}
static function getServer($job = false) {
if (self::$server_array == false) {
self::$server_array = explode(",", self::$server);
self::$workingServers = array();
}
$serverList = array();
if ($job) {
if (array_key_exists($job, self::$workingServers)) {
foreach (self::$server_array as $server) {
if (array_key_exists($server, self::$workingServers[$job])) {
if (self::$workingServers[$job][$server]) {
$serverList[] = $server;
}
} else {
$serverList[] = $server;
}
}
if (count($serverList) == 0) {
# All servers have failed, need to insert all the servers again and retry.
$serverList = self::$workingServers[$job] = self::$server_array;
}
return $serverList[rand(0, count($serverList) - 1)];
} else {
return self::randomServer();
}
} else {
return self::randomServer();
}
}
static function serverWorked($server, $job) {
self::$workingServers[$job][$server] = $server;
}
static function serverFailed($server, $job) {
self::$workingServers[$job][$server] = false;
}
static function Connect($server = false, $job = false) {
if ($server) {
self::$server = self::getServer();
}
self::$gmclient= new GearmanClient();
self::$gmclient->setTimeout(self::$timeout);
# add the default job server
self::$gmclient->addServer($server = self::getServer($job));
return $server;
}
static function Destroy() {
self::$gmclient = false;
}
static function Client($name, $vars, $timeout = false) {
if (is_int($timeout)) {
self::$timeout = $timeout;
} else {
self::$timeout = self::$defaultTimeout;
}
do {
$server = self::Connect(false, $name);
$value = self::$gmclient->do($name, $vars);
$return_code = self::$gmclient->returnCode();
if (!$value) {
$error_message = self::$gmclient->error();
if ($return_code == 47) {
self::serverFailed($server, $name);
if (count(self::$server_array) > 1) {
// ADDED SINGLE SERVER LOOP AVOIDANCE // echo "Timeout on server $server, trying another server...n";
continue;
} else {
return false;
}
}
echo "ERR: $error_message ($return_code)n";
}
# printf("Worker has returnedn");
$short_value = substr($value, 0, 80);
switch ($return_code)
{
case GEARMAN_WORK_DATA:
echo "DATA: $short_valuen";
break;
case GEARMAN_SUCCESS:
self::serverWorked($server, $name);
break;
case GEARMAN_WORK_STATUS:
list($numerator, $denominator)= self::$gmclient->doStatus();
echo "Status: $numerator/$denominatorn";
break;
case GEARMAN_TIMEOUT:
// self::Connect();
// Fall through
default:
echo "ERR: $error_message " . self::$gmclient->error() . " ($return_code)n";
break;
}
}
while($return_code != GEARMAN_SUCCESS);
$rv = unserialize($value);
return $rv["rv"];
}
}
# Example usage:
# $rv = MyGearmanClient::Client("Function", $args);
?>
由于来自gearman客户端的'addServer'不能正常工作，此代码可以随机选择一个jobserver，如果失败则尝试下一个，这样可以平衡负载。
// job servers
$jobservers = array('192.168.1.1','192.168.1.2');
// prepare gearman client
$gmclient = new GearmanClient();
// shuffle job servers (deliver jobs equally by server)
shuffle($jobservers);
// add job servers
foreach($jobservers as $jobserver) {
// add random jobserver
$gmclient->addServer($jobserver);
// check server state if ok end foreach
if (@$gmclient->ping('ping')) break;
// if connections fails reset client
$gmclient = new GearmanClient();
}
解决方案经过测试，运行正常
$client = new GearmanClient();
if(!$client->addServer("11.11.65.73",4730))
$client->addServer("11.11.65.79",4730);