php 中使用Rabbitmq实现实现消息发送和接收

66 阅读 0 评论 44 点赞

我是靠谱客的博主过时雪碧，这篇文章主要介绍php 中使用Rabbitmq实现实现消息发送和接收，现在分享给大家，希望可以做个参考。

php 中使用Rabbitmq实现实现消息发送和接收

1，建立一个send.php文件用来发送消息

2，建立一个 receive.php 文件用来接收消息

代码如下
send.php

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
<?php
/**
* 发送消息
*/
$exchangeName = 'demo';
$routeKey = 'hello';
$message = 'Hello World!';
// 建立TCP连接
$connection = new AMQPConnection([
'host' => 'localhost',
'port' => '5672',
'vhost' => '/',
'login' => 'guest',
'password' => 'guest'
]);
$connection->connect() or die("Cannot connect to the broker!n");
try {
$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->declareExchange();
echo 'Send Message: ' . $exchange->publish($message, $routeKey) . "n";
echo "Message Is Sent: " . $message . "n";
} catch (AMQPConnectionException $e) {
var_dump($e);
}
$connection->disconnect();// 断开连接

receive.php

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
<?php
/**
* 接收消息
*/
$exchangeName = 'demo';
$queueName = 'hello';
$routeKey = 'hello';
// 建立TCP连接
$connection = new AMQPConnection([
'host' => 'localhost',
'port' => '5672',
'vhost' => '/',
'login' => 'guest',
'password' => 'guest'
]);
$connection->connect() or die("Cannot connect to the broker!n");
$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_DIRECT);
echo 'Exchange Status: ' . $exchange->declareExchange() . "n";
$queue = new AMQPQueue($channel);
$queue->setName($queueName);
echo 'Message Total: ' . $queue->declareQueue() . "n";
echo 'Queue Bind: ' . $queue->bind($exchangeName, $routeKey) . "n";
var_dump("Waiting for message...");
// 消费队列消息
while(TRUE) {
$queue->consume('processMessage');
}
// 断开连接
$connection->disconnect();
function processMessage($envelope, $queue) {
$msg = $envelope->getBody();
var_dump("Received: " . $msg);
$queue->ack($envelope->getDeliveryTag()); // 手动发送ACK应答
}