RabbitMQ 和重复数据删除插件的问题答案

【问题标题】：Problem with RabbitMQ and deduplication pluginRabbitMQ 和重复数据删除插件的问题
【发布时间】：2019-10-05 15:30:01
【问题描述】：

我是 RabbitMQ 新手；对于一个新项目，我需要使用重复数据删除插件。我正在使用 AspNet Core 3.0 工作进程，语言是 C#。

我尝试了一个非常简单的示例，两个发布者发送 10 条编号为 1 到 10 的消息，一个消费者收到消息并确认它们。

我得到了非常奇怪和不可预测的结果：

如果我在同一个进程中运行 3 个工作人员（2 个发布者和一个消费者），看起来重复数据删除插件工作正常，并且只在队列中插入 10 条唯一消息，但消费者只读取前 2 条消息并且只确认其中之一。

如果我在两个不同的进程中运行发布者和消费者，消费者会收到所有 10 条消息，但在确认后消息仍保留在队列中，如果我再次运行消费者进程，它们会再次被重新处理。

我尝试用谷歌搜索 C# 中用于重复数据删除的完整工作示例，但没有成功

出版商

        int cnt = 1;
        var factory = new ConnectionFactory() { HostName = "localhost" };
        using (var connection = factory.CreateConnection())
        using (var channel = connection.CreateModel())
        {
            Dictionary<string, object> dd = new Dictionary<string, object>();
            dd["x-message-deduplication"] = true;
            channel.QueueDeclare(queue: qname,
                                 durable: true,
                                 exclusive: false,
                                 autoDelete: false,
                                 arguments: dd);

            while (!stoppingToken.IsCancellationRequested)
            {
                var message = GetMessage(cnt);
                var body = Encoding.UTF8.GetBytes(message);
                var properties = channel.CreateBasicProperties();
                properties.Persistent = true;
                Dictionary<string, object> d = new Dictionary<string, object>();
                d["x-deduplication-header"] = cnt;
                properties.Headers = d;

                channel.BasicPublish(exchange: "",
                                     routingKey: qname,
                                     basicProperties: properties,
                                     body: body);
                Console.WriteLine(" [x] Sent {0}", message);

                logDB(cnt, "Sender"+Wname);
                cnt++;
                if (cnt > 10)
                    break;
                await Task.Delay(1000, stoppingToken);
            }

消费者：

        while (!stoppingToken.IsCancellationRequested)
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                Dictionary<string, object> dd = new Dictionary<string, object>();
                dd["x-message-deduplication"] = true;
                channel.QueueDeclare(queue: qname,
                                     durable: true,
                                     exclusive: false,
                                     autoDelete: false,
                                     arguments: dd);

                _logger.LogInformation("{0} Waiting for messages.", Cname);

                var consumer = new EventingBasicConsumer(channel);
                consumer.Received += (model, ea) =>
                {
                    var body = ea.Body;
                    var message = Encoding.UTF8.GetString(body);
                    _logger.LogInformation("{0} Received {1}", Cname, message);


                    string[] parts = message.Split('-');
                    int cntmsg = int.Parse(parts[1]);

                    logDB(cntmsg, Cname);

                    Thread.Sleep((cntmsg % 5) * 1000);


                    _logger.LogInformation("{0} Received {1} done", Cname, message);

                    channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: true);
                };
                channel.BasicConsume(queue: qname,
                                     autoAck: false,
                                     consumer: consumer);

                _logger.LogInformation("{0} After BasicConsume", Cname);

                while (true)
                    await Task.Delay(1000, stoppingToken);

            }

【问题讨论】：

我对这个插件如何处理requeued 消息感到困惑。我更喜欢 Exchange level deduplication 到 Queue level deduplication。
感谢您的评论，我将尝试使用交换...您有完整的工作示例我可以开始吗？

标签： c# asp.net-core rabbitmq

【解决方案1】：

在联系重复数据删除插件的开发人员后，发现问题与重复数据删除标头的类型 (int) 有关，使用字符串值有效。

他将很快发布支持 int 数据类型的新版本。

【讨论】：