masstransit - 发送给 MassTransit 消费者的重复消息

Question

我有一个 MassTransit 示例程序，它发布 1 条正文为当前时间戳的消息。然而，奇怪的是，消费者总是有 5 次被调用。

程序如下:

public class MassTransitTest
{
    static void Main(string[] args)
    {
        var bus = ServiceBusFactory.New(x =>
        {
            x.UseRabbitMq();
            x.ReceiveFrom("rabbitmq://localhost/test");
            x.Subscribe(s => s.Consumer(() => new MyConsumer()));
        });

        bus.Publish<IMyObject>(new MyObject { Timestamp = DateTime.Now.ToString("yyyyMMddHHmmssffff"), CorrelationId = Guid.NewGuid() });
    }
}

public interface IMyObject : CorrelatedBy<Guid>
{
    string Timestamp { get; }
}

public class MyObject : IMyObject
{
    public string Timestamp { get; set; }
    public Guid CorrelationId { get; set; }
}

public class MyConsumer : Consumes<IMyObject>.All, IBusService
{
    private IServiceBus bus;

    private static int count = 0; // to gauge the call to handler

    public void Consume(IMyObject message)
    {
        count++;
        Console.WriteLine("Encounter message " + count);
        Console.WriteLine(message.Timestamp);
    }

    public void Start(IServiceBus bus)
    {
        this.bus = bus;
    }

    public void Stop()
    {            
    }

    public void Dispose()
    {
        bus.Dispose();
    }
}

输出如下:

Encounter message 1
201410131349034661
Encounter message 2
201410131349034661
Encounter message 3
201410131349034661
Encounter message 4
201410131349034661
Encounter message 5
201410131349034661

Answer 1

您不应在 Consumer 中处置总线，也不应成为 IBusService。 .Dispose 方法可能抛出异常，导致消费者重试。

您还应该在程序退出之前等待，并在完成处理后调用 bus.Dispose。

IBusService start 没有调用，因为消费者不是总线服务，所以总线成员始终为空。