README

这是一个轻量级的PHP消息队列，不需要专用队列服务器。只需一个Redis服务器。有关更多信息，请参阅smrchy/rsmq。

这是基于eislambey/php-rsmq的分支，以下是一些更改：

• 使用predis而不是Redis扩展
• 为QueueAttributes和Message提供一些OO包装器
• 提供一个简单的QueueWorker

目录

• 安装
• 方法
  • 构造
  • 队列
    • createQueue
    • listQueues
    • deleteQueue
    • getQueueAttributes
    • setQueueAttributes
  • 消息
    • sendMessage
    • receiveMessage
    • deleteMessage
    • popMessage
    • changeMessageVisibility
  • 实时
• QueueWorker
• 许可证

安装

composer require andrewbreksa/rsmq

方法

构造

RSMQ的新实例。

参数

• $predis (\Predis\ClientInterface): *必需 The Predis实例
• $ns (string): 可选 (默认: "rsmq") RSMQ创建的所有键使用的命名空间前缀
• $realtime (Boolean): 可选 (默认: false) 启用新消息的实时PUBLISH

示例

<?php
use Predis\Client;
use AndrewBreksa\RSMQ\RSMQClient;

$predis = new Client(
    [
        'host' => '127.0.0.1',
        'port' => 6379
    ]
);
$this->rsmq = new RSMQClient($predis);

队列

createQueue

创建一个新队列。

参数

• $name (string): 队列名称。最大160个字符；允许字母数字字符、连字符（-）和下划线（_）。
• $vt (int): 可选 (默认: 30) 从队列接收到的消息在请求接收消息时对其他接收组件不可见的时间长度，以秒为单位。允许的值：0-9999999（约115天）
• $delay (int): 可选 (默认: 0) 队列中所有新消息投递的延迟时间，以秒为单位。允许的值：0-9999999（约115天）
• $maxsize (int): 可选 (默认: 65536) 消息的最大大小（以字节为单位）。允许的值：1024-65536和-1（无限大小）

返回

• true (Bool)

抛出

• \AndrewBreksa\RSMQ\Exceptions\QueueAlreadyExistsException

示例

<?php
/**
 * @var AndrewBreksa\RSMQ\RSMQClientInterface $rsmq
 */

$rsmq->createQueue('myqueue');

listQueues

列出所有队列

返回一个数组

• ["qname1", "qname2"]

示例

<?php
/**
 * @var AndrewBreksa\RSMQ\RSMQClientInterface $rsmq
 */

$queues = $rsmq->listQueues();

deleteQueue

删除队列及其所有消息。

参数

• $name (string): 队列名称。

返回

• true (Bool)

抛出

• \AndrewBreksa\RSMQ\Exceptions\QueueNotFoundException

示例

<?php
/**
 * @var AndrewBreksa\RSMQ\RSMQClientInterface $rsmq
 */

$rsmq->deleteQueue('myqueue');

getQueueAttributes

获取队列属性、计数器和统计信息

参数

• $queue (string): 队列名称。

返回一个具有以下属性的\AndrewBreksa\RSMQ\QueueAttributes对象

• vt (int): 队列的可见性超时时间（以秒为单位）
• delay (int): 新消息的延迟时间（以秒为单位）
• maxSize (int): 消息的最大大小（以字节为单位）
• totalReceived (int): 从队列接收到的消息总数
• totalSent (int): 发送到队列的消息总数
• created (float): 队列创建的时间戳（以秒为单位的纪元）
• modified (float): 使用setQueueAttributes最后修改队列的时间戳（以秒为单位的纪元）
• messageCount (int): 队列中当前的消息数量
• hiddenMessageCount (int): 当前隐藏/不可见消息的数量。由于vt参数或使用delay发送而处于“在途中”的消息可以被隐藏

示例

<?php
/**
 * @var AndrewBreksa\RSMQ\RSMQClientInterface $rsmq
 */

$attributes =  $rsmq->getQueueAttributes('myqueue');
echo "visibility timeout: ", $attributes->getVt(), "\n";
echo "delay for new messages: ", $attributes->getDelay(), "\n";
echo "max size in bytes: ", $attributes->getMaxSize(), "\n";
echo "total received messages: ", $attributes->getTotalReceived(), "\n";
echo "total sent messages: ", $attributes->getTotalSent(), "\n";
echo "created: ", $attributes->getCreated(), "\n";
echo "last modified: ", $attributes->getModified(), "\n";
echo "current n of messages: ", $attributes->getMessageCount(), "\n";
echo "hidden messages: ", $attributes->getHiddenMessageCount(), "\n";

