README

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

该库被用于 RabbitMQ in Action 的 PHP 示例以及官方 RabbitMQ 教程。

请注意，该项目遵循贡献者行为准则。参与此项目即表示您同意遵守其条款。

项目维护者

感谢 videlalvaro 和 postalservice14 维护 php-amqplib 所付出的辛勤努力！没有他们，这个库不可能发展到现在的水平。

现在，该包由 nubeiro 和一些在 RabbitMQ 及相关项目上工作的 Pivotal 工程师维护。

支持的 RabbitMQ 版本

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

支持的 RabbitMQ 扩展

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

Exchange to Exchange Bindings
Basic Nack
Publisher Confirms
Consumer Cancel Notify

支持修改现有方法（如 alternate exchanges）的扩展。

设置

确保您已安装 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通道中读取并立即丢弃，因此您无法在回调中检索它。如果您有另一个可以处理具有更大有效负载的消息的消费者，您可以使用basic_reject或basic_nack告诉服务器（它仍然有一个完整的副本）将其转发到死信交换。

默认情况下，不会发生截断。要禁用已启用截断的通道上的截断，请将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

基于信号的Heartbeat

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

<?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代码移植到PHP版本。

它已与RabbitMQ服务器进行过测试。

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

有关讨论，请加入此小组

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

有关错误报告，请使用项目页面的错误跟踪系统。

欢迎提交补丁！

作者：Vadim Zaliva lord@crocodile.org

greenhat616 / php-amqplib

维护者

详情