具有多个服务器和 php 工作人员的 Gearman答案

【问题标题】：Gearman with multiple servers and php workers具有多个服务器和 php 工作人员的 Gearman
【发布时间】：2011-08-16 09:35:12
【问题描述】：

我遇到了在多台服务器上运行的 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 个独立的服务器上运行。当每台服务器都启动并运行时，两个工作人员都按预期执行作业。当其中一个工作进程被取消时，另一个工作进程按预期执行所有作业。

但是，当带有已取消工作进程的服务器关闭并完全脱机时，对客户端脚本的请求会挂起，并且剩余的工作进程不会执行任何作业。

我从剩余的工作进程中收到以下一组错误：

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
....

当我启动另一台服务器时，没有启动其上的工作进程，剩余的工作进程会立即启动并执行任何剩余的作业。

我似乎很清楚，我需要工作进程中的一些代码来处理任何可能离线的服务器，但是我不知道该怎么做。

非常感谢，

安迪

【问题讨论】：

标签： php gearman

【解决方案1】：

我们对多个 gearman 服务器的测试表明，如果列表中的最后一个服务器（在您的情况下为 192.168.1.201）被关闭，工作人员将停止按照您描述的方式执行。（此外，工作人员从最后一个服务器获取作业。只有在 0.201 上没有作业时，他们才会在 .200 上处理作业）。

这似乎是gearman服务器中链表的一个错误，据报道已多次修复，但对于所有可用版本的gearman，该错误仍然存在。抱歉，我知道这不是解决方案，但我们遇到了同样的问题，但没有找到解决方案。（如果有人可以为这个问题提供可行的解决方案，我同意给予大赏）

【讨论】：

很有趣，谢谢。我已经更改了服务器的顺序，以便我要关闭的工作服务器是第一个服务器而不是最后一个服务器，尽管仍然会产生一些错误，但工作人员确实可以正确处理作业。我建议解决此问题的方法是在客户端服务器上运行一个工作人员并将其设置为最后一个服务器。这样，如果任何工作服务器出现故障，那么这不是问题，一切正常，因为它们都不是最后添加的服务器，但是如果工作/客户端服务器出现故障，那么客户端无论如何都会停机，所以没有新的工作可以进行处理。
有人有错误报告的链接吗？

【解决方案2】：

以上@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 个工作人员，并在他们完成 x 个请求后使他们过期。

【讨论】：

【解决方案3】：

我使用这个类，它跟踪哪些作业在哪些服务器上运行。还没有完全测试，现在才写。我粘贴了一个经过编辑的版本，因此可能存在拼写错误或类似问题，但似乎可以解决问题。

<?
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 returned\n");
                        $short_value = substr($value, 0, 80);
                        switch ($return_code)
                        {
                        case GEARMAN_WORK_DATA:
                                echo "DATA: $short_value\n";
                                break;
                        case GEARMAN_SUCCESS:
                                self::serverWorked($server, $name);
                                break;
                        case GEARMAN_WORK_STATUS:
                                list($numerator, $denominator)= self::$gmclient->doStatus();
                                echo "Status: $numerator/$denominator\n";
                                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);

?>

【讨论】：

这是一段方便的代码，谢谢 - 我看到它随机返回一个服务器，然后构建一个工作服务器数组以返回多个请求。对于工作脚本，您只需添加单个服务器？而不是像我一直在做的那样将所有服务器添加到工作服务器，因为您只是将单个工作服务器添加到客户端？
大多数情况下，我将所有工作人员添加到所有服务器。但我确实有一名工作人员需要返回快速结果，为此我使用专用的“本地”（每个网络一个）服务器。这提醒了我，如果您只指定一个服务器，我必须纠正一个导致无限重试循环的错误 - 这对我来说只是一个问题，因为如果我不执行本地（阅读：“exec（）”）处理1秒内得到答案。

【解决方案4】：

由于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();
        }

【讨论】：

在堆栈溢出时通常会感谢一些解释。

【解决方案5】：

解决方案经过测试，工作正常。

     $client = new GearmanClient();
     if(!$client->addServer("11.11.65.73",4730))
        $client->addServer("11.11.65.79",4730);

【讨论】：