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<SubmitOrderConsumer>();
});
});
}
}

继承 IConsumer,实现 Consume 方法

class SubmitOrderConsumer :
IConsumer<SubmitOrder>
{
public async Task Consume(ConsumeContext<SubmitOrder> context)
{
await context.Publish<OrderSubmitted>(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<SubmitOrder>(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<SubmitOrder>
{
private readonly IOrderSubmitter _orderSubmitter; public SubmitOrderConsumer(IOrderSubmitter submitter)
=> _orderSubmitter = submitter; public async Task Consume(IConsumeContext<SubmitOrder> 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<StartDelivery>(new Uri(ConfigurationManager.AppSettings["deliveryServiceQueue"]));

直接发送

public class SubmitOrderConsumer :
IConsumer<SubmitOrder>
{
private readonly IOrderSubmitter _orderSubmitter; public SubmitOrderConsumer(IOrderSubmitter submitter)
=> _orderSubmitter = submitter; public async Task Consume(IConsumeContext<SubmitOrder> 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<OrderSubmitted>(new
{
OrderId = "27",
OrderDate = DateTime.UtcNow,
});
}

Request-Response 请求-响应

Request-Response 模式让应用程序之间解耦之后,依然采用同步的方式

  • Consumer
  • IClientFactory
  • IRequestClient
  • Send a request

Consumer

public async Task Consume(ConsumeContext<CheckOrderStatus> context)
{
var order = await _orderRepository.Get(context.Message.OrderId);
if (order == null)
throw new InvalidOperationException("Order not found"); await context.RespondAsync<OrderStatusResult>(new
{
OrderId = order.Id,
order.Timestamp,
order.StatusCode,
order.StatusText
});
}

需要处理返回类型 OrderStatusResult,异步方式模拟同步,实际上同样有消息队列,消费者处理过程

IClientFactory

public interface IClientFactory
{
IRequestClient<T> CreateRequestClient<T>(ConsumeContext context, Uri destinationAddress, RequestTimeout timeout); IRequestClient<T> CreateRequestClient<T>(Uri destinationAddress, RequestTimeout timeout); RequestHandle<T> CreateRequest<T>(T request, Uri destinationAddress, CancellationToken cancellationToken, RequestTimeout timeout); RequestHandle<T> CreateRequest<T>(ConsumeContext context, T request, Uri destinationAddress, CancellationToken cancellationToken, RequestTimeout timeout);
}

通过 IBusControl 的 CreateClientFactory 方法可以得到 ClientFactory

IRequestClient

public interface IRequestClient<TRequest>
where TRequest : class
{
RequestHandle<TRequest> Create(TRequest request, CancellationToken cancellationToken, RequestTimeout timeout); Task<Response<T>> GetResponse<T>(TRequest request, CancellationToken cancellationToken, RequestTimeout timeout);
}

RequestClient 可以创建请求,或者直接获得响应

Send a request

var serviceAddress = new Uri("rabbitmq://localhost/check-order-status");
var client = bus.CreateRequestClient<CheckOrderStatus>(serviceAddress); var response = await client.GetResponse<OrderStatusResult>(new { OrderId = id});

本作品采用知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议进行许可。

欢迎转载、使用、重新发布,但务必保留文章署名 郑子铭 (包含链接: http://www.cnblogs.com/MingsonZheng/ ),不得用于商业目的,基于本文修改后的作品务必以相同的许可发布。

如有任何疑问,请与我联系 (MingsonZheng@outlook.com) 。

.NET 云原生架构师训练营(模块二 基础巩固 RabbitMQ Masstransit 详解)--学习笔记的更多相关文章

  1. .NET 云原生架构师训练营(模块二 基础巩固 RabbitMQ Masstransit 介绍)--学习笔记

    2.6.6 RabbitMQ -- Masstransit 介绍 Masstransit 是什么 Quickstart 消息 Message Masstransit 是什么 Masstransit 是 ...

  2. .NET 云原生架构师训练营(模块二 基础巩固 RabbitMQ Masstransit 异常处理)--学习笔记

    2.6.8 RabbitMQ -- Masstransit 异常处理 异常处理 其他 高级功能 异常处理 异常与重试 重试配置 重试条件 重新投递信息 信箱 异常与重试 Exception publi ...

  3. .NET 云原生架构师训练营(模块一 架构师与云原生)--学习笔记

    目录 什么是软件架构 软件架构的基本思路 单体向分布式演进.云原生.技术中台 1.1 什么是软件架构 1.1.1 什么是架构? Software architecture = {Elements, F ...

  4. .NET 云原生架构师训练营(模块二 基础巩固 敏捷开发)--学习笔记

    2.7.1 敏捷开发 敏捷介绍 敏捷的起源 敏捷软件开发宣言 敏捷开发十二原则 生命周期对比 敏捷开发的特点 敏捷的发展 敏捷的核心 敏捷的起源 2001年,17个老头子在一起一边滑雪,一边讨论工作, ...

  5. .NET 云原生架构师训练营(模块二 基础巩固 MongoDB 介绍和基础)--学习笔记

    2.5.1 MongoDB -- 介绍 mysql vs mongo 快速开始 mysql vs mongo 对比 mysql mongo 数据存储 table 二维表结构,需要预先定义结构 json ...

  6. .NET 云原生架构师训练营(模块二 基础巩固 MongoDB 问答系统)--学习笔记

    2.5.6 MongoDB -- 问答系统 MongoDB 数据库设计 API 实现概述 MongoDB 数据库设计 设计优化 内嵌(mongo)还是引用(mysql) 数据一致性 范式:将数据分散到 ...

  7. .NET 云原生架构师训练营(模块二 基础巩固 Scrum 团队)--学习笔记

    2.7.3 Scrum 团队 理想的环境 团队章程 如何组建 Scrum 团队 产品待办事项列表 用户故事 敏捷开发流程 理想的环境 5-9人 100% 跨职能 在一起 自组织 自组织 目标 授权 沟 ...

  8. .NET 云原生架构师训练营(模块二 基础巩固 EF Core 更新和迁移)--学习笔记

    2.4.6 EF Core -- 更新 状态 自动变更检测 不查询删除和更新 并发 状态 Entity State Property State Entity State Added 添加 Uncha ...

  9. .NET 云原生架构师训练营(模块二 基础巩固 MongoDB 写入和查询)--学习笔记

    2.5.3 MongoDB -- 写入和查询 写入 查询 查找操作符 逻辑操作符 其他 嵌套对象 数组 游标方法 写入 https://docs.mongodb.com/manual/tutorial ...

随机推荐

  1. js 类型转换学习

    类型转换分为显示转换和隐式转换 参考http://www.cnblogs.com/mizzle/archive/2011/08/12/2135885.html 先事件显示的 通过手动进行类型转换,Ja ...

  2. 求二叉树的深度和宽度[Java]

    这个是常见的对二叉树的操作.总结一下: 设节点的数据结构,如下: class TreeNode { char val; TreeNode left = null; TreeNode right = n ...

  3. [python]pythonic的字典常用操作

    注意:dct代表字典,key代表键值 1.判断字典中某个键是否存在 实现 dct.has_key(key) #False 更Pythonic方法 key in dct #False 2.获取字典中的值 ...

  4. unity HideInInspector 默认值 坑 记录 bug

    1:   如果  一个public字段 刚开始有默认值,然后你你觉得这个值不应该给别人看设置为HideInInspector 后,你再在代码里面调整这个默认属性的值,这个时候代码里面调整的值无效.   ...

  5. CSS 选择器参考手册

    CSS3 选择器 在 CSS 中,选择器是一种模式,用于选择需要添加样式的元素. "CSS" 列指示该属性是在哪个 CSS 版本中定义的.(CSS1.CSS2 还是 CSS3.) ...

  6. 写好shell脚本

    转载自https://mp.weixin.qq.com/s/f3xDHZ7dCQr7sHJ9KDvuyQ 1.提供--help标记 安装在系统上的二进制文件通常带有man帮助文档,但对于脚本来说就不一 ...

  7. Docker镜像构建

    一.简介 在构建容器化应用时,相当重要的步骤莫过于镜像制作,本文将介绍镜像制作方法以及镜像制作的建议.通常镜像的制作有两种方式: 使用现有的容器使用docker commit 生成镜像 使用Docke ...

  8. [No0000160]常用C# 正则表达式大全

    正则表达式的本质是使用一系列特殊字符模式,来表示某一类字符串.正则表达式无疑是处理文本最有力的工具,而.NET提供的Regex类实现了验证正则表达式的方法.Regex 类表示不可变(只读)的正则表达式 ...

  9. python技巧 合并两个字典

    python 3.5+ 版本 In [1]: a={'x':2,'y':4} In [2]: b={'c':1,'d':3} In [3]: c={'c':3,'y':6} In [4]:  w={* ...

  10. 【Android】21.4 图片动画缩放示例

    分类:C#.Android.VS2015: 创建日期:2016-03-21 一.简介 该例子演示如何动画缩放图片,实现类似"点击看大图"的效果. 二.示例 1.运行截图    2. ...