当前位置：首页 > news >正文

php使用rabbitmq

news 2025/8/10 5:22:46

这里使用的是 php-amqplib

 composer require php-amqplib/php-amqplib

生产端send.php

<?phprequire_once "./vendor/autoload.php";use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;//交换机名称
$exc_name = 'exchange.canal';
//路由键
$routing_key = 'canal-routing-key';
//队列名称
$queue_key = 'canal_queue';//生产者
//Connection: publisher／consumer和broker之间的TCP连接
//Channel: 如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCP Connection的开销将是巨大的，效率也较低。
//Channel是在connection内部建立的逻辑连接Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销。//1.建立connction
/*** 主要参数说明：* $host:  RabbitMQ服务器主机IP地址* $port:  RabbitMQ服务器端口* $user:  连接RabbitMQ服务器的用户名* $password:  连接RabbitMQ服务器的用户密码* $vhost:   连接RabbitMQ服务器的vhost（服务器可以有多个vhost，虚拟主机，类似nginx的vhost）*/
$connection = new AMQPStreamConnection('192.168.0.105', 5672, 'admin', 'admin', "/");
//2.建立通道Channel
$channel = $connection->channel();//3.声明交换器
/*** exchange_declare($exchange, $type, $passive = false, $durable = false, $auto_delete = true, $internal = false, $nowait = false, $arguments = array(), $ticket = null) 。* 试探性申请一个交换器，若该交换器不存在，则创建；若存在，则跳过。* 主要参数说明：* $exchange：队列名称* $type 交换器类型，常见类型如：fanout,dircet,topic,hraders四种* $passive 只判断不创建，一般是判断该交换机是否存在* $durable：是否开启持久化 设置true表示持久化，反之非持久化。持久化可以将交换器存盘，在服务器重启的适合不会丢失相关数据。* $auto_delete：是否自动删除* $internal：设置是否内置，设置true表示内置交换器，客户端程序无法直接发送消息到这个交换器中,只能通过交换器路由到交换器这个方式* $nowait：如果为true表示不等待服务器回执信息，函数将返回NULL，可提高访问速度。*/
$channel->exchange_declare($exc_name, 'direct', false, true, false);//4.声明队列
/*** queue_declare($queue, $passive = false, $durable = false, $exclusive = false, $auto_delete = true, $nowait = false, $arguments = null, $ticket = null)* 试探性申请一个队列，若该队列不存在，则创建；若存在，则跳过。* 主要参数说明：* $queue：队列名称* $passive 只判断不创建，一般是判断该队列是否存在* $durable：是否开启持久化 设置true表示持久化，反之非持久化。持久化可以将队列存盘，在服务器重启的适合不会* $exclusive：是否独占，设置true表示独占，只能有一个消费者监听该队列，false表示共享，可以有多个消费者监听该队列。*/
$channel->queue_declare($queue_key, false, true, false, false);//5.绑定队列到交换器
/*** queue_bind($queue, $exchange, $routing_key = '', $nowait = false, $arguments = null, $ticket = null)* 绑定队列到交换器* 主要参数说明：* $queue：队列名称* $exchange：交换器名称* $routing_key：路由键*/
$channel->queue_bind($queue_key, $exc_name, $routing_key);//4.创建消息
$content = 'this is ' . $routing_key . ' message';
/*** 主要参数说明：*  data：消息*  properties Array 设置的属性，比如设置该消息是否持久化* DELIVERY_MODE_NON_PERSISTENT 不持久化* DELIVERY_MODE_PERSISTENT 持久化*/
$msg = new AMQPMessage($content, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]);//5.发布消息
/*** basic_publish主要参数说明：* $msg 消息内容* $exchange 交换器* $routing_key routing_key或者队列名称* $mandatory 匹配不到队列时，是否立即丢弃消息* $immediate 队列无消费者时，是否立即丢弃消息* $ticket 不清楚是什么作用*/
$channel->basic_publish($msg, $exc_name, $routing_key);
//7.通道关闭
$channel->close();
//8.连接关闭
$connection->close();

消费端receive.php

<?php
require_once "./vendor/autoload.php";use PhpAmqpLib\Connection\AMQPStreamConnection;//交换机名称
$exc_name  = 'exchange.canal';
//路由键
$routing_key = 'canal-routing-key';//1.建立connction
$connection = new AMQPStreamConnection('192.168.0.105', 5672, 'admin', 'admin', "/");
//2.建立通道Channel
$channel = $connection->channel();//3.声明交换器
$channel->exchange_declare($exc_name, 'direct', false, true, false);//4.获取系统生成的消息队列名称，这里也可以指定一个队列名称
list($queue_name, ,) = $channel->queue_declare('', false, true, true, false);//5.将队列名与交换器名进行绑定，并指定routing_key
$channel->queue_bind($queue_name,$exc_name,$routing_key);//6.设置消费成功后才能继续进行下一个消费
$channel->basic_qos(null, 1, null);//7.定义收到消息回调函数
$callback = function ($msg) {echo 'received = ', $msg->body . "\n";//确认消息已被消费，从生产队列中移除$msg->ack();
};//8.开启消费no_ack=false,设置为手动应答
$channel->basic_consume($queue_name, '', false, false, false, false, $callback);//9.不断的循环进行消费
while ($channel->is_open()) {$channel->wait();
}//关闭连接
$channel->close();
$connection->close();

参考文章