RabbitMQ C# 验证消息已发送

Question

我是 RabbitMQ 的新手，正在尝试写入队列并验证消息是否已发送。如果它失败了，我需要知道它。 我做了一个假队列来观察它失败，但无论我看到什么都没有异常，当我在寻找一个确认时，我总是得到一个。我从没见过 BasicNack。

我什至不确定我是否是 BasicAcks 是要走的路。

    private void button1_Click(object sender, EventArgs e)
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        using (var connection = factory.CreateConnection())
        {
            using (var channel = connection.CreateModel())
            {
                channel.QueueDeclare("task_queue", true, false, false, null);

                var message = ("Helllo world");
                var body = Encoding.UTF8.GetBytes(message);
                channel.ConfirmSelect();

                var properties = channel.CreateBasicProperties();
                properties.SetPersistent(true);
                properties.DeliveryMode = 2;
                channel.BasicAcks += channel_BasicAcks;
                channel.BasicNacks += channel_BasicNacks;
                //fake queue should be task_queue
                channel.BasicPublish("", "task_2queue", true, properties, body);

                channel.WaitForConfirmsOrDie();

                Console.WriteLine(" [x] Sent {0}", message);
            }
        }
    }

    void channel_BasicNacks(IModel model, BasicNackEventArgs args)
    {

    }

    void channel_BasicAcks(IModel model, BasicAckEventArgs args)
    {

    }

Answer 1

对于那些寻找 C# 答案的人 - 这就是您所需要的。

https://rianjs.net/2013/12/publisher-confirms-with-rabbitmq-and-c-sharp

类似这样的：（BasicAcks 附加了一个事件处理程序 - 还有 BasicNacks）

using (var connection = FACTORY.CreateConnection())
{
    var channel = connection.CreateModel();
    channel.ExchangeDeclare(QUEUE_NAME, ExchangeType.Fanout, true);
    channel.QueueDeclare(QUEUE_NAME, true, false, false, null);
    channel.QueueBind(QUEUE_NAME, QUEUE_NAME, String.Empty, new Dictionary<string, object>());
     channel.BasicAcks += (sender, eventArgs) =>
                {
                    //implement ack handle
                };
    channel.ConfirmSelect();

    for (var i = 1; i <= numberOfMessages; i++)
    {
        var messageProperties = channel.CreateBasicProperties();
        messageProperties.SetPersistent(true);

        var message = String.Format("{0}\thello world", i);
        var payload = Encoding.Unicode.GetBytes(message);
        Console.WriteLine("Sending message: " + message);
        channel.BasicPublish(QUEUE_NAME, QUEUE_NAME, messageProperties, payload);
        channel.WaitForConfirmsOrDie();
    }
}

Answer 2

你需要一个Publisher Confirms

如您所见，您可以实现：

交易：

ch.txSelect(); <-- start transaction
ch.basicPublish("", QUEUE_NAME,
                            MessageProperties.PERSISTENT_BASIC,
                            "nop".getBytes());
ch.txCommit();<--commit transaction

消息存储到队列和磁盘中。 这种方式可能会很慢，如果您需要性能，则不应使用它。

您可以使用 Streaming Lightweight Publisher Confirms，使用：

ch.setConfirmListener(new ConfirmListener() {
    public void handleAck(long seqNo, boolean multiple) {
        if (multiple) {
            unconfirmedSet.headSet(seqNo+1).clear();
        } else {
            unconfirmedSet.remove(seqNo);
        }
    }
    public void handleNack(long seqNo, boolean multiple) {
        // handle the lost messages somehow
    }

希望对你有帮助

Answer 3

好的，您总是会收到您发送的消息的 ACK，因为“每次消息都成功传递到默认 Exchange。”

PS：您不会直接将消息发送到队列，一旦 Exchange 收到它给您 ACK 的消息，它就会使用路由键（如果有）将消息路由到所有绑定队列。