最近因为业务需要,要用到MQ就去研究了一下,说实话,安装环境给我搞自闭了,大概是我太菜
刚开始使用yum换源,各种安装卸载始终找不到自己要用的版本,后来全部卸载,下载安装包
编译安装解百忧
我用的是erlang 25.3 的版本,MQ使用的是3.11.3的版本,符合官方要求,这里的版本是有强制要求的,也就是固定erlang对应固定MQ,版本如下
搞错,会无法运行
编译安装erlang25
编译安装老一套,
解压,
cd 进目录
./configure (这里我记得需要有一些 选项,不过我好像没安装,就不多说了)
make && make install
漫长的等待...
结束后
vim /etc/profile
增加
export PATH=$PATH:(你的erlang解压目录)/otp_src_25.3/bin
source /etc/profile(刷新环境变量,使其生效)
然后运行erl 看会不会有反应 安装成功后应该是
接下来就是 安装MQ 相比erlang 简单很多,因为是编译安装,所以systemctl无法管理这个
下载MQ安装包.然后 解压缩,cd进解压后的目录,里面有个sbin cd 进入
然后使用 ./rabbitmq-server 命令就可以启动了
这样就是成功了
接下来就是安装插件,进入目录 使用 ./rabbitmq-plugins list 查看安装的插件目录
这里需要安装一个插件 指令为 rabbitmq-plugins enable rabbitmq_management
这是MQ的web管理插件,记得开启15672端口,不然外部无法访问,默认账号密码都是guest
但是在本地登录时不行的,需要创建账号密码,
./rabbitmqctl add_user <你的用户名> <你的密码>
./rabbitmqctl set_permissions -p / <你的用户名> ".*" ".*" ".*"
然后就可以登陆了
到这里,算是安装完成了.接下来就是基础使用
我使用的是thinkPHP
首先使用composer 安装 MQ官方库
composer require php-amqplib/php-amqplib:3.5.0
安装后先启动 消费端
//这是链接,个人习惯,有问题的话还请指正
public function __construct()
{
parent::__construct();
if ($this->connection != null) {
return $this->connection;
}
try {
$this->connection = (new AMQPStreamConnection(
$this->host, ip
$this->port, 端口
$this->user, 用户
$this->password, 密码
$this->vhost, 暂未用到 默认 /
));
$this->channel = $this->connection->channel();
}catch (\Exception $e) {
Log::write("连接错误：".$e->getMessage(), "rabbitmq-log");
throw new apiException("MQ连接异常".$e->getMessage());
}
}
protected function execute(Input $input, Output $output)
{
//交换机 交换机分多种,常用的是直连
$this->exchange = $input->getArgument('exchange');
//通道 消费对应通道的消息
$this->quent = $input->getArgument('quent');
//测试
$type = $mqType ? 'x-delayed-message' : "direct";
$key = $mqType ? 'my_routing_key' : "";
$durable = $mqType;
exchange：要声明的exchange名称，必须是一个非空字符串。
//type：exchange类型，可以是direct、topic、headers或fanout之一。
//passive：如果设置为true，则不会创建exchange，仅检查是否存在指定名称的exchange。默认为false。
//durable：如果设置为true，则exchange将在重启后继续存在。默认为false。
//auto_delete：如果设置为true，则exchange将在最后一个绑定被删除后自动删除。默认为false。
//internal：如果设置为true，则exchange将用于内部用途，不允许客户端发布消息到该exchange。默认为false。
//arguments：其他参数，通常为键值对，用于指定exchange的特定属性。默认为null。
$this->channel->exchange_declare($this->exchange, "direct", false, true, false);
queue：队列名称，如果该队列不存在则会创建。
passive：如果为true，则不会创建队列。如果队列不存在，则会返回一个错误。
durable：如果为true，则队列将被持久化。当RabbitMQ服务器重启时，队列将保留下来。
exclusive：如果为true，则队列只能被当前连接使用，当连接关闭时队列将被删除。该选项通常用于创建临时队列。
auto_delete：如果为true，则当没有消费者使用时自动删除队列。
arguments：一些额外的参数，例如队列的最大长度、消息的过期时间等。
$this->channel->queue_declare($this->quent, false, true, false, false);
//$this->quent 队列
//$this->exchange 交换机 进行绑定
$this->channel->queue_bind($this->quent, $this->exchange, "");
//一次处理1条
$this->channel->basic_qos(0, 1, false);
$callback = function ($msg) {
//接收数据
$info = json_decode($msg->body, true);
//进行接收,否则以上文的配置 消息会一直存在
$this->channel->basic_ack($msg->delivery_info['delivery_tag']);
};
//queue：需要消费的队列名称。
//consumer_tag 是用于唯一标识一个消费者的字符串
//no_local 防止消费者接收到自己发送的消息
//no_ack指的是消费者在消费完消息后不需要向服务器发送确认消息
//Exclusive指定了队列的排他性。如果一个队列被标记为Exclusive，那么只有创建该队列的连接可以使用它
//nowait是指在使用 IBM MQ API 发送或获取消息时，如果队列中没有可用的消息，API 将立即返回，而不是等待消息到达。这可以避免程序在等待消息时被阻塞，从而提高程序的效率。
//callback 接收消息后的处理函数
$this->channel->basic_consume($this->quent, '', false, false, false, false, $callback,);
while (count($this->channel->callbacks)) {
try {
$this->channel->wait();
} catch (\Exception $e) {
echo $e->getFile() . $e->getLine() . $e->getMessage();
return "error";
}
}
//关闭链接
$this->channel->close();
$this->connection->close();
}因为使用的tp6 所有在根目录 使用 php think 文件名就可以执行了 自定义指令 · ThinkPHP6.0完全开发手册 · 看云 这是官方文档
接下来说发布者
public function __construct()
{
parent::__construct();
if ($this->connection != null) {
return $this->connection;
}
try {
$this->connection = (new AMQPStreamConnection(
$this->host, ip
$this->port, 端口
$this->user, 用户
$this->password, 密码
$this->vhost, 暂未用到 默认 /
));
$this->channel = $this->connection->channel();
}catch (\Exception $e) {
Log::write("连接错误：".$e->getMessage(), "rabbitmq-log");
throw new apiException("MQ连接异常".$e->getMessage());
}
}
public function send(){
//exchange：要声明的 Exchange 的名称。
//type：Exchange 的类型，如 direct、topic、fanout、headers 等。
//passive：如果设置为 true，则不会创建 Exchange，只会检查该 Exchange 是否存在。默认为 false。
//durable：如果设置为 true，则 Exchange 会被持久化到磁盘上，即使 RabbitMQ 服务器重启也不会丢失。默认为 false。
//auto_delete：如果设置为 true，则当没有队列或者交换机绑定到该 Exchange 时，自动删除该 Exchange。默认为 false。
//internal：如果设置为 true，则 Exchange 不能被客户端用来发送消息，只能被 Exchange 和队列之间的绑定使用。默认为 false。
//arguments：用于声明 Exchange 的其他参数，如 x-message-ttl、x-dead-letter-exchange 等。默认为 null。
$this->channel->exchange_declare($this->exchange, $this->type, true, false, false);
//queue：队列名称，如果为空则表示由消息代理自动生成一个唯一的队列名称。
//passive：如果为true，则不会创建该队列，而是检查该队列是否存在。如果该队列不存在，则返回一个通道级别的异常。
//durable：如果为true，则表示该队列是持久化的，即在消息代理重启后该队列仍然存在。
//exclusive：如果为true，则表示该队列只能被当前连接使用，并且在连接断开时会自动删除该队列。
//auto_delete：如果为true，则表示该队列在没有任何消费者使用时会自动删除。
//arguments：用于传递额外的参数，例如队列的最大长度、消息过期时间等。*/
$this->channel->queue_declare($this->quent, false, false, false, false);
//发送消息
$message = new AMQPMessage($msg);
}
$this->channel->basic_publish($message, $this->exchange, $this->key);
}到这里就算完结了 这里是即时到达,还有延时到达会麻烦一点
首先两种实现方法,我用的比较简单,安装插件
./rabbitmq-plugins enable rabbitmq_delayed_message_exchange
然后创建延时交换机,和即时到达的有区别,建议去web管理端创建
插件完成后 会有这个选项
这就是最终填写的数据
提交后就有了.然后 发布和上面是一样的 ,不过类型要变 type 变成 x-delayed-message 也要设置 key 参数durable 变量为 true
接收的话 也是同理,type变化为 x-delayed-message 参数durable 变量为 true
这里是你设置的key
到这里 就结束了,安装 发布,消费,.整个过程,写的比较粗糙,有啥错误还请指正
