Thinkphp6使用RabbitMQ消息队列

Thinkphp6连接使用RabbitMQ（不止tp6，其他框架对应改下也一样），如何使用Docker部署RabbitMQ，在上一篇已经讲了->传送门<-。

部署环境

开始前先进入RabbitMQ的web管理界面，选择Queues菜单，点击底部的Add a new queue，新建一个test的队列。

安装thinkphp6框架

composer create-project topthink/think tp 

安装workerman扩展

composer require topthink/think-worker

安装rabbitmq扩展

composer require workerman/rabbitmq

代码编写

生产者

• 在app目录下新建workerman目录，并在其下创建Send.php文件，$options数组中的host地址改成你的rabbitmq地址。

<?phpnamespace app\workerman;
use Bunny\Channel;
use Workerman\RabbitMQ\Client;
use think\worker\Server;
class Send extends Server
{//websocket地址，一会用于测试。protected $socket = 'websocket://127.0.0.1:2345';/*** 收到信息* @param $connection* @param $data*/public function onMessage($connection, $data){//websocket发送过来的消息$connection->send('我收到你的信息了:'.$data);//rabbitMQ配置$options = ['host'=>'127.0.0.1',//rabbitMQ IP'port'=>5672,//rabbitMQ 通讯端口'user'=>'admin',//rabbitMQ 账号'password'=>'123456'//rabbitMQ 密码];(new Client($options))->connect()->then(function (Client $client) {return $client->channel();})->then(function (Channel $channel) {/*** 创建队列(Queue)* name: ceshi         // 队列名称* passive: false      // 如果设置true存在则返回OK，否则就报错。设置false存在返回OK，不存在则自动创建* durable: true       // 是否持久化，设置false是存放到内存中RabbitMQ重启后会丢失,*                        设置true则代表是一个持久的队列，服务重启之后也会存在，因为服务会把持久化的Queue存放在硬盘上，当服务重启的时候，会重新加载之前被持久化的Queue* exclusive: false    // 是否排他，指定该选项为true则队列只对当前连接有效，连接断开后自动删除*  auto_delete: false // 是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除*/return $channel->queueDeclare('ceshi', false, true, false, false)->then(function () use ($channel) {return $channel;});})->then(function (Channel $channel) use($data){echo "发送消息内容：".$data."\n";/*** 发送消息* body 发送的数据* headers 数据头，建议 ['content_type' => 'text/plain']，这样消费端是springboot注解接收直接是字符串类型* exchange 交换器名称* routingKey 路由key* mandatory* immediate* @return bool|PromiseInterface|int*/return $channel->publish($data, ['content_type' => 'text/plain'], '', 'ceshi')->then(function () use ($channel) {return $channel;});})->then(function (Channel $channel) {//echo " [x] Sent 'Hello World!'\n";$client = $channel->getClient();return $channel->close()->then(function () use ($client) {return $client;});})->then(function (Client $client) {$client->disconnect();});}/*** 当连接建立时触发的回调函数* @param $connection*/public function onConnect($connection){}/*** 当连接断开时触发的回调函数* @param $connection*/public function onClose($connection){}/*** 当客户端的连接上发生错误时触发* @param $connection* @param $code* @param $msg*/public function onError($connection, $code, $msg){echo "error $code $msg\n";}/*** 每个进程启动* @param $worker*/public function onWorkerStart($worker){}
}

• 在config/worker_server.php中设置worker_class值为'app\workerman\Send'

• 启动这个生产者

php think worker:server

方式1：通过tcp发送数据

• 发送数据
  通过在线网址发送数据（websocket方式），->传送门<-
  输入【ws://127.0.0.1:2345】后点击发送数据！
  
• 前往rabbitMQ控制台查看
  
  至此，生产这一步就走完了，那么如果我不想通过websocket方式，想用tcp方式生产数据怎么办？

方式2：通过tcp发送数据

接口给内置服务器发消息->内置服务去发消息给rabbitMQ

• 将Send.php中websocket:127.0.0.1改成tcp:127.0.0.1
• 重启服务
• 把controller目录中Index.php修改为以下内容

<?php
namespace app\controller;use app\BaseController;class Index extends BaseController
{public function index(string $msg){//连接本地tcp服务$client = stream_socket_client('tcp://127.0.0.1:2345', $errno, $errmsg, 1);//发送字符串fwrite($client, $msg."\n");//断开服务fclose($client);return 'OK';}}

• 用Postman访问对应接口就好，也会有数据进入队列

消费者

同生产者一样新创建一个thinkphp6项目，注意端口别和生产者冲突！这里我设置的是2346端口

• 在app目录下新建workerman目录，并在其下创建Receive.php文件，$options数组中的host地址改成你的rabbitmq地址。

<?phpnamespace app\workerman;
use Bunny\Channel;
use Bunny\Message;
use Workerman\RabbitMQ\Client;
use think\worker\Server;
class Receive extends Server
{protected $socket = 'tcp://127.0.0.1:2346';/*** 收到信息* @param $connection* @param $data*/public function onMessage($connection, $data){}/*** 当连接建立时触发的回调函数* @param $connection*/public function onConnect($connection){}/*** 当连接断开时触发的回调函数* @param $connection*/public function onClose($connection){}/*** 当客户端的连接上发生错误时触发* @param $connection* @param $code* @param $msg*/public function onError($connection, $code, $msg){echo "error $code $msg\n";}/*** 每个进程启动* @param $worker*/public function onWorkerStart($worker){//rabbitMQ配置$options = ['host'=>'127.0.0.1',//rabbitMQ IP'port'=>5672,//rabbitMQ 通讯端口'user'=>'admin',//rabbitMQ 账号'password'=>'123456'//rabbitMQ 密码];(new Client($options))->connect()->then(function (Client $client) {return $client->channel();})->then(function (Channel $channel) {/*** 创建队列(Queue)* name: ceshi         // 队列名称* passive: false      // 如果设置true存在则返回OK，否则就报错。设置false存在返回OK，不存在则自动创建* durable: true       // 是否持久化，设置false是存放到内存中RabbitMQ重启后会丢失,*                        设置true则代表是一个持久的队列，服务重启之后也会存在，因为服务会把持久化的Queue存放在硬盘上，当服务重启的时候，会重新加载之前被持久化的Queue* exclusive: false    // 是否排他，指定该选项为true则队列只对当前连接有效，连接断开后自动删除*  auto_delete: false // 是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除*/return $channel->queueDeclare('ceshi', false, true, false, false)->then(function () use ($channel) {return $channel;});})->then(function (Channel $channel) {echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";$channel->consume(function (Message $message, Channel $channel, Client $client) {echo "接收消息内容：", $message->content, "\n";},'ceshi','',false,true);});}
}

• 在config/worker_server.php中设置worker_class值为'app\workerman\Receive'，并将端口改为2346

• 启动这个消费者

php think worker:server

到这里消费者也就结束啦！

使用

接下来我用cmd来启动两个服务，然后用接口发送消息和消费测试！

部分参考自：
https://www.workerman.net/doc/workerman/components/workerman-rabbitmq.html
https://blog.csdn.net/weixin_47723549/article/details/124493059