使用Newlife.RocketMQ(C#)遇消息重复与多实例消费问题求助

我正在用C#结合Newlife.RocketMQ测试RocketMQ，本地RocketMQ服务已正常启动，编写了如下WPF测试程序：

WPF窗口代码 #

<Window x:Class="WpfApp1.MainWindow"
    xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
    xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
    xmlns:d="http://schemas.microsoft.com/expression/blend/2008"
    xmlns:mc="http://schemas.openxmlformats.org/markup-compatibility/2006"
    xmlns:local="clr-namespace:WpfApp1"
    mc:Ignorable="d"
    Title="MainWindow" Height="450" Width="800" >
<DockPanel >
    <TextBox x:Name="txtProduce" Text="message 1" DockPanel.Dock="Top" Height="30" FontSize="16"/>
    <Button x:Name="btnProduce" Content="Produce" HorizontalAlignment="Left" VerticalAlignment="Top" Width="75" Click="BtnProduce_Click" DockPanel.Dock="Top"/>
    <DataGrid x:Name="dgProduce" Height="100" DockPanel.Dock="Top" IsReadOnly="True"/>
    <Label Content="Consumer" DockPanel.Dock="Top"/>
    <DataGrid x:Name="dgConsume" IsReadOnly="True"/>
</DockPanel>
</Window>

C#逻辑代码 #

public partial class MainWindow : Window
{
    Producer _producer;
    Consumer _consumer;
    ObservableCollection<SendResult> _results = new ObservableCollection<SendResult>();
    object _resultsLock = new object();
    ObservableCollection<MessageExt> _messages = new ObservableCollection<MessageExt>();
    object _messagesLock = new object();

    public MainWindow()
    {
        InitializeComponent();
        dgProduce.ItemsSource = _results;
        BindingOperations.EnableCollectionSynchronization(_results, _resultsLock);
        dgConsume.ItemsSource = _messages;
        BindingOperations.EnableCollectionSynchronization(_messages, _messagesLock);
        _producer = new Producer
        {
            Topic = "TopicTest",
            NameServerAddress = "localhost:9876",
            RetryTimesWhenSendFailed = 10
        };
        _producer.Start();
        _consumer = new Consumer
        {
            Topic = "TopicTest",
            NameServerAddress = "localhost:9876"
        };
        _consumer.OnConsume = (q, ms) =>
        {
            foreach(var item in ms.ToList())
                _messages.Add(item);
            return true;
        };
        _consumer.Start();
    }
    private int _i = 1;
    private void BtnProduce_Click(object sender, RoutedEventArgs e)
    {
        try
        {
            var content = DateTime.Now.ToString("yyyy/MM/dd HH:mm:ss.fff") + ":" + txtProduce.Text;
            var message = new Message()
            {
                Body = System.Text.Encoding.Default.GetBytes(content),
                Keys = (_i++).ToString(),
                Tags = _i % 2 == 0 ? "even" : "odd",
                Flag = 0,
                WaitStoreMsgOK = true
            };
            var sr = _producer.Publish(message);
            _results.Add(sr);
            txtProduce.Text = "message " + _i;
        }
        catch (Exception ex)
        {
            MessageBox.Show(ex.Message);
        }
    }
}

测试中遇到两个问题：

• 消费者消息列表中"message 3"重复出现；
• 启动两个应用实例时，发送的消息仅被其中一个实例消费，无法同时被两者接收。

问题1：消息重复出现的原因 #

1. RocketMQ重试机制：RocketMQ默认是「至少一次」投递策略，当Broker没收到消费端的确认响应（比如网络延迟、消费处理超时），就会重新投递这条消息。你的消费逻辑直接返回true，但如果Broker未及时收到确认信号，就会触发重复投递。
2. 未指定消费组ID：Newlife.RocketMQ未手动指定GroupId时会用默认值，若消费者重启后，默认消费组的消费位点未正确提交，重启后会从上次的位点重新消费，导致旧消息重复出现。
3. 代码逻辑小偏差：生成消息时，Keys = (_i++).ToString()后Tags又用了自增后的_i，导致message 3对应的标签逻辑和预期有偏差，虽不是重复直接原因，但可能干扰排查判断。

问题2：多实例无法同时消费的原因 #

RocketMQ默认是集群消费模式：同一个消费组（GroupId）下的多个消费者会分摊消息，一条消息只会被组内一个实例消费。你的两个应用实例都没指定GroupId，默认用同一个消费组，所以消息被负载均衡到其中一个实例。要让每个实例都接收所有消息，需将消费者设置为广播消费模式，或给每个实例指定不同的GroupId。

内容的提问来源于stack exchange，提问作者AIMIN PAN