摘要://安装类库composerquirephp amqplib/php amqplb//consumer。php消费者代码要求__ DIR__。“/../vendor/autoload.php”;使用PhpAmqpLibConnectionAMQPStreamConnection;$conf=['host'=˃'127.0.0.1','port'=˃5672,'user'=˃'','
//安装类库
composer require php-amqplib/php-amqplib
//consumer.php消费者代码
require __DIR__ . "/../vendor/autoload.php";
usePhpAmqpLibConnectionAMQPStreamConnection;
$conf =[
'host' => '127.0.0.1',
'port' => 5672,
'user' => '',
'pwd' => '',
'vhost' => '/',];
$exchangeName = '_ex'; //交换机名
$queueName = '_que'; //队列名称
$routingKey = '_send'; //路由关键字(也可以省略)
$conn = new AMQPStreamConnection( //建立生产者与mq之间的连接
$conf['host'], $conf['port'], $conf['user'], $conf['pwd'], $conf['vhost']
);
$channel = $conn->channel(); //在已连接基础上建立生产者与mq之间的通道
$channel->exchange_declare($exchangeName, 'direct', false, true, false); //声明初始化交换机
$channel->queue_declare($queueName, false, true, false, false); //声明初始化一条队列
$channel->queue_bind($queueName, $exchangeName, $routingKey); //将队列与某个交换机进行绑定,并使用路由关键字
//1:queue 要取得消息的队列名
//2:consumer_tag 消费者标签
//3:no_local false这个功能属于AMQP的标准,但是rabbitMQ并没有做实现.参考
//4:no_ack false收到消息后,是否不需要回复确认即被认为被消费
//5:exclusive false排他消费者,即这个队列只能由一个消费者消费.适用于任务不允许进行并发处理的情况下.比如系统对接
//6:nowait false不返回执行结果,但是如果排他开启的话,则必须需要等待结果的,如果两个一起开就会报错
//7:callback null回调函数
//8:ticket null
//9:arguments null
$channel->basic_consume($queueName, '', false, true, false, false, function($msg){
var_dump($msg->delivery_info);
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);//手动应答
});
//监听消息
while(count($channel->callbacks)){
$channel->wait();
}
//publisher.php生产者代码
require __DIR__ . "/../vendor/autoload.php";
usePhpAmqpLibConnectionAMQPStreamConnection;
usePhpAmqpLibMessageAMQPMessage;
$conf =[
'host' => '127.0.0.1',
'port' => 5672,
'user' => '',
'pwd' => '',
'vhost' => '/',];
$exchangeName = '_ex'; //交换机名
$routingKey = '_send'; //路由关键字(也可以省略)
$conn = new AMQPStreamConnection( //建立生产者与mq之间的连接
$conf['host'], $conf['port'], $conf['user'], $conf['pwd'], $conf['vhost']
);
$channel = $conn->channel(); //在已连接基础上建立生产者与mq之间的通道
$channel->exchange_declare($exchangeName, 'direct', false, true, false); //声明初始化交换机
$msgBody = json_encode(["msg" => "hello"]);
$msg = new AMQPMessage($msgBody, ['content_type' => 'text/plain', 'delivery_mode' => 2]); //生成消息
$r = $channel->basic_publish($msg, $exchangeName, $routingKey); //推送消息到某个交换机
$channel->close();
$conn->close();