qimall/docs/guide_zh_cn/service/serv-rabbitmq.md

Mq服务

获取一个rabbitMQ连接

/**
 * 获取一个rabbitMQ连接
 * @Author bing
 * @DateTime 2021-02-24 10:35:37
 * @copyright: Copyright (c) 2020 广东七件事集团
 * @return AMQPStreamConnection
 * @throws Exception
 */
 $res = Yii::$app->services->rabbitMq->connect();

生产一个消息并发送到指定（direct）交换机

/**
 * 生产一个消息并发送到指定（direct）交换机
 * @Author bing
 * @DateTime 2021-02-24 15:28:47
 * @copyright: Copyright (c) 2020 广东七件事集团
 * @param string $exchange
 * @param string $queue_name  队列名称
 * @param array $data
 * @param string $handler_class
 * @param string $method
 * @param Closure $error_callback
 * @param Closure $success_callback
 * @param string $exchange_type   交换机类型：默认直连交换机
 * @throws Exception
 * @return int
 */
 $res = Yii::$app->services->rabbitMq->push($exchange,$queue_name,$data,$handler_class,$method, $error_callback, $success_callback,$exchange_type)

延时队列生产

/**
 * 延时队列生产
 * @notice:此延时队列不能保证时间的精准，当业务处理出现阻塞，则在队列里已达到过期时间的消息并不会被发送到对应的队列。
 * @Author bing
 * @DateTime 2021-02-24 16:49:31
 * @copyright: Copyright (c) 2020 广东七件事集团
 * @param int $sec 延时秒数
 * @param array $data 传递的数据
 * @param string $handler_class 处理类
 * @param string $method 处理方法
 * @param array $params 处理类参数
 * @throws Exception
 * @return void
 */
 $res = Yii::$app->services->rabbitMq->delay($sec, $data, $handler_class, $method, $params);

消费监听

/**
 * 消费者监听
 * @Author bing
 * @DateTime 2021-02-24 15:18:08
 * @copyright: Copyright (c) 2020 广东七件事集团
 * @param string  $exchange
 * @param string  $queue_name
 * @param Closure $callback
 * @param string $exchange_type   交换机类型：默认直连交换机
 * @throws Exception
 * @return void
 */
 $res = Yii::$app->services->rabbitMq->listen($exchange,$queue_name, $callback,$exchange_type); 

取出一个队列里的消息

/**
 * 取出一个队列里的消息
 * @Author bing
 * @DateTime 2021-02-27 17:19:25
 * @copyright: Copyright (c) 2020 广东七件事集团
 * @param string $queue_name 队列名称
 * @return AMQPMessage|null
 * @throws
 */
 $res = Yii::$app->services->rabbitMq->pull($queue_name); 

消息添加延时重试

/**
 * 消息添加延时重试
 * @desc 消息消费失败，加入重试队列
 * @param object  $message
 * @param string $exchange_type   交换机类型：默认直连交换机
 * @throws Exception
 * @return void
 */
 $res = Yii::$app->services->rabbitMq->retry($message,$exchange_type); 

生成MQ消息全局唯一ID

/**
 * 生成MQ消息全局唯一ID
 * @return mixed
 */
 $res = Yii::$app->services->rabbitMq->createMessageId(); 

redis设置消息信息

/**
 * redis设置消息信息
 * @param $message_id 消息ID
 * @param $data
 * @return bool
 * @throws
 */
 $res = Yii::$app->services->rabbitMq->setMessage($message_id, $data);

判断消息是否存在

/**
 * 判断消息是否存在（不存在：已消费）
 * @param $message_id 消息ID
 * @return mixed
 * @throws
 */
 $res = Yii::$app->services->rabbitMq->checkMessage($message_id);

消息消费完删除

/**
 * 消息消费完删除
 * @param $message_id 消息ID
 * @return mixed
 */
 $res = Yii::$app->services->rabbitMq->deleteMessage($message_id);