MassTransit 故障处理配置

Question

我在我的应用程序中配置故障使用者时遇到问题。问题是消费的消息被传递到一个*_error_skipped队列，它并没有完全消失。

下面是一个非常简单的例子。客户端应用程序收到失败的消息，它从test_error队列中消失，但它仍然存在于test_error_skipped队列中。

服务项目

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using GreenPipes;
using MassTransit;
using MassTransit.Util;

namespace MassTransitTest.Service
{

    public class RequestModel
    {
        public DateTime RequestTime { get; set; }
    }

    class MassTransitService : IDisposable
    {
        private readonly IBusControl _busControl;
        public MassTransitService()
        {
            _busControl = Bus.Factory.CreateUsingRabbitMq(configure =>
            {
                var host = configure.Host(new Uri("rabbitmq://localhost/mt_test"), h =>
                {
                    h.Username("guest");
                    h.Password("guest");
                });
                configure.ReceiveEndpoint(host, "test", c =>
                {
                    c.UseRetry(r => r.None());
                    c.Consumer<RequestConsumer>();
                });
            });
            TaskUtil.Await(_busControl.StartAsync());
            Console.WriteLine("bus started");
        }

        public void Dispose()
        {
            _busControl?.StopAsync().Wait();
        }
    }

    class RequestConsumer : IConsumer<RequestModel>
    {
        public Task Consume(ConsumeContext<RequestModel> context)
        {
            Console.WriteLine($"request with message id {context.MessageId} received: {context.Message.RequestTime}");
            throw new NotImplementedException();
        }
    }
}

客户项目

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using GreenPipes;
using MassTransit;
using MassTransit.Util;
using MassTransitTest.Service;

namespace MassTransitTest.Client
{
    class MassTransitClient
    {
        private readonly IBusControl _busControl;

        public MassTransitClient()
        {
            _busControl = Bus.Factory.CreateUsingRabbitMq(configure =>
            {
                var host = configure.Host(new Uri("rabbitmq://localhost/mt_test"), h =>
                {
                    h.Username("guest");
                    h.Password("guest");
                }); enter code here
                configure.ReceiveEndpoint(host, "test_error", c =>
                {
                    c.Consumer<ErrorConsumer>();
                });
            });
            TaskUtil.Await(_busControl.StartAsync());
            Console.WriteLine("bus started");
        }

        public async Task Send()
        {
            Console.WriteLine("sending request");

            await (await _busControl.GetSendEndpoint(new Uri("rabbitmq://localhost/mt_test/test"))).Send(new RequestModel()
            {
                RequestTime = DateTime.Now
            });
            Console.WriteLine("request sent");

        }
    }
    class ErrorConsumer : IConsumer<Fault<RequestModel>>
    {
        public Task Consume(ConsumeContext<Fault<RequestModel>> context)
        {
            Console.WriteLine($"request with message id {context.Message.FaultedMessageId} failed. requested time: {context.Message.Message.RequestTime}");
            return Task.CompletedTask;
        }
    }
}

我正在使用 .net core 2.1.1 和 MassTransit 5.1.3

Answer 1

要回答您的问题，该问题有几个部分，首先：

_error当消费者抛出异常时，MassTransit 接收端点将消息移动到队列中。_error不建议在队列上创建接收端点，也不应该这样做。

如果您只是想观察消费者是否发生故障，您可以创建一个单独的接收端点（例如fault-queue）并注册您的Fault<T>消费者。MassTransit 将发布实现 的消息，Fault<T>代理将通过接收端点将其路由到您的消费者。

但是，根据上面的示例，您正在发送请求并希望客户端知道是否发生了故障。为此，我建议使用请求客户端 - 它设置消息标头以将错误返回给请求发起者。它还允许发送响应。如果您不想等待响应，或者等待查看是否发生故障，上述故障观察器是您的最佳选择。

您可以在文档中看到如何使用请求客户端。