README

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

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

使用predis而不是Redis扩展
为QueueAttributes和Message提供了一些OO封装器
提供了一个简单的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";
}

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

bongrun / rsmq

维护者

详细信息