您当前的位置： 首页 > 慢生活 > 程序人生 网站首页程序人生

workerman异步消息队列组件-redis-queue - 在非workerman环境向队列发送消息

发布时间：2021-12-30 23:00:41编辑：雪饮阅读（）

 在非workerman环境向队列发送消息

有时候一些项目运行在apache或者php-fpm环境，无法使用workerman/redis-queue项目，可以参考如下函数实现发送
订阅者：
<?php
use Workerman\Worker;
use Workerman\Timer;
use Workerman\RedisQueue\Client;
require_once __DIR__ . '/vendor/autoload.php';
$client_client_1=0;
$client_client_2=0;
$worker = new Worker();
$worker->onWorkerStart = function () {
    $client = new Client('redis://127.0.0.1:6379');
    // 订阅 假定逻辑为客户端1只接受一次订阅的消息
    $client->subscribe('user-1', function($data) use($client){
        global $client_client_1;
        if($client_client_1>0){
            //客户端1取消订阅
            $client->unsubscribe('user-1');
        }
        else{
            echo "user-1\n";
            var_export($data);
            $client_client_1++;
        }
 
    });
    // 订阅 假定逻辑为客户端2只接受两次订阅的消息
    $client->subscribe('user-2', function($data) use($client){
        global $client_client_2;
        if($client_client_2>1){
            //客户端2取消订阅
            $client->unsubscribe('user-2');
        }
        else{
            echo "user-2\n";
            var_export($data);
            $client_client_2++;
        }
 
    });
};
 
Worker::runAll();
订阅者运行并有向订阅者们所订阅的channel发送消息时：
[root@localhost workerman]# php redisQueue.php start
Workerman[redisQueue.php] start in DEBUG mode
------------------------------------- WORKERMAN --------------------------------------
Workerman version:4.0.26          PHP version:7.3.31
-------------------------------------- WORKERS ---------------------------------------
proto   user            worker          listen          processes    status
tcp     root            none            none            1             [OK]
--------------------------------------------------------------------------------------
Press Ctrl+C to stop. Start success.
user-1
array (
  0 => 'some1',
  1 => 'data1',
)user-2
array (
  0 => 'some1',
  1 => 'data1',
)
 
较为原生点的向队列发送消息的实现：
<?php
function redis_queue_send($redis, $queue, $data, $delay = 0) {
    //这里的版本是指workerman/redis-queue的版本
    $queue_waiting = '{redis-queue}-waiting'; //1.0.5版本之前为redis-queue-waiting
    $queue_delay = '{redis-queue}-delayed';//1.0.5版本之前为redis-queue-delayed
 
    $now = time();
    $package_str = json_encode([
        'id'       => rand(),
        'time'     => $now,
        'delay'    => 0,
        'attempts' => 0,
        'queue'    => $queue,
        'data'     => $data
    ]);
    if ($delay) {
        return $redis->zAdd($queue_delay, $now + $delay, $package_str);
    }
    return $redis->lPush($queue_waiting.$queue, $package_str);
}
 
$redis = new Redis;
$redis->connect('127.0.0.1', 6379);
$queue = 'user-1';
$data= ['some1', 'data1'];
redis_queue_send($redis, $queue, $data);
 
$queue = 'user-2';
$data= ['some1', 'data1'];
redis_queue_send($redis, $queue, $data);
 
较为原生点的向队列发送消息的实现的运行一般是没有什么输出的：
[root@localhost workerman]# php send.php