README

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

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

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

项目维护者

感谢 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通道中读取并立即丢弃，因此您无法在回调中检索它。如果您有另一个可以处理较大有效负载的消息的消费者，可以使用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代码的移植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

php-amqplib3.0 / php-amqplib

维护者

详情