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地址。
<?php
namespace 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地址。
<?php
namespace 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