README

这个库是 纯 PHP 实现的 AMQP 0-9-1 协议。它已经与 RabbitMQ 进行过测试。

这个库被用于 RabbitMQ in Action 的 PHP 示例和官方 RabbitMQ 教程。

请注意，该项目采用贡献者行为准则发布。通过参与此项目，您同意遵守其条款。

项目维护者

感谢 videlalvaro 和 postalservice14 创建 php-amqplib。

该项目现在由 Ramūnas Dronga、Luke Bakken 以及 RabbitMQ 上的几位 VMware 工程师维护。

支持的 RabbitMQ 版本

从版本 2.0 开始，这个库默认使用 AMQP 0.9.1，因此需要 RabbitMQ 2.0 或更高版本。通常服务器升级不需要修改任何应用程序代码，因为协议变更非常少，但在升级之前请自行测试。

支持的 RabbitMQ 扩展

由于库使用 AMQP 0.9.1，我们添加了对以下 RabbitMQ 扩展的支持

交换机到交换机的绑定
基本否定
发布者确认
消费者取消通知

修改现有方法（如 替代交换机）的扩展也得到支持。

设置

请确保您已安装 composer，然后运行以下命令

$ composer require php-amqplib/php-amqplib

这将从您的 vendor 文件夹中检索库及其依赖项。然后您可以将以下内容添加到您的 .php 文件中，以便使用该库

require_once __DIR__.'/vendor/autoload.php';

然后您需要使用相关的类，例如

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

用法

在 RabbitMQ 运行时，打开两个终端，在第一个终端中执行以下命令以启动消费者

$ cd php-amqplib/demo
$ php amqp_consumer.php

然后在另一个终端中

$ cd php-amqplib/demo
$ php amqp_publisher.php some text to publish

您应该会看到消息到达另一个终端上的进程

然后要停止消费者，向其发送 quit 消息

$ php amqp_publisher.php quit

如果您需要监听连接到 RabbitMQ 的套接字，请参阅非阻塞消费者中的示例。

$ php amqp_consumer_non_blocking.php

变更日志

请参阅 CHANGELOG 了解最近更改的详细信息。

API 文档

http://php-amqplib.github.io/php-amqplib/

教程

为了避免重复，如果您想了解更多关于这个库的信息，请参考官方RabbitMQ教程。

多个主机连接

如果您有一个应用程序可以连接的多个节点集群，您可以使用主机数组启动一个连接。为此，您应该使用create_connection静态方法。

例如

$connection = AMQPStreamConnection::create_connection([
    ['host' => HOST1, 'port' => PORT, 'user' => USER, 'password' => PASS, 'vhost' => VHOST],
    ['host' => HOST2, 'port' => PORT, 'user' => USER, 'password' => PASS, 'vhost' => VHOST]
],
$options);

此代码将首先尝试连接到HOST1，如果第一个连接失败，则连接到HOST2。该方法返回第一个成功连接的连接对象。如果所有连接都失败，它将抛出最后一次连接尝试的异常。

有关更多示例，请参阅demo/amqp_connect_multiple_hosts.php。

批量发布

假设您有一个生成大量消息并使用相同exchange、相同routing_key和选项（如mandatory）进行发布的进程。然后，您可以利用batch_basic_publish库功能。您可以像这样批量消息

$msg = new AMQPMessage($msg_body);
$ch->batch_basic_publish($msg, $exchange);

$msg2 = new AMQPMessage($msg_body);
$ch->batch_basic_publish($msg2, $exchange);

然后像这样发送批量

$ch->publish_batch();

何时发布消息批量？

假设我们的程序需要从文件中读取并逐行发布一条消息。根据消息大小，您将不得不决定何时发送批量。您可以选择每50条消息发送一次，或者每100条。这取决于您。

优化的消息发布

加快消息发布速度的另一种方法是重用AMQPMessage消息实例。您可以创建新的消息如下

$properties = array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT);
$msg = new AMQPMessage($body, $properties);
$ch->basic_publish($msg, $exchange);

现在假设，虽然您想要更改未来消息的消息体，但您将保持相同的属性，即，您的消息仍然是text/plain，并且delivery_mode仍然是AMQPMessage::DELIVERY_MODE_PERSISTENT。如果您为每条发布的消息创建一个新的AMQPMessage实例，那么这些属性将必须重新编码在AMQP二进制格式中。您可以通过重用AMQPMessage并按如下方式重置消息体来避免所有这些

$msg->setBody($body2);
$ch->basic_publish($msg, $exchange);

截断大型消息

AMQP对消息大小没有限制；如果消费者收到一个非常大的消息，PHP的内存限制可能在调用传递给basic_consume的回调之前在库中被达到。

为了避免这种情况，您可以在您的Channel实例上调用方法AMQPChannel::setBodySizeLimit(int $bytes)。超过此限制的正文大小将被截断，并通过设置标志AMQPMessage::$is_truncated为true的回调传递。属性AMQPMessage::$body_size将反映接收到的消息的真实正文大小，如果消息已被截断，则该大小将高于strlen(AMQPMessage::getBody())。

