译MassTransit 创建消息消费者

2023-11-08 12:48:34

创建消息消费者
一个消息消费者是一个可以消费一个或多个消息类型的类，指定IConsumer<T>接口，T为消息类型

public class UpdateCustomerConsumer :
    IConsumer<UpdateCustomerAddress>
{
    public async Task Consume(ConsumeContext<UpdateCustomerAddress> context)
    {
        await Console.Out.WriteLineAsync($"Updating customer: {context.Message.CustomerId}");

        // update the customer address
    }
}

当消费者订阅接收端点时，由端点接收消费者所消费的消息。创建一个消费者实例（using a consumer factory, which is covered（覆盖） --> here <--）然后，通过Consume方法将消息（包裹在 ConsumeContext）传递给消费者。

Consume方法是异步的，并返回一个task。。MassTransit 等待该任务, 在此期间消息对其他接收端点不可用。如果consume 方法成功完成 (RanToCompletion 的task状态),则消息将被确认并从队列中删除。

注意：如果消费者错误（例如抛出异常，导致Faulted的任务状态），或者以某种方式被取消cancelled （被取消的Canceled任务状态），则异常被传播回管道，在那里它可以最终被重试或移动到错误队列。

连接消息消费者

对于消费者接收消息，消费者必须连接到接收端点。这是在总线配置期间进行的，尤其是在接收端点的配置中。

下面显示了将消费者连接到接收端点的示例。

var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    var host = cfg.Host(new Uri("rabbitmq://localhost/"), h =>
    {
        h.Username("guest");
        h.Password("guest");
    });

    cfg.ReceiveEndpoint(host, "customer_update_queue", e =>
    {
        e.Consumer<UpdateCustomerConsumer>();
    });
});

该示例创建一个总线，该总线使用使用默认的用户名和密码(Guest/Guest)连接到在本地机器上运行的RabbitMQ，.在那个总线上.使用命名customer_update_queue创建单个接收端点。使用最简单的方法连接消费者，该方法接受具有默认构造函数的消费者类。

注意：当一个消费者连接到一个接收端点时，由连接到同一个接收端点的所有消费者所消费的组合消息类型被*subscribed*订阅到队列中。(in the case of RabbitMQ exchange bindings are created for the message types to the exchange/queue for the receive endpoint.)订阅方法因broker代理而异，在RabbitMQ Exchange绑定的情况下，将消息类型创建为接收端点的Exchange /queue。这些订阅是持久的，并在进程退出后保持不变。这样可以确保发布或发送的消息交付到接收端点消费者之一，即使进程终止。当进程启动时，队列中等待的消息将交付给消费者。

上面的示例使用默认构造函数消费者工厂来连接消费者。有几个其他的消费工厂支持，如下所示。

var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    var host = ...;

    cfg.ReceiveEndpoint(host, "customer_update_queue", e =>
    {
        // an anonymous factory method
        e.Consumer(() => new YourConsumer());

        // an existing consumer factory for the consumer type
        e.Consumer(consumerFactory);

        // a type-based factory that returns an object (container friendly)
        e.Consumer(consumerType, type => container.Resolve(type));

        // an anonymous factory method, with some middleware goodness
        e.Consumer(() => new YourConsumer(), x =>
        {
            // add middleware to the consumer pipeline
            x.UseLog(ConsoleOut, async context => "Consumer created");
        });
    });
});

连接到现有总线
一旦创建了总线，就会创建接收端点，无法修改。然而，总线本身提供了一个临时的（自动删除）队列，可以用来接收消息。为了将消费者连接到总线临时队列，可以使用一系列连接方法。

警告：临时队列将不接收已发布的消息。由于队列是临时的，当消费者连接时，不会创建绑定或订阅。这使得它对于临时消费者非常快，并且避免用临时绑定来击乱消息代理。

临时队列对于接收请求响应和故障（via the response/fault address header）和路由滑移事件（via an event subscription in the routing slip）非常有用。

var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    var host = ...;
});

busControl.Start();

ConnectHandle handle = busControl.ConnectConsumer<FaultConsumer>();
...
handle.Disconnect(); // disconnect the consumer from the bus pipeline

除了 ConnectConsumer 方法之外, 还包括每个消费者类型的方法 (ConnectHandler、ConnectInstance、ConnectSaga 和 ConnectStateMachineSaga)。

连接现有的消费者实例
虽然强烈建议使用每个消息的消费者实例，但可以连接一个现有的消费者实例，每个实例都将调用该实例。消费者必须是线程安全的，因为消费方法将同时从多个线程调用。若要连接现有实例，请参见下面的示例。

var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    var host = ...;

    cfg.ReceiveEndpoint(host, "customer_update_queue", e =>
    {
        e.Instance(existingConsumer);
    });
});

处理无法传递的消息

如果端点的配置改变，或者如果消息被错误地发送到端点，则可能接收到没有任何连接的消费者的消息类型。如果发生这种情况，则将消息移动到_skipped队列（由原始队列名称前缀）。保留原始消息内容，并添加附加标题来指示移动消息的主机。

处理没有消费者的消息

虽然创建消费者是首选的消息消费方式，但也有可能创建一个简单的消息处理程序。通过指定方法、匿名方法或lambda方法，消息可以在接收端点上被消耗。若配置简单的消息处理程序, 请参阅下面的示例。

var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    var host = ...;

    cfg.ReceiveEndpoint(host, "customer_update_queue", e =>
    {
        e.Handler<UpdateCustomerAddress>(context =>
            return Console.Out.WriteLineAsync($"Update customer address received: {context.Message.CustomerId}"));
    });
});

在这种情况下，对接收到的每个消息都调用该方法。没有创建消费者，也不执行生命周期管理。

通过 IObserver 观察消息
通过添加IObserver 接口，将观察者的概念添加到.NET Framework中。MassTransit 支持观察者直接连接接收端点。

可惜，观察者不是异步的。因此，当使用观测者时，无法对编译程序提供的异步支持进行良好的运行。

一个观察者使用内置的IObserver<T>接扣定义，如下所示。

public class CustomerAddressUpdatedObserver :
    IObserver<ConsumeContext<CustomerAddressUpdated>>
{
    public void OnNext(ConsumeContext<CustomerAddressUpdated> context)
    {
        Console.WriteLine("Customer address was updated: {0}", context.Message.CustomerId);
    }

    public void OnError(Exception error)
    {
    }

    public void OnCompleted()
    {
    }
}

一旦创建，观察者连接到接收端点，类似于消费者。

var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    var host = ...;

    cfg.ReceiveEndpoint(host, "customer_update_queue", e =>
    {
        e.Observer<CustomerAddressUpdatedObserver>();
    });
});

码农公寓

相关文章