MassTransit 响应进入同一队列并导致循环

Question

我们目前正在将NServiceBus 换成MassTransit，我在请求/响应模式方面遇到了一点困难。

在NServiceBus 中，我可以在Handler 中回复，它会返回给发送它的客户端。

在MassTransit 中，似乎响应正在被发送回使用它的队列，从而创建了一个循环...

奇怪的是，如果我使用 InMemory 创建总线，并且客户端和消费者都在同一台机器上，我没有问题。

我希望我的客户能够收到响应，但我的 Consumer 却收到了它，这也很奇怪，因为它没有设置为接收该消息类型...

我是否在客户端的请求设置中遗漏了什么？

客户：

....
IRequestClient<IWorklistRequest, IWorklistResponse> client = CreateRequestClient(busControl, WorklistEndpointUri);

Console.Write("Sending Request");

Task.Run(async () =>
{
    IWorklistRequest request = new WorklistRequest
    {
        CurrentDateFrom = new DateTime(2016, 11, 07)
    };

    var response = await client.Request(request);

    Console.WriteLine("Worklist Items retrieved: {0}", response.ExamItemList.Length);

}).Wait();
....



static IRequestClient<IWorklistRequest, IWorklistResponse> CreateRequestClient(IBusControl busControl, string endpointAddress)
{
    Console.WriteLine("Creating Request client...");

    var serviceAddress = new Uri(endpointAddress);
    IRequestClient<IWorklistRequest, IWorklistResponse> client =
    busControl.CreateRequestClient<IWorklistRequest, IWorklistResponse>(serviceAddress, TimeSpan.FromSeconds(10));

    return client;
}

消费者：

    public Task Consume(ConsumeContext<IWorklistRequest> context)
    {
        _log.InfoFormat("Received Worklist Request with Id: {0}", context.RequestId);


        try
        {
            var result = _provider.GetAllWorklistsByStartDate(context.Message.CurrentDateFrom);

            IWorklistResponse response = new WorklistResponse
            {
                ExamItemList = result.ToArray()
            };

            // the below is sending the response right back to the original queue and is getting picked up again by this same consumer
            context.Respond(response);
        }
        catch (Exception ex)
        {
            _log.Info(ex.Message);
        }

        return Task.FromResult(0);
    }

Answer 1

如果您使用的是 RabbitMQ，并且您正在使用请求客户端，您应该不会看到这种行为。

在 MassTransit GitHub 存储库中有一个示例演示了如何使用请求客户端：https://github.com/MassTransit/Sample-RequestResponse

上面的代码看起来是正确的，Respond() 调用应该使用请求消息中的响应地址，这是一个直接端点发送到临时总线地址。

在这个领域有相当广泛的单元测试覆盖范围，上面的示例已使用最新版本的 MassTransit 进行了更新和验证。您可以考虑删除/重新创建您的 RabbitMQ 虚拟主机并从头开始运行您的应用程序（首先启动响应服务，以便设置端点）。