問(wèn)題描述

在gateway中使用AsyncTcpConnection把任務(wù)發(fā)給task_worker去處理，參考了異步任務(wù)文檔http://m.wtbis.cn/doc/workerman/faq/async-task.html
發(fā)現(xiàn)task_worker的onMessage一直無(wú)法觸發(fā)，如果在send前加上connect就能觸發(fā)?？吹揭灿腥颂釂?wèn)過(guò)類似問(wèn)題但是沒有解答

public static function onWorkerStart($businessWorker)
{
    if ($businessWorker->id == 0) {
        Timer::add($msgInterval, function () {
        // 與遠(yuǎn)程task服務(wù)建立異步連接，ip為遠(yuǎn)程task服務(wù)的ip，如果是本機(jī)就是127.0.0.1，如果是集群就是lvs的ip
        $task_connection = new AsyncTcpConnection('Text://127.0.0.1:12345');
        //                       $task_connection->connect();
        // 發(fā)送數(shù)據(jù)
        $task_connection->send(json_encode($taskData));
        // 異步獲得結(jié)果
        $task_connection->onMessage = function(AsyncTcpConnection $task_connection, $task_result) {
            // 執(zhí)行結(jié)果
                var_dump($task_result);
            // 獲得結(jié)果后記得關(guān)閉異步連接
                $task_connection->close();
            };
        });
    }
}

// task worker，使用Text協(xié)議
$task_worker = new Worker('Text://0.0.0.0:12345');
// task進(jìn)程數(shù)可以根據(jù)需要多開一些
$task_worker->count = 1;
// task進(jìn)程名稱
$task_worker->name = 'TaskWorker';
$task_worker->onWorkerStart = function(Worker $worker)
{
    echo "Worker starting...\n";
};
$task_worker->onConnect = function($connection)
{
    echo "new connection from ip " . $connection->getRemoteIp() . "\n";
};
// 設(shè)置業(yè)務(wù)處理函數(shù)
$task_worker->onMessage = function($connection, $task_data)
{
    // 假設(shè)發(fā)來(lái)的是json數(shù)據(jù)
    $task_data = json_decode($task_data, true);
    trace([$task_data,'msg'=>'發(fā)送后'],'crontab');
    // 根據(jù)task_data處理相應(yīng)的任務(wù)邏輯.... 得到結(jié)果，這里省略....
    $task_result = 'done';
    // 發(fā)送結(jié)果
    $connection->send(json_encode($task_result));
};