README

这个库是 纯 PHP 实现的 AMQP 0-9-1 协议。已经过 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 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

基于信号的 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文件内容。归功于原始作者。

实现高级消息队列协议（AMQP）的PHP库。

该库是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

triwinvendor / amqp

维护者

详细信息