swoole如何做消息通知

function run()
{
 try {
 $swoole = new swoole_server(127.0.0.1, 9999);
 $swoole->set([
 'daemonize' => 1, //是否开启守护进程
 'worker_num' => 8, //实际需要去设定
 'log_file' => __APP_LOGS_PATH__ . '/swoole.log'
 ]);
 $swoole->on('WorkerStart', 'onWorkerStart');
 $swoole->on('Receive', 'onReceive');
 $swoole->start();
 } catch (Exception $e) {
 logs(['err_code' => $e->getCode(), 'err_msg' => $e->getMessage()], 'error');
 }
}

function onWorkerStart(swoole_server $swoole, $worker_id)
{
 $chQuick = [0, 1, 2, 3];
 $chNormal = [4, 5];
 $chSlow = [6];
 for ($i = 1; $i <= 3000; $i++) {
 $redis = connectRedis();//断线重连redis
 $queueData = $keys = [];
 if (in_array($worker_id, $chQuick)) {
 if ($redis->llen(QUEUE_QUICK))
 $keys[] = QUEUE_QUICK;
 if ($keys)
 $queueData = $redis->brpop(QUEUE_QUICK, 5);
 } elseif (in_array($worker_id, $chNormal)) {
 if ($redis->llen(QUEUE_NORMAL))
 $keys[] = QUEUE_NORMAL;
 if ($redis->llen(QUEUE_QUICK))
 $keys[] = QUEUE_QUICK;
 if ($keys)
 $queueData = $redis->brpop(QUEUE_NORMAL, QUEUE_QUICK, 5);
 } elseif (in_array($worker_id, $chSlow)) {
 if ($redis->llen(QUEUE_SLOW))
 $keys[] = QUEUE_SLOW;
 if ($redis->llen(QUEUE_NORMAL))
 $keys[] = QUEUE_NORMAL;
 if ($redis->llen(QUEUE_QUICK))
 $keys[] = QUEUE_QUICK;
 if ($keys)
 $queueData = $redis->brpop(QUEUE_SLOW, QUEUE_QUICK, QUEUE_NORMAL, 5);
 } else {
 if ($redis->llen(QUEUE_FAIL))
 $keys[] = QUEUE_FAIL;
 if ($redis->llen(QUEUE_SLOW))
 $keys[] = QUEUE_SLOW;
 if ($redis->llen(QUEUE_NORMAL))
 $keys[] = QUEUE_NORMAL;
 if ($redis->llen(QUEUE_QUICK))
 $keys[] = QUEUE_QUICK;
 if ($keys)
 $queueData = $redis->brpop(QUEUE_FAIL, QUEUE_QUICK, QUEUE_NORMAL, QUEUE_SLOW, 5);
 }
 logs('test'.$keys.'%%'.$queueData);
 if ($queueData) {
 $queueName = $queueData[0];
 $message = $queueData[1];
 if ($worker_id == QUEUE_FAIL_WORKER_ID && $queueName == QUEUE_FAIL) {
 call_user_func_array('retryPostMessage', [&$message, &$redis]);
 } else {
 call_user_func_array('postMessage', [&$message, &$redis]);
 }
 }
 else
 {
 sleep(5);
 }
 }
sleep(10);
 $redis->close();
 unset($redis);
 method_exists($swoole, 'stop') ? $swoole->stop() : @exit;
}