不知道你現(xiàn)在有多少個(gè)

什么類型的任務(wù)?里面會(huì)不會(huì)有長(zhǎng)時(shí)間阻塞的,如果沒(méi)有的話,只是類似于鬧鐘的話,完全可以用一個(gè)定時(shí)器來(lái)解決,不需要每個(gè)任務(wù)一個(gè)定時(shí)器.

不用掃表,在onWorkerStart 一次性獲取全部,然后用自增id作為數(shù)組鍵保存,然后開(kāi)啟個(gè)text 協(xié)議監(jiān)聽(tīng),類似于這樣

   /**
     * @param TcpConnection $connection
     * @param $buffer
     * @return void
     * 開(kāi)啟text協(xié)議監(jiān)聽(tīng)
     */
    public function onMessage(TcpConnection $connection, $buffer)
    {
        $data = json_decode($buffer, true);

        $id = $data['id'] ?? '';
        $item = $data['item'] ?? [];
        $route_type = $data['route_type'] ?? '';

        if ($route_type === 'addTask'){
            $this->addTask($id,$item);
        }else if ($route_type === 'removeTask'){
            $this->removeTask($id);
        }
        $connection->send(true ? 'ok' : 'fail');
        $connection->close();
    }

    /**
     * @param $id int 數(shù)據(jù)庫(kù)自增id
     * @param $item array 任務(wù)內(nèi)容
     * @return void
     * 添加任務(wù)
     */
    private function addTask($id,$item){
        $this->task_list[$id] = $item;
    }

    /**
     * @param $id int 數(shù)據(jù)庫(kù)自增id
     * @return void
     * 刪除任務(wù)
     */
    private function removeTask($id){
        if (isset($this->task_list[$id])){
            unset($this->task_list[$id]);
        }
    }

然后添加任務(wù),或者刪除任務(wù)的時(shí)候,用這個(gè)連接text端口添加或者刪除任務(wù)就行了.

/**
 * @param array $data
 * @return bool
 * 發(fā)送text協(xié)議
 */
function sendTextSocket( $data = [])
{
    $address = '127.0.0.1:5678';
    // 建立socket連接到內(nèi)部推送端口
    $client = stream_socket_client('tcp://' . $address , $errno, $errmsg, 1);

    // 發(fā)送數(shù)據(jù)，注意5678端口是Text協(xié)議的端口，Text協(xié)議需要在數(shù)據(jù)末尾加上換行符
    fwrite($client, json_encode($data) . "\n");
    // 讀取推送結(jié)果
    $res = fread($client, 8192);
}