住房与建设局网站,网站建设项目需求费用,网站关键词排名没有了,个人网站设计规划书Thinkphp6连接使用RabbitMQ#xff08;不止tp6#xff0c;其他框架对应改下也一样#xff09;#xff0c;如何使用Docker部署RabbitMQ#xff0c;在上一篇已经讲了-传送门-。 部署环境 开始前先进入RabbitMQ的web管理界面#xff0c;选择Queues菜单#xff0c;点… 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 [host127.0.0.1,//rabbitMQ IPport5672,//rabbitMQ 通讯端口useradmin,//rabbitMQ 账号password123456//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 [host127.0.0.1,//rabbitMQ IPport5672,//rabbitMQ 通讯端口useradmin,//rabbitMQ 账号password123456//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 CTRLC, \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