请注意，所有超出限制的数据都将从AMQP Channel中读取并立即丢弃，因此您无法在回调中检索它。如果您有另一个可以处理具有更大有效载荷的消息的消费者，您可以使用basic_reject或basic_nack告诉服务器（它仍然有一个完整的副本）将其转发到死信交换。

默认情况下，不会发生截断。要禁用在已启用截断的Channel上的截断，请将0（或null）传递给AMQPChannel::setBodySizeLimit()。

连接恢复

一些RabbitMQ客户端使用自动化的连接恢复机制，在网络错误发生时重新连接并恢复通道和消费者。

由于此客户端使用单线程，您可以使用异常处理机制设置连接恢复。

可能因连接错误而抛出的异常

PhpAmqpLib\Exception\AMQPConnectionClosedException
PhpAmqpLib\Exception\AMQPIOException
\RuntimeException
\ErrorException

可能会抛出其他异常，但连接仍然存在。在重新连接之前处理异常时，总是清理旧连接是个好主意。

例如，如果你想设置一个恢复连接

$connection = null;
$channel = null;
while(true){
    try {
        $connection = new AMQPStreamConnection(HOST, PORT, USER, PASS, VHOST);
        // Your application code goes here.
        do_something_with_connection($connection);
    } catch(AMQPRuntimeException $e) {
        echo $e->getMessage();
        cleanup_connection($connection);
        usleep(WAIT_BEFORE_RECONNECT_uS);
    } catch(\RuntimeException $e) {
        cleanup_connection($connection);
        usleep(WAIT_BEFORE_RECONNECT_uS);
    } catch(\ErrorException $e) {
        cleanup_connection($connection);
        usleep(WAIT_BEFORE_RECONNECT_uS);
    }
}

完整的示例在 demo/connection_recovery_consume.php 中。

此代码将在每次异常发生时重新连接并重试应用程序代码。某些异常仍然可能抛出，不应作为重新连接过程的组成部分进行处理，因为它们可能是应用程序错误。

这种方法主要适用于消费者应用程序，生产者将需要一些额外的应用程序代码来避免多次发布相同的信息。

这是一个最简单的示例，在实际应用程序中，你可能想控制重试次数，并可能优雅地降低重新连接的等待时间。

你可以在 #444 中找到一个更极端的例子。

UNIX 信号

如果你已安装 PCNTL 扩展，则在消费者未处理消息时将处理信号的分发。

$pcntlHandler = function ($signal) {
    switch ($signal) {
        case \SIGTERM:
        case \SIGUSR1:
        case \SIGINT:
            // some stuff before stop consumer e.g. delete lock etc
            pcntl_signal($signal, SIG_DFL); // restore handler
            posix_kill(posix_getpid(), $signal); // kill self with signal, see https://www.cons.org/cracauer/sigint.html
        case \SIGHUP:
            // some stuff to restart consumer
            break;
        default:
            // do nothing
    }
};

pcntl_signal(\SIGTERM, $pcntlHandler);
pcntl_signal(\SIGINT,  $pcntlHandler);
pcntl_signal(\SIGUSR1, $pcntlHandler);
pcntl_signal(\SIGHUP,  $pcntlHandler);

要禁用此功能，只需将常量 AMQP_WITHOUT_SIGNALS 定义为 true。

<?php
define('AMQP_WITHOUT_SIGNALS', true);

... more code

基于信号的心跳

如果你已安装 PCNTL 扩展并使用 PHP 7.1 或更高版本，则可以注册一个基于信号的发送器。

<?php

$sender = new PCNTLHeartbeatSender($connection);
$sender->register();
... code
$sender->unregister();

调试

如果你想了解协议级别上正在发生什么，请在你的代码中添加以下常量

<?php
define('AMQP_DEBUG', true);

... more code

?>

基准测试

要运行发布/消费基准测试，请输入

$ make benchmark

测试

要成功运行测试，你首先需要在本地上运行一个库存 RabbitMQ 代理。然后，像这样运行测试

$ make test

贡献

请参阅 CONTRIBUTING 获取详细信息。

使用 AMQP 0.8

如果你仍然想使用旧版本的协议，则可以在配置代码中设置以下常量来实现

define('AMQP_PROTOCOL', '0.8');

默认值是 '0.9.1'。

提供你自己的自动加载器

如果你出于某种原因不想使用 composer，那么你需要为库类提供一个自动加载器。有人已经报告使用这个自动加载器成功。

原始 README

以下为原始 README 文件内容。归功于原始作者。

PHP 库实现高级消息队列协议 (AMQP)。

该库是 py-amqplib 的 python 代码的移植版本 http://barryp.org/software/py-amqplib/

已在 RabbitMQ 服务器上进行了测试。

项目主页： http://code.google.com/p/php-amqplib/

请加入小组进行讨论

http://groups.google.com/group/php-amqplib-devel

请使用项目页面上的错误跟踪系统提交错误报告。

欢迎贡献补丁！

作者：Vadim Zaliva lord@crocodile.org

game-worker / kiri-amqplib

维护者

详情