基于PHP使用集成RabbitMQ

基于PHP使用集成RabbitMQ

一。RabbitMQ简要概括

1、AMQP：Advanced Message Queuing Protocol，是一个提供统一消息服务的应用层标准协议。

2、IPC（单一系统进程间通信） -> socket（不同机器间进程通信） -> AMQP（解决大型系统模块与组件间通信）

3、RabbitMQ 基于 Erlang 开发，是 AMQP 的一个开源实现。

4、RabbitMQ 系统架构图：

5、名词术语：

RabbitMQ Server（broker server）：维护一条从 Producer 到 Consumer 的路线，保证数据能够按照指定的方式进行传输；Client A & B：数据发送方，Producers create messages and publish (send) them to a broker server (RabbitMQ)，一个有效的 Message 包含 payload 和 label 两部分Client 1、2、3：数据消费方，Consumers attach to a broker server (RabbitMQ) and subscribe to a queueExchange：Exchanges are where producers publish their messagesQueue： Queues are where the messages end up and are received by consumersBinding：Bindings are how the messages get routed from the exchange to particular queues

还有几个隐式的概念：

Connection：Producer 和 Consumer 通过 TCP 连接到 RabbitMQChannel：它建立在上述的 TCP 连接中，数据流动都是在 Channel 中进行的

此外，Exchanges 分三种类型：

direct：如果 routing key 匹配，那么 Message 就会被传递到相应的 queuefanout：会向响应的 queue 广播topic：对 key 进行模式匹配，比如 ab* 可以传递到所有 ab* 的 queue

二、安装PHP-amqp扩展

https://pecl.php.net/package/amqp/1.4.0/windows

先查看自己的php版本

接下来下载dll文件 地址http://pecl.php.net/package/amqp

下载稳定版的，点击DLL

php版本 ，X86 和X64 根据自己情况  ， NTS  和 TS 就是那个thread safty 的状态 这个大家都会看吧 就不多说了

下载解压

将php_amqp.dll文件放到php目录的ext文件夹下  见下图：

 将rabbitmq.1.dll文件放到php根目录 见下图：

 php.ini里面添加    

之后重启NGinx

报错：Fatal error: Uncaught exception 'AMQPConnectionException' with message 'Library error: connection closed unexpectedly - Potential login failure.' in 

未解决

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

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

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

代码如下

send.php

 '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

 '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应答
}
测试：

打开两个终端，先运行接收者脚本监听消息发送：

php receive.php

在另一个终端中运行消息发送脚本：

php send.php