swoole 消息队列

Posted 2020-11-24 cshaptx4869

 

<?php
/**
 * 场景：
 * 监控订单表状态 队列通信
 * 一个进程向队列发布消息 另外两个进程争抢
 */

//设置主进程名
echo ‘主进程id:‘ . posix_getpid() . PHP_EOL;
cli_set_process_title(‘php_main‘);

//1、此子进程用于监听数据的改变
$process1 = new SwooleProcess(function (SwooleProcess $process) {
//    cli_set_process_title(‘php_child‘);
    $process->name(‘php_child1‘);
    $pdo = new PDO(‘mysql:host=mysql;dbname=test‘, ‘root‘, ‘csh.aptx4869#‘);

    while (true) {
        $statement = $pdo->query(‘select * from `order` where is_pay=1 and is_notice=0 order by id desc‘);
        $data = $statement->fetch(PDO::FETCH_ASSOC);
        if ($data) {
            //投递数据到消息队列中
            $process->push(‘php_child1发送‘ . $data[‘client_name‘] . ‘已支付 ‘);
            $pdo->exec(‘update `order` set is_notice=1 where id=‘ . $data[‘id‘]);
        }
        sleep(3);
    }
}, false, SOCK_STREAM, true);
//启动消息队列作为进程间通信
$process1->useQueue(2);
echo ‘子进程1 id:‘ . $process1->start() . PHP_EOL;

//2、此子进程用于发送邮件
$process2 = new SwooleProcess(function (SwooleProcess $process) {
    $process->name(‘php_child2‘);
    while (true) {
        //从队列中提取数据
        $orderInfo = $process->pop();
        if ($orderInfo) {
            echo $orderInfo;
            echo ‘php_child2发送邮件‘ . PHP_EOL;
        }
        sleep(3);
    }
}, false, SOCK_STREAM, true);
$process2->useQueue(2);
echo ‘子进程2 id:‘ . $process2->start() . PHP_EOL;

//3、此子进程用于发送邮件
$process3 = new SwooleProcess(function (SwooleProcess $process) {
    $process->name(‘php_child3‘);
    while (true) {
        //从队列中提取数据
        $orderInfo = $process->pop();
        if ($orderInfo) {
            echo $orderInfo;
            echo ‘php_child3发送邮件‘ . PHP_EOL;
        }
        sleep(3);
    }
}, false, SOCK_STREAM, true);
$process3->useQueue(2);
echo ‘子进程3 id:‘ . $process3->start() . PHP_EOL;

while (true) {
    sleep(3);
}

//SwooleProcess::wait();
//使用Process作为监控父进程，创建管理子进程时，父类必须注册信号SIGCHLD对退出的进程执行wait，否则子进程退出时会变成僵尸进程
SwooleProcess::signal(SIGCHLD, function ($signo) {
    //false 非阻塞模式
    while ($ret = SwooleProcess::wait(false)) {
        var_dump($ret);
    }
});