AMQP RabbitMQ消费者相互封锁？

Question

我编写了一个C（Rabbitmq-c）工作应用程序，它使用由Python脚本（pika）发布的队列。

我有以下奇怪的行为，我似乎无法解决：

1. 在消息发布到队列之前启动所有工作人员按预期工作
2. 队列发布后启动1个工作人员按预期工作
3. 然而：在一个工作人员开始从队列中消费后启动其他工作人员意味着这些工作人员在队列中看不到任何消息（消息计数=0），因此只是等待（即使队列中还有许多消息）。杀死第一个工作人员会突然开始消息流向所有其他（等待）消费者。

知道会发生什么吗？

我已经尝试确保每个消费者都有自己的渠道（这有必要吗？）但还是一样的行为。。。

以下是消费者（工作者）的代码：

conn = amqp_new_connection();
sock = (amqp_socket_t *)(uint64_t)amqp_tcp_socket_new(conn);
amqp_socket_open(sock, "localhost", 5672);
amqp_login(conn,
           "/",
           0,
           131072,
           0,
           AMQP_SASL_METHOD_PLAIN,
           "guest",
           "guest");

if (amqp_channel_open(conn, chan) == NULL)
    LOG_ERR(" [!] Failed to open amqp channel!\n");

if ((q = amqp_queue_declare(conn,
                            chan,
                            amqp_cstring_bytes("ranges"),
                            0,
                            0,
                            0,
                            0,
                            amqp_empty_table)) == NULL)
    LOG_ERR(" [!] Failed to declare queue!\n");

LOG_INFO(" [x] Queue (message count = %d)\n", q->message_count);

amqp_queue_bind(conn, chan, amqp_cstring_bytes("ranges"), amqp_empty_bytes, amqp_empty_table);
amqp_basic_consume(conn, chan, amqp_cstring_bytes("ranges"), amqp_empty_bytes, 0, 0, 0, amqp_empty_table);

while(1) {
    amqp_maybe_release_buffers(conn);
    amqp_consume_message(conn, &e, NULL, 0);

    {
        int n;
        amqp_frame_t f;
        unsigned char buf[8];
        unsigned char *pbuf = buf;

        amqp_simple_wait_frame(conn, &f);       // METHOD frame
        amqp_simple_wait_frame(conn, &f);       // HEADER frame

        n = f.payload.properties.body_size;
        if (n != sizeof(range_buf))
            LOG_ERR(" [!] Invalid message size!");

        while (n) {
            amqp_simple_wait_frame(conn, &f);   // BODY frame
            memcpy(pbuf,
                   f.payload.body_fragment.bytes,
                   f.payload.body_fragment.len);
            n -= f.payload.body_fragment.len;
            pbuf += f.payload.body_fragment.len;
        }

        // do something with buf

        LOG_INFO(" [x] Message recevied from queue\n");
    }

    amqp_destroy_envelope(&e);

    amqp_maybe_release_buffers(conn);
}

Answer 1

这可能对你有帮助

消息确认

完成一项任务可能需要几秒钟的时间。您可能想知道，如果其中一个消费者开始了一项长期任务，但只完成了部分任务就去世了，会发生什么情况。使用我们当前的代码，一旦RabbitMQ向客户发送消息，它就会立即将其从内存中删除。在这种情况下，如果您杀死一个工人，我们将丢失它刚刚处理的消息。我们还将丢失发送给此特定工作进程但尚未处理的所有消息。

但我们不想失去任何任务。如果一名工人死亡，我们希望将任务交付给另一名工人。

为了确保消息不会丢失，RabbitMQ支持消息确认。消费者会发回一个确认（knowledgement），告诉RabbitMQ已经收到并处理了一条特定的消息，RabbitMQ可以自由删除它。

如果消费者死亡而未发送ack，RabbitMQ将理解消息未被完全处理，并将其重新传递给另一个消费者。这样，即使工人偶尔死亡，也可以确保不会丢失任何信息。

没有任何消息超时；RabbitMQ将仅在工作连接断开时重新传递消息。即使处理一条消息需要非常非常长的时间，也没关系。

默认情况下打开消息确认。

Answer 2

这里的问题很可能是您的消费者在启动时预取了所有消息。这是RabbitMQ的默认行为，但您可以减少消费者预取的消息数量，以便更好地将工作负载分配给多个工作人员。

这仅仅意味着一个或多个消费者将接收所有消息，而不会为新消费者留下任何消息。

如果将qos应用于消费者，并将预取限制为10条消息。消费者只会排队等待前10条消息，而新消费者可以填补空缺。

您正在寻找的实现此功能称为amqp\u basic\u qos，此外，您可以在此处阅读有关消费者预取的更多信息。