分享一个基于workerman的rabbitmq客户端生产者、消费者（基于Timer实现事件驱动）

zgh419566

通过综合分析、研究和探索workerman和PhpAmqpLibr相关手册，经过长期的实践，现分享一套基于workerman的rabbitmq客户端生产者和消费者代码，供大家测试，使用。

个人觉得这套代码比workerman官方的代码逻辑更清晰，更便于使用

希望walkor采用，这样的话我也算为开源社区做了一些贡献。

rabbitmq 生产者

rabbitmq_productor.php

<?php
require_once ('./vendor/autoload.php');
require_once ("./Lib_global.php");

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use Workerman\Worker;
use Workerman\Lib\Timer;
use Workerman\Connection\TcpConnection;
use Workerman\Connection\AsyncUdpConnection;
use Workerman\Connection\AsyncTcpConnection;

$worker = new Worker();
//开启进程数量
$worker->count = 4;
$worker->name = "rabbitmq_productor";

$date = date("Y-m-d");
Worker::$pidFile = "var/mq_service_productor.pid";
Worker::$logFile = "var/mq_service_productor_logFile.log";
Worker::$stdoutFile = "var/mq_service_productor_stdout.log";

$worker->onWorkerStart = function () {
    global $rabbit_connection, $rabbit_channel ,  $rabbitmq_exchange_name , $rabbitmq_queueName;

    $rabbitmq_exchange_name = "exchange_name";
    $rabbitmq_queueName = "queuePrefix_QueueName";

    // 连接 rabbitmq 服务
    $rabbit_connection = new AMQPStreamConnection(RABBITMQ_SERVER_IP, RABBITMQ_SERVER_PORT, RABBITMQ_USERNAME, RABBITMQ_PASSWORD);

    // 获取信道
    $rabbit_channel = $rabbit_connection->channel();

    //声明创建交换机
    $rabbit_channel->exchange_declare( $rabbitmq_exchange_name , 'topic', false, true, false);

    // 声明创建队列
    $rabbit_channel->queue_declare( $rabbitmq_queueName , false, true, false, false);

    // 绑定队列
    $rabbit_channel->queue_bind($rabbitmq_queueName , $rabbitmq_exchange_name, $rabbitmq_queueName);

    //可以修改时间间隔，如果为0.002秒，则每秒产生500*4=2000条
    Timer::add( 0.002 , function() {
        global $rabbit_connection, $rabbit_channel , $rabbitmq_exchange_name , $rabbitmq_queueName;

        //需要向rabbitmq队列投递消息的内容，通常为数组，经过json转换再发送
        $data_all = array(
            'name' => "张三",
            'time' => time(),
        );
        $data_all_out_json = json_encode($data_all , JSON_UNESCAPED_UNICODE );
        $data_all_out_msg = new AMQPMessage($data_all_out_json, ['content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]);

        //向队列里面写内容
        @$rabbit_channel->basic_publish($data_all_out_msg , $rabbitmq_exchange_name , $rabbitmq_queueName);
    });
};

Worker::runAll();

rabbitmq 消费者

rabbitmq_comsumer.php

<?php

require_once ('./vendor/autoload.php');
require_once ("./Lib_global.php");

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use Workerman\Worker;
use Workerman\Lib\Timer;
use Workerman\Connection\TcpConnection;
use Workerman\Connection\AsyncUdpConnection;
use Workerman\Connection\AsyncTcpConnection;

$worker = new Worker();
//开启进程数量
$worker->count = 10;
$worker->name = "rabbitmq_comsumer";

$date = date("Y-m-d");
Worker::$pidFile = "var/rabbitmq_comsumer.pid";
Worker::$logFile = "var/rabbitmq_comsumer_logFile.log";
Worker::$stdoutFile = "var/rabbitmq_comsumer_stdout.log";

$worker->onWorkerStart = function () {
    global $rabbit_connection, $rabbit_channel , $rabbitmq_exchange_name , $rabbitmq_queueName;

    $rabbitmq_exchange_name = "exchange_name";
    $rabbitmq_queueName = "queuePrefix_QueueName";

    // 连接 rabbitmq 服务
    $rabbit_connection = new AMQPStreamConnection(RABBITMQ_SERVER_IP, RABBITMQ_SERVER_PORT, RABBITMQ_USERNAME, RABBITMQ_PASSWORD);

    // 获取信道
    $rabbit_channel = $rabbit_connection->channel();

    // 声明队列
    $rabbit_channel->queue_declare( $rabbitmq_queueName , false, true, false, false);

    // 绑定队列
    $rabbit_channel->queue_bind($rabbitmq_queueName , $rabbitmq_exchange_name, $rabbitmq_queueName);

    // 消费者订阅队列
    $rabbit_channel->basic_consume($rabbitmq_queueName , '', false, false, false, false,
        function ($msg){
            global $rabbit_channel , $rabbitmq_exchange_name , $rabbitmq_queueName;
            $data_all_str = $msg->body;
            // 消息确认，表明已经收到这条信息
            @$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
            //echo "{$data_all_str}\n";

            //这里是业务处理逻辑
            //如果这条消息处理失败，你可以在这里将其再次放回消息队列（最好给消息做个放回去的次数判断，避免无限失败和无限循环）
        });

    //这里是重点，网上很多教程起的使用while死循环容易导致程序异步代码无法执行，这种方法能避免
    //按照每个进程每秒处理1000条来设定定时器，每个进程每秒消费1000条，4个进程每秒消费4000条，经过实际验证，将时间改小也无法提升单个进程的处理速度
    //实际测试，4个进程每秒的消费能力有4000左右，可以满足很多中小型系统的应用，如果想提升系统处理能力，
    //可以增加消费者进程数量来解决，比如我将进程数量提升到10个，每秒处理能力约为1万
    //这个机制，希望能力更强的你来进行优化
    Timer::add( 0.0001 , function() {
        global $rabbit_channel;
        if( count($rabbit_channel->callbacks) > 0 ){
            $rabbit_channel->wait();
        }
    });
};

Worker::runAll();

附上我的代码运行情况

3537 6 11
6个回答
walkor 打赏

感谢分享

  • doit 2022-12-06

    大佬，有没有准备集成到webman框架中？
  • chaz6chez 2022-12-06

    webman插件有rabbitmq的消费者插件：
    https://www.workerman.net/plugin/67
  • chaz6chez 2022-12-06

    插件包含生产者、消费者；同步、异步；延迟队列、普通队列；
  • doit 2022-12-06

    看见了的，在研究的，已在生产环境使用了？
  • chaz6chez 2022-12-06

    用了很久了
  • doit 2022-12-06

    @chaz6chez，安装测试了一下，发现每创建一个队列，默认创建了一个交换机，请教一下同一类型为何不使用同一个交换机呢？对于已经创建好的队列，如何指定交换机和队列名称呢。还有就是消息消费失败重试如何操作比较好呢？
  • chaz6chez 2022-12-06

    @doit FastBuilder是一个实现点对点消费模式的消费队列Builder，在这个模型里就是“N个生产者对应一个业务对应一个交换机对应一个队列对应N个消费者”；简单讲就是一个业务对应一条队列。这样的好处是所有业务是隔离的，业务对于rabbitMQ是透明的，它仅仅只做队列的分发而已。
    FastBuilder默认以ClassName为名，也有默认的消费失败重试方式，通常来说有两种重试机制，队头阻塞重试，回到队尾重试；
    如果要实现其他消费模式的话，可以继承Builder，自行实现即可
evilk

666,感谢分享,赞一个,有机会试用一下
另外,对于其他方面,说一下个人愚见(跟楼主这个没有太大关系)
很多时候,当遇到的确需要使用MQ的情况,中小型公司,真的不会花额外的成本去用专业的MQ,最多用redis
我知道,可能很多人会说,这样很操蛋,但事实上,有很多公司,就是这样,成本能低就低
但redis目前有一个很大的问题,目前的webman队列,是使用的普通的list结构,这个是没有ack机制的
这对于某些对消息可靠性有比较高的要求的情况,就很尴尬
我们目前使用webman的redis队列插件,都是用在对消息可靠性要求不是特别高的场景,比如通知等
我有个想法,基于redis stream 写一套类似的消息队列插件
这样就可以真正满足实际需求了
有时间一定写一个出来,相信这个可能会更加实用

  • 暂无评论
monty

大佬，require_once ("./Lib_global.php");这个是什么文件，用到了吗，能发一下吗？

  • zgh419566 2022-10-25

    一些预定义的变量，设置PHP运行环境等

    <?php
    define('START_TIME', microtime(true));
    define('START_MEM', memory_get_usage());
    define('DS', DIRECTORY_SEPARATOR);

    //defined('ROOT_PATH') or define('ROOT_PATH', dirname(realpath(APP_PATH)) . DS);
    defined('ROOT_PATH') or define('ROOT_PATH', dirname($_SERVER['SCRIPT_FILENAME']) . DS);
    defined('LOG_PATH') or define('LOG_PATH', ROOT_PATH . 'Log' . DS);
    // 环境常量
    define('IS_CLI', PHP_SAPI == 'cli' ? true : false);
    define('IS_WIN', strpos(PHP_OS, 'WIN') !== false);

    $config = array();
    $global = array();

    // do NOT run this script through a web browser
    if (!isset($_SERVER['argv'][0]) || isset($_SERVER['REQUEST_METHOD']) || isset($_SERVER['REMOTE_ADDR'])) {
    die('<br><strong>This script is only meant to run at the command line.</strong>');
    }

    / let PHP run just as long as it has to /
    ini_set('max_execution_time', '0');

    //这个必须有
    ini_set('memory_limit','320M');

    //error_reporting('E_ALL');

    / this should be auto-detected, set it manually if needed /
    $config["server_os"] = (strstr(PHP_OS, "WIN")) ? "win32" : "unix";

    $config['root_path'] = ROOT_PATH;

    $config['logPath'] = LOG_PATH;

    chdir($config['root_path']);
  • monty 2022-10-25

    谢谢啦，测了一下没这些也能跑，哈哈
powerbowen

妙

  • 暂无评论
zgh419566

分享一个进一步优化的方法，不使用Timer驱动，使用Swoole协程方式

//全局里面加个：
Worker::$eventLoopClass = 'Workerman\Events\Swoole';    //将事件引擎替换为Swoole，前提是安装了Swoole，我安装的是4.8.12
Co::set(['hook_flags' => SWOOLE_HOOK_SLEEP]);   //将sleep一键协程化
Swoole\Coroutine::set(['enable_deadlock_check' => false]);    //取消时间死锁检查机制，避免报错

将原来的以下代码进行替换
/*
    //按照每个进程每秒处理1万条来设定定时器
    Timer::add( 0.001 , function() {
        global $rabbit_channel;
        if( count($rabbit_channel->callbacks) > 0 ){
            $rabbit_channel->wait();
        }
    });
*/

//替换为：
// 开始消费
while (count($rabbit_channel->callbacks)) {
        $rabbit_channel->wait();
        usleep(1000);
}
  • zgh419566 2022-12-21

    分享一下我写的rabbitmq客户端类库，支持定时与服务器握手，支持发生错误时进行生重连

    <?php
    /**

    • Created by PhpStorm.
    • User: Administrator
    • Date: 2019/6/13 0013
    • Time: 2:04
      */

    //composer require php-amqplib/php-amqplib

    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    use PhpAmqpLib\Wire\AMQPWriter;
    use PhpAmqpLib\Wire\AMQPReader;

    class rabbitmq_client {
    public $connection = "";
    public $channel = "";
    public $is_connected = false ;
    public $exchange_name = "" ;
    public $queen_name = "";
    public $comsume_callback = null;
    public $config_option = array(
    'host' => "127.0.0.1",
    'port' => "5672",
    'user' => "admin",
    'password' => "admin",
    'exchange_name' => 'default_exchange_name',
    'queen_name' => 'default_queen_name',
    );

    public function __construct($option = array()){
    $this->config_option['host'] = $option['host'];
    $this->config_option['port'] = $option['port'];
    $this->config_option['user'] = $option['user'];
    $this->config_option['password'] = $option['password'];
    $this->exchange_name = $option['exchange_name'];
    $this->queen_name = $option['queen_name'];
}

function app_log($log){
    //将日志信息发送给日志服务器
    $ts = round(microtime(true) - time() , 6);
    @list($ts1 , $ts2) = @explode("." , $ts);

    $logData = "{$ts1}.{$ts2} {$log}";

    //logToFile($logData);
    if(function_exists("logToScreen") == true){
        logToScreen($logData , true);
    }else{
        echo $logData."\n";
    }
}

public function connect(){
    try{
        $this->connection = new AMQPStreamConnection(
            $this->config_option['host'] ,
            $this->config_option['port'] ,
            $this->config_option['user'] ,
            $this->config_option['password'] ,
            '/' ,
            false ,
            'AMQPLAIN' ,
            null,
            'en_US' ,
            3.0 ,
            3.0 ,
            null ,
            true ,
            60
        );

        if( $this->connection ->isConnected() == true){

            $this->channel = $this->connection->channel();

            //声明交换机
            $this->channel->exchange_declare( $this->exchange_name , 'topic', false, true, false);

            // 声明队列
            $this->channel->queue_declare( $this->queen_name , false, true, false, false);

            // 绑定队列
            $this->channel->queue_bind($this->queen_name , $this->exchange_name , $this->queen_name );

            $this->is_connected = true;
            $this->app_log("rabbitmq connected");
            return true;
        }
    }catch (Exception $e) {
        $this->app_log("error catched :".$e->getMessage());
        $this->is_connected = false;
    }
    return false;
}

public function reconnect(){
    if( $this->is_connected == false ){
        if( $this->connect() == true ){
            //重新连接到服务器
            $this->is_connected = true;
            return true;
        }
    }
    return false;
}

/**
 * @return void
 *
 * 向服务器发送
 */
function write_heartbeat(){
    if($this->is_connected == true){
        try{
            //app_log("heartbeat");
            $pkt = new AMQPWriter();
            $pkt->write_octet(8);
            $pkt->write_short(0);
            $pkt->write_long(0);
            $pkt->write_octet(0xCE);
            $this->connection->write($pkt->getvalue());
        }catch (Exception $e) {
            $this->app_log("error catched :".$e->getMessage());
            $this->is_connected = false;
            $this->reconnect();
        }
    }else{
        // false
        if( $this->connect() == true ){
            //尝试连接到服务器
            $this->is_connected = true;
        }
    }
}

/**
 * @param $data
 * @param $queen_name
 * @param $is_persistent
 * @param $is_debug
 * @return void
 */
function publish( $data = "" , $is_persistent = true , $exchange_name_input = "" , $queen_name_input = ""){
    $delivery_mod = AMQPMessage::DELIVERY_MODE_PERSISTENT;
    if($is_persistent == false){
        $delivery_mod = AMQPMessage::DELIVERY_MODE_NON_PERSISTENT;
    }

    $exchange_name = "";
    if(strlen($exchange_name_input) > 0){
        $exchange_name = $exchange_name_input;
    }else{
        $exchange_name = $this->exchange_name;
    }

    $queen_name = "";
    if(strlen($queen_name_input) > 0){
        $queen_name = $queen_name_input;
    }else{
        $queen_name = $this->exchange_name;
    }

    $rabbit_msg = new AMQPMessage($data , ['content_type'=>'text/plain','delivery_mode'=>$delivery_mod]); //定义消息
    try{
        // 发送消息
        $this->channel->basic_publish($rabbit_msg, $exchange_name, $queen_name);
    }catch (Exception $e) {
        $this->app_log("error catched :".$e->getMessage());
        $this->is_connected = false;
        if( $this->reconnect() == true ){
            $this->is_connected = true;
        }
    }
}

//在做消费时，对流量进行控制，防止出现丢数据
function set_comsume_qos( $prefetch_size = 0 , $prefetch_count = 1 ){
    $this->channel->basic_qos( $prefetch_size , $prefetch_count ,false);   //当有消息在处理时不要发过来
}

/*
function comsume_callback($msg){
    //收到MQ消息
    $message_body = $msg->body;
    $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
    //echo "[x] Received ", $message_body, "\n";
    //redis_add_statistic( $redis , "rabbitmq:qos_test_consumption"  , 0.1);
}
*/

/*
Co::set(['hook_flags' => SWOOLE_HOOK_SLEEP]);
Swoole\Coroutine::set(['enable_deadlock_check' => false]);
 * */
function comsume_swoole_go( $queen_name_input = ""){
    $queen_name = "";
    if(strlen($queen_name_input) > 0){
        $queen_name = $queen_name_input;
    }else{
        $queen_name = $this->queen_name;
    }
    if($this->is_connected == true) {
        go(function () {
            // 消费者订阅队列
            try {
                if( !$this->comsume_callback ){
                    $this->app_log("function comsume_callback must be set");
                    return false;
                }
                $this->channel->basic_consume( $this->queen_name , '' , false , false , false , false , $this->comsume_callback);
            } catch (Exception $e) {
                $this->app_log("error catched :" . $e->getMessage());
                $this->is_connected = false;
                $this->reconnect();
            }

            // 开始消费
            try {
                /*
                while ( count($this->channel->callbacks) ) {
                    $this->channel->wait();
                    usleep(1);
                }*/
                while ( $this->channel->is_consuming() ) {
                    $this->channel->wait();
                    usleep(1);
                }
            } catch (Exception $e) {
                app_log("error catched when consuming:" . $e->getMessage());
                $this->is_connected = false;
                $this->reconnect();
            }
        });
    }else{
        // false
        $this->connect();
    }
}

function comsume( $queen_name_input = ""){
    $queen_name = "";
    if(strlen($queen_name_input) > 0){
        $queen_name = $queen_name_input;
    }else{
        $queen_name = $this->queen_name;
    }

    if($this->is_connected == true) {
        // 消费者订阅队列
        try {
            if( !$this->comsume_callback ){
                $this->app_log("function comsume_callback must be set");
                return false;
            }
            $this->channel->basic_consume( $this->queen_name , '' , false , false , false , false , $this->comsume_callback);
        } catch (Exception $e) {
            $this->app_log("error catched :" . $e->getMessage());
            $this->is_connected = false;
            $this->reconnect();
        }

        // 开始消费
        try {
            /*
            while ( count($this->channel->callbacks) ) {
                $this->channel->wait();
                usleep(1);
            }*/
            while ( $this->channel->is_consuming() ) {
                $this->channel->wait();
            }
        } catch (Exception $e) {
            app_log("error catched when consuming:" . $e->getMessage());
            $this->is_connected = false;
            $this->reconnect();
        }
    }else{
        // false
        $this->connect();
    }
}

    }

    生产者：

    chdir(dirname($_SERVER['SCRIPT_FILENAME']));

    include_once __DIR__ . '/vendor/autoload.php';
    include_once("./Lib_global.php");
    include_once("./Lib_functions_rabbitmq.php");

    use Workerman\Worker;
    use Workerman\Lib\Timer;
    use Workerman\Connection\AsyncTcpConnection;
    use Workerman\Connection\TcpConnection;
    use Swoole\Coroutine;

    $worker = new Worker();

    //开启进程数量
    $worker->count = 2;
    $processName = "test_mq_pub";
    $worker->name = $processName;
    $date_ymd = date("Y-m-d");
    Worker::$pidFile = ROOT_PATH."var/{$processName}.pid";
    Worker::$logFile = ROOT_PATH."var/{$processName}_logFile.log";
    Worker::$stdoutFile = ROOT_PATH."var/{$processName}_stdout.log";
    Worker::$eventLoopClass = 'Workerman\Events\Swoole';

    $redis = "";
    $is_debug = false; //全局配置，是否开启调试模式
    $rabbitmq_client = "";

    function app_log($log){
    global $workerId;
    //将日志信息发送给日志服务器
    $ts = round(microtime(true) - time() , 6);
    @list($ts1 , $ts2) = @explode("." , $ts);

    //logToFile(".{$ts2} [$workerId] ".$log);
logToScreen(".{$ts2} [$workerId] ".$log , true);

    }

    $worker->onWorkerStart = function() {
    global $redis, $worker, $workerId , $rabbitmq_client;

    Co::set(['hook_flags' => SWOOLE_HOOK_SLEEP]);
Swoole\Coroutine::set(['enable_deadlock_check' => false]);

//根据daemon顺序延时，这是确保系统正常运行的关键
usleep(1000 * 100 * ($worker->id + 1));
$workerId = $worker->id ;
echo date("Y-m-d H:i:s") . " 服务进程{$workerId}已经启动！\n";

//连接到Redis服务器
$redis = redis_connect(REDIS_SERVER , REDIS_PORT , REDIS_PASSWORD , REDIS_DBNAME);
//定期与redis握手，避免被断掉，该动作每个进程都得执行
Timer::add( 60 , function (){
    global $redis, $worker;
    $redis_status = $redis->ping();
    if($redis_status == false){
        $redis = redis_connect(REDIS_SERVER , REDIS_PORT , REDIS_PASSWORD , REDIS_DBNAME);
    }
});

$rabbitmq_config_option = array();
$rabbitmq_config_option['host'] = RABBITMQ_SERVER_IP;
$rabbitmq_config_option['port'] = RABBITMQ_SERVER_PORT;
$rabbitmq_config_option['user'] = RABBITMQ_USERNAME;
$rabbitmq_config_option['password'] = RABBITMQ_PASSWORD;
$rabbitmq_config_option['exchange_name'] = "4b_ads_CLASS_mqTest";
$rabbitmq_config_option['queen_name'] = "4b_ads_CLASS_mqTest";
$rabbitmq_client = new rabbitmq_client($rabbitmq_config_option);

//不断尝试与Rabbitmq服务器建立连接
while( $rabbitmq_client->is_connected == false ){
    if($rabbitmq_client->connect() == true){
        break;
    }
    $rabbitmq_client->app_log("rabbitmq server connect failed");
    sleep(1);
}
//执行定时握手任务
Timer::add( 55 , function ()use($rabbitmq_client) {
    // 发送心跳数据
    $rabbitmq_client->write_heartbeat();
});

Timer::add( 0.001 , function ()use($rabbitmq_client) {
    if($rabbitmq_client->is_connected == true){
        $data_json = GetRandStr(128);
        $rabbitmq_msg = json_encode($data_json);
        $rabbitmq_client->publish($rabbitmq_msg);
    }
});

    };

    Worker::runAll();

    消费者：

    chdir(dirname($_SERVER['SCRIPT_FILENAME']));

    include_once __DIR__ . '/vendor/autoload.php';
    include_once("./Lib_global.php");
    include_once("./Lib_functions_rabbitmq.php");

    use Workerman\Worker;
    use Workerman\Lib\Timer;
    use Workerman\Connection\AsyncTcpConnection;
    use Workerman\Connection\TcpConnection;
    use Swoole\Coroutine;

    $worker = new Worker();

    //开启进程数量
    $worker->count = 2;
    $processName = "test_mq_sub";
    $worker->name = $processName;
    $date_ymd = date("Y-m-d");
    Worker::$pidFile = ROOT_PATH . "var/{$processName}.pid";
    Worker::$logFile = ROOT_PATH . "var/{$processName}_logFile.log";
    Worker::$stdoutFile = ROOT_PATH . "var/{$processName}_stdout.log";
    Worker::$eventLoopClass = 'Workerman\Events\Swoole';

    $redis = "";
    $is_debug = false; //全局配置，是否开启调试模式
    $rabbitmq_client = "";

    function app_log($log)
    {
    global $workerId;
    //将日志信息发送给日志服务器
    $ts = round(microtime(true) - time(), 6);
    @list($ts1, $ts2) = @explode(".", $ts);

    //logToFile(".{$ts2} [$workerId] ".$log);
logToScreen(".{$ts2} [$workerId] " . $log, true);

    }

    $worker->onWorkerStart = function () {
    global $redis, $worker, $workerId, $rabbitmq_client;

    Co::set(['hook_flags' => SWOOLE_HOOK_SLEEP]);
Swoole\Coroutine::set(['enable_deadlock_check' => false]);

//根据daemon顺序延时，这是确保系统正常运行的关键
usleep(1000 * 100 * ($worker->id + 1));
echo date("Y-m-d H:i:s") . " 服务进程{$worker->id}已经启动！\n";
$workerId = $worker->id ;

//连接到Redis服务器
$redis = redis_connect(REDIS_SERVER , REDIS_PORT , REDIS_PASSWORD , REDIS_DBNAME);
//定期与redis握手，避免被断掉，该动作每个进程都得执行
Timer::add( 60 , function (){
    global $redis, $worker;
    $redis_status = $redis->ping();
    if($redis_status == false){
        $redis = redis_connect(REDIS_SERVER , REDIS_PORT , REDIS_PASSWORD , REDIS_DBNAME);
    }
});

//初始化Rabbitmq连接
$rabbitmq_config_option = array();
$rabbitmq_config_option['host'] = RABBITMQ_SERVER_IP;
$rabbitmq_config_option['port'] = RABBITMQ_SERVER_PORT;
$rabbitmq_config_option['user'] = RABBITMQ_USERNAME;
$rabbitmq_config_option['password'] = RABBITMQ_PASSWORD;
$rabbitmq_config_option['exchange_name'] = "4b_ads_CLASS_mqTest";
$rabbitmq_config_option['queen_name'] = "4b_ads_CLASS_mqTest";
$rabbitmq_client = new rabbitmq_client($rabbitmq_config_option);

//不断尝试与Rabbitmq服务器建立连接
while( $rabbitmq_client->is_connected == false ){
    if($rabbitmq_client->connect() == true){
        break;
    }
    $rabbitmq_client->app_log("rabbitmq server connect failed");
    sleep(1);
}
//执行定时握手任务
Timer::add( 55 , function ()use($rabbitmq_client) {
    // 发送心跳数据
    $rabbitmq_client->write_heartbeat();
});

$rabbitmq_client->comsume_callback = function ($msg){
    //收到MQ消息
    $message_body = $msg->body;
    $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
    //echo "[x] Received ", $message_body, "\n";
};

//$rabbitmq_client->comsume_swoole_go();   //采用协程方式处理
$rabbitmq_client->comsume();               //传统方式

    };

    Worker::runAll();
zgh419566

在使用时发现或多或少有一些问题（最大的问题是CPU抢占问题，导致workerman内的基于定时任务长时间得不到执行），毕竟官方的内容都是同步机制的，我在想有没有可能使用异步实现。
经过长时间的研究，终于解决了这个问题，个人认为比官方基于bunny+React的方式更好使用一些。

https://www.workerman.net/a/1485

  • 暂无评论
相似问题
分享一个基于workerman的rabbitmq客户端生产者、消费者（基于Event实现事件驱动）
基于AsyncTcpConnection实现的WS客户端reConnect 方法问题。
分享: 使用workerman实现基于UDP的异步SIP服务器，服务器端可主动发送UDP数据给客户端
FlowerMQ 基于workerman和redis实现的消息队列
我的第一个Webman插件 - 基于Redis Stream实现的消息队列插件
rabbitmq有没有办法一个消费者消费多个队列啊
年代过于久远，无法发表回答
月贡献榜
1
walkor
🚀
2
nitron
3
tanhongbin
夕阳无限好，只是近黄昏
4
邹意
5
six
赞助商
在线客服系统源码
聊天源码|聊天室源码
+成为赞助商

app,小程序,H5聊天源码

语音|视频|文字聊天

客服系统源码

小程序客服|h5客服|网页客服
×
🔝