在 RabbitMQ 站点的 RPC 教程中,有一种方法可以传递“相关 ID”,该 ID 可以向队列中的用户标识您的消息。
我建议使用某种ID将消息放入前3个队列中,然后使用另一个进程将消息从3队列排入某种存储桶。当这些存储桶收到我假设的3个任务的完成时,将最终消息发送到第4个队列进行处理。
如果要为一个用户向每个队列发送多个工作项,则可能需要执行一些预处理,以找出特定用户放入队列中的项数,以便 4 之前排队的进程知道在排队之前需要多少个。
我在C#中做了我的rabbymq,所以很抱歉我的伪代码不是php样式
// Client
byte[] body = new byte[size];
body[0] = uniqueUserId;
body[1] = howManyWorkItems;
body[2] = command;
// Setup your body here
Queue(body)
// Server
// Process queue 1, 2, 3
Dequeue(message)
switch(message.body[2])
{
// process however you see fit
}
processedMessages[message.body[0]]++;
if(processedMessages[message.body[0]] == message.body[1])
{
// Send to queue 4
Queue(newMessage)
}
对更新 #1 的响应
不要将客户端视为终端,而是将客户端视为服务器上的进程可能会很有用。因此,如果您在这样的服务器上设置 RPC 客户端,那么您需要做的就是让服务器处理用户的唯一 ID 的生成,并将消息发送到相应的队列:
public function call($uniqueUserId, $workItem) {
$this->response = null;
$this->corr_id = uniqid();
$msg = new AMQPMessage(
serialize(array($uniqueUserId, $workItem)),
array('correlation_id' => $this->corr_id,
'reply_to' => $this->callback_queue)
);
$this->channel->basic_publish($msg, '', 'rpc_queue');
while(!$this->response) {
$this->channel->wait();
}
// We assume that in the response we will get our id back
return deserialize($this->response);
}
$rpc = new Rpc();
// Get unique user information and work items here
// Pass even more information in here, like what queue to use or you could even loop over this to send all the work items to the queues they need.
$response = rpc->call($uniqueUserId, $workItem);
$responseBuckets[array[0]]++;
// Just like above code that sees if a bucket is full or not