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

Question

我在多个服务器上运行的 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
....

当我启动另一台服务器，而不是在其上启动工作进程时，剩余的工作进程立即开始运行并执行任何剩余的作业。

对我来说，很明显我需要在工作进程中使用一些代码来处理任何可能离线的服务器，但是我不知道如何做到这一点。

非常感谢，

安迪

Answer 1

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

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

Answer 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 个请求后使它们过期。

Answer 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);

?>

Answer 4

由于 gearman 客户端的“addServer”无法正常工作，此代码可以随机选择一个作业服务器，如果失败，请尝试下一个，这样您就可以平衡负载。

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

Answer 5

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

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