创建消息消费者
一个消息消费者是一个 可以消费一个或多个消息类型的类，指定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>(); }); });
