# 模块二 基础巩固 RabbitMQ Masstransit 详解
## 2.6.7 RabbitMQ -- Masstransit 详解
* Consumer 消费者
* Producer 生产者
* Request-Response 请求-响应
### Consumer 消费者
在 MassTransit 中，一个消费者可以消费一种或多种消息
消费者的类型包括：普通消费者，saga，saga 状态机，路由活动（分布式追踪），处理器 handlers，工作消费者 job comsumers
* Consumer
* Instance
* Handler
* Others
#### Consumer
```
public class Program
{
public static async Task Main()
{
var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
cfg.ReceiveEndpoint("order-service", e =>
{
e.Consumer();
});
});
}
}
```
继承 IConsumer，实现 Consume 方法
```
class SubmitOrderConsumer :
IConsumer
{
public async Task Consume(ConsumeContext context)
{
await context.Publish(new
{
context.Message.OrderId
});
}
}
```
三个原则：
* 拥抱 The Hollywood Principle, which states, "Dont't call us, we'll call you."
* Consume 方法是一个被等待的方法，在执行中时其他消费者无法接收到这个消息，当这个方法完成的时候，消息被 ack，并且从队列中移除
* Task 方法异常会导致消息触发 retry，如果没有配置重试，消息将被投递到失败队列
#### Instance
```
public class Program
{
public static async Task Main()
{
var submitOrderConsumer = new SubmitOrderConsumer();
var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
cfg.ReceiveEndpoint("order-service", e =>
{
e.Instance(submitOrderConsumer);
});
});
}
}
```
所有接收到的消息都由一个消费者来实例来处理（请确保这个消费者类是线程安全）
Consumer 每次接收到消息都会 new 一个实例
#### Handler
```
public class Program
{
public static async Task Main()
{
var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
cfg.ReceiveEndpoint("order-service", e =>
{
e.Handler(async context =>
{
await Console.Out.WriteLineAsync($"Submit Order Received: {context.Message.OrderId}");
});
});
});
}
}
```
通过一个委托 Lambda 方法，来消费消息
#### Others
* Saga<>
* StateMachineSaga<>
### Producer 生产者
消息的生产可以通过两种方式产生：发送和发布
发送的时候需要指定一个具体的地址 DestinationAddress，发布的时候消息会被广播给所有订阅了这个消息类型的消费者
基于这两种规则，消息被定义为：命令 command 和事件 event
* send
* publish
#### send
可以调用以下对象的 send 方法来发送 command：
* ConsumeContext （在 Consumer 的 Consumer 方法参数中传递）
* ISendEndpointProvider（可以从 DI 中获取）
* IBusControl（最顶层的控制对象，用来启动和停止 masstransit 的控制器）
**ConsumeContext**
```
public class SubmitOrderConsumer :
IConsumer
{
private readonly IOrderSubmitter _orderSubmitter;
public SubmitOrderConsumer(IOrderSubmitter submitter)
=> _orderSubmitter = submitter;
public async Task Consume(IConsumeContext context)
{
await _orderSubmitter.Process(context.Message);
await context.Send(new StartDelivery(context.Message.OrderId, DateTime.UtcNow));
}
}
```
**ISendEndpointProvider**
```
public async Task SendOrder(ISendEndpointProvider sendEndpointProvider)
{
var endpoint = await sendEndpointProvider.GetSendEndpoint(_serviceAddress);
await endpoint.Send(new SubmitOrder { OrderId = "123" });
}
```
#### publish
* 发送地址
* 短地址
* Convention Map
**发送地址**
* rabbitmq://localhost/input-queue
* rabbitmq://localhost/input-queue?durable=false
**短地址**
* GetSendEndpoint(new Uri("queue:input-queue"))
**Convention Map**
在配置文件中指定 map 规则
```
EndpointConvention.Map(new Uri(ConfigurationManager.AppSettings["deliveryServiceQueue"]));
```
直接发送
```
public class SubmitOrderConsumer :
IConsumer
{
private readonly IOrderSubmitter _orderSubmitter;
public SubmitOrderConsumer(IOrderSubmitter submitter)
=> _orderSubmitter = submitter;
public async Task Consume(IConsumeContext context)
{
await _orderSubmitter.Process(context.Message);
await context.Send(new StartDelivery(context.Message.OrderId, DateTime.UtcNow));
}
}
```
可以调用以下对象的 publish 方法来发送 event：
* ConsumeContext （在 Consumer 的 Consumer 方法参数中传递）
* IPublishEndpoint（可以从 DI 中获取）
* IBusControl（最顶层的控制对象，用来启动和停止 masstransit 的控制器）
IPublishEndpoint
```
public async Task NotifyOrderSubmitted(IPublishEndpoint publishEndpoint)
{
await publishEndpoint.Publish(new
{
OrderId = "27",
OrderDate = DateTime.UtcNow,
});
}
```
### Request-Response 请求-响应
Request-Response 模式让应用程序之间解耦之后，依然采用同步的方式
* Consumer
* IClientFactory
* IRequestClient
* Send a request
#### Consumer
```
public async Task Consume(ConsumeContext context)
{
var order = await _orderRepository.Get(context.Message.OrderId);
if (order == null)
throw new InvalidOperationException("Order not found");
await context.RespondAsync(new
{
OrderId = order.Id,
order.Timestamp,
order.StatusCode,
order.StatusText
});
}
```
需要处理返回类型 OrderStatusResult，异步方式模拟同步，实际上同样有消息队列，消费者处理过程
#### IClientFactory
```
public interface IClientFactory
{
IRequestClient CreateRequestClient(ConsumeContext context, Uri destinationAddress, RequestTimeout timeout);
IRequestClient CreateRequestClient(Uri destinationAddress, RequestTimeout timeout);
RequestHandle CreateRequest(T request, Uri destinationAddress, CancellationToken cancellationToken, RequestTimeout timeout);
RequestHandle CreateRequest(ConsumeContext context, T request, Uri destinationAddress, CancellationToken cancellationToken, RequestTimeout timeout);
}
```
通过 IBusControl 的 CreateClientFactory 方法可以得到 ClientFactory
#### IRequestClient
```
public interface IRequestClient
where TRequest : class
{
RequestHandle Create(TRequest request, CancellationToken cancellationToken, RequestTimeout timeout);
Task> GetResponse(TRequest request, CancellationToken cancellationToken, RequestTimeout timeout);
}
```
RequestClient 可以创建请求，或者直接获得响应
#### Send a request
```
var serviceAddress = new Uri("rabbitmq://localhost/check-order-status");
var client = bus.CreateRequestClient(serviceAddress);
var response = await client.GetResponse(new { OrderId = id});
```