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 告诉服务器（它仍然有一个完整的副本）将其转发到死信交换。

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

连接恢复

一些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代码的移植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

mruell / php-amqplib

维护者

详细信息