setQueueAttributes

设置队列参数。

参数

• $queue (string): 队列名称。
• $vt (int): 可选 * 从队列接收到的消息在请求接收消息时对其他接收组件不可见的时间长度，以秒为单位。允许的值：0-9999999（约115天）
• $delay (int): 可选 所有新消息在队列中发送的延迟时间（秒）。允许的值：0-9999999（约115天）
• $maxsize (int): 可选 消息的最大大小（字节）。允许的值：1024-65536和-1（表示无限制大小）

注意：至少必须提供一个属性（vt、delay、maxsize）。只有提供的属性才会被修改。

返回一个具有以下属性的\AndrewBreksa\RSMQ\QueueAttributes对象

• vt (int): 队列的可见性超时时间（以秒为单位）
• delay (int): 新消息的延迟时间（以秒为单位）
• maxSize (int): 消息的最大大小（以字节为单位）
• totalReceived (int): 从队列接收到的消息总数
• totalSent (int): 发送到队列的消息总数
• created (float): 队列创建的时间戳（以秒为单位的纪元）
• modified (float): 使用setQueueAttributes最后修改队列的时间戳（以秒为单位的纪元）
• messageCount (int): 队列中当前的消息数量
• hiddenMessageCount (int): 当前隐藏/不可见消息的数量。由于vt参数或使用delay发送而处于“在途中”的消息可以被隐藏

抛出

• \AndrewBreksa\RSMQ\QueueAttributes
• \AndrewBreksa\RSMQ\QueueParametersValidationException
• \AndrewBreksa\RSMQ\QueueNotFoundException

示例

<?php
/**
 * @var AndrewBreksa\RSMQ\RSMQClientInterface $rsmq
 */

$queue = 'myqueue';
$vt = 50;
$delay = 10;
$maxsize = 2048;
$rsmq->setQueueAttributes($queue, $vt, $delay, $maxsize);

消息

sendMessage

发送一条新消息。

参数

• $queue (string)
• $message (string)
• $delay (int): 可选 （默认：队列设置） 消息发送的延迟时间（秒）。允许的值：0-9999999（约115天）

返回

• $id (string): 内部消息ID。

抛出

• \AndrewBreksa\RSMQ\Exceptions\MessageToLongException
• \AndrewBreksa\RSMQ\Exceptions\QueueNotFoundException
• \AndrewBreksa\RSMQ\Exceptions\QueueParametersValidationException

示例

<?php
/**
 * @var AndrewBreksa\RSMQ\RSMQClientInterface $rsmq
 */

$id = $rsmq->sendMessage('myqueue', 'a message');
echo "Message Sent. ID: ", $id;

receiveMessage

从队列中接收下一条消息。

参数

• $queue (string): 队列名称。
• $vt (int): 可选 （默认：队列设置） 接收到的消息将不可见的时间长度，以秒为单位。允许的值：0-9999999（约115天）

返回一个具有以下属性的 \AndrewBreksa\RSMQ\Message 对象

• message (string): 消息内容。
• id (string): 内部消息ID。
• sent (int): 消息发送/创建的时间戳。
• firstReceived (int): 消息首次接收的时间戳。
• receiveCount (int): 此消息接收的次数。

注意：如果没有消息，将返回一个空数组

抛出

• \AndrewBreksa\RSMQ\Exceptions\QueueNotFoundException
• \AndrewBreksa\RSMQ\Exceptions\QueueParametersValidationException

示例

<?php
/**
 * @var AndrewBreksa\RSMQ\RSMQClientInterface $rsmq
 */

$message = $rsmq->receiveMessage('myqueue');
echo "Message ID: ", $message->getId();
echo "Message: ", $message->getMessage();

deleteMessage

参数

• $queue (string): 队列名称。
• $id (string): 要删除的消息ID。

返回

• true 如果成功，false 如果未找到消息（bool）。

抛出

• \AndrewBreksa\RSMQ\Exceptions\QueueParametersValidationException

示例

<?php
/**
 * @var AndrewBreksa\RSMQ\RSMQClientInterface $rsmq
 */

$id = $rsmq->sendMessage('queue', 'a message');
$rsmq->deleteMessage('queue', $id);

popMessage

从队列中接收下一条消息 并删除它。

重要：此方法会立即删除接收到的消息。如果在处理消息时出现问题，则无法再次接收该消息。

参数

• $queue (string): 队列名称。

返回一个具有以下属性的 \AndrewBreksa\RSMQ\Message 对象

• message (string): 消息内容。
• id (string): 内部消息ID。
• sent (int): 消息发送/创建的时间戳。
• firstReceived (int): 消息首次接收的时间戳。
• receiveCount (int): 此消息接收的次数。

注意：如果没有消息，将返回一个空对象

抛出

• \AndrewBreksa\RSMQ\Exceptions\QueueNotFoundException
• \AndrewBreksa\RSMQ\Exceptions\QueueParametersValidationException

示例

<?php
/**
 * @var AndrewBreksa\RSMQ\RSMQClientInterface $rsmq
 */

$message = $rsmq->popMessage('myqueue');
echo "Message ID: ", $message->getId();
echo "Message: ", $message->getMessage();

changeMessageVisibility

更改单个消息的可见性计时器。消息再次可见的时间是从当前时间（now）+ vt 计算得出的。

参数

• qname (string): 队列名称。
• id (string): 消息ID。
• vt (int): 此消息不可见的时长（秒）。允许的值：0-9999999（约115天）

返回

• true 如果成功，false 如果未找到消息（bool）。

抛出

• \AndrewBreksa\RSMQ\Exceptions\QueueParametersValidationException
• \AndrewBreksa\RSMQ\Exceptions\QueueNotFoundException

示例

<?php
/**
 * @var AndrewBreksa\RSMQ\RSMQClientInterface $rsmq
 */

$queue = 'myqueue';
$id = $rsmq->sendMessage($queue, 'a message');
if($rsmq->changeMessageVisibility($queue, $id, 60)) {
	echo "Message hidden for 60 secs";
}

实时

在创建 AndrewBreksa\RSMQ\RSMQClient 实例时，可以通过传递 \AndrewBreksa\RSMQ\RSMQClient::__construct 的 $realtime 参数为 true 来启用新消息的实时 PUBLISH。每当通过 sendMessage 发送新消息时，都会向 {rsmq.ns}:rt:{qname} 发出 Redis PUBLISH 命令。

使用默认设置的 RSMQ 示例

• 队列 testQueue 已包含 5 条消息。
• 一条新消息正在发送到队列 testQueue。
• 将发出以下 Redis 命令：PUBLISH rsmq:rt:testQueue 6

实时选项启用在新消息发送到 RSMQ 时发送 PUBLISH，但是没有在此功能上构建其他功能。您的应用程序可以使用 Redis SUBSCRIBE 命令来接收新消息的通知，然后尝试从队列中轮询，但是由于 Redis pub/sub 系统的工作方式，所有监听器都将被通知新消息，此方法不适用于驱动具有多个订阅进程的环境中的消息处理。

QueueWorker

QueueWorker 类提供了一个简单的方法来消费 RSMQ 消息，要使用它

<?php
/**
 * @var AndrewBreksa\RSMQ\RSMQClientInterface $rsmq
 */

use AndrewBreksa\RSMQ\ExecutorInterface;
use AndrewBreksa\RSMQ\Message;
use AndrewBreksa\RSMQ\QueueWorker;
use AndrewBreksa\RSMQ\WorkerSleepProvider;

$executor = new class() implements ExecutorInterface{
    public function __invoke(Message $message) : bool {
        //@todo: do some work, true will ack/delete the message, false will allow the queue's config to "re-publish"
        return true;
    }
};

$sleepProvider = new class() implements WorkerSleepProvider{
    public function getSleep() : ?int {
        /**
         * This allows you to return null to stop the worker, which can be used with something like redis to mark.
         *
         * Note that this method is called _before_ we poll for a message, and therefore if it returns null we'll eject
         * before we process a message.
         */
        return 1;
    }
};

$worker = new QueueWorker($rsmq, $executor, $sleepProvider, 'test_queue');
$worker->work(); // here we can optionally pass true to only process one message

许可证

MIT 许可证。查看 LICENSE