一丶概述

RocketMQ 消息发送的原理流程可以分为以下几个步骤：

1. 创建生产者

在发送消息前，客户端首先需要创建一个消息生产者（Producer）实例，并设置必要的配置参数，如NameServer地址、生产组名称、消息发送失败的重试次数等。

2. 启动生产者

创建生产者后，需要调用启动方法来初始化生产者实例。在这个过程中，生产者会与NameServer建立连接，从NameServer获取到所有Broker的地址信息。

3. 发送消息

消息发送分为同步发送、异步发送和单向发送三种方式：

• 同步发送（Synchronous）：
  生产者发送消息后，会在发送线程中等待服务器的响应，直到收到消息发送确认。
• 异步发送（Asynchronous）：
  生产者发送消息后，不会等待服务器的响应，而是通过回调接口处理服务器的响应。
• 单向发送（One-way）：
  生产者只负责发送消息，不等待服务器响应，也不关心消息是否到达服务器。

无论采用哪种发送方式，消息发送的主要流程如下：

1. 消息路由：
   生产者通过负载均衡算法选择一个队列，通常是根据topic和队列选择一个Broker的一个队列来发送消息。
2. 消息发送：
   生产者向选定的Broker发送消息。消息包含了topic、tags、keys、body等信息。
3. 消息存储：
   Broker接收到消息后，会将消息存储到
   CommitLog
   （消息存储文件）中。如果配置了消息重试或者高可靠性相关的配置，Broker可能会执行额外的消息复制或持久化操作以确保消息的可靠性。
4. 写入响应：
   Broker将消息存储确认响应返回给生产者。如果是同步发送，生产者会在这一步等待该响应；如果是异步发送，生产者会在回调函数中处理该响应。

本篇，我们就来简单看下rocketmq从生产者发送消息，学习一下其中优秀的设计！

二丶生产者消息发送

生产者消息发送本质是通过网络io将消息发送到broker中，通常通过
DefaultMQProducer#send(Message)
进行简单的消息发送，如下是其源码

可看到如果设置了autoBatch并且消息本身不是一个批量消息，那么会调用
sendByAccumulator
(使用消息累计器进行发送，猜测会累计到内存中然后批量进行发送)

反之会调用
sendDirect
进行消息发送

1.sendByAccumulator 如何累计消息发送

rocketmq抽象出
ProduceAccumulator
进行消息的累计发送

ProduceAccumulator会将消息根据Topic和tag进行分组存储，然后包装为MessageBatch调用DefaultMQProducer进行发送

2.sendDirect

DefaultMQProducer消息发送会委托给DefaultMQProducerImpl进行发送，这两个类名称很像但是DefaultMQProducerImpl不是DefaultMQProducer的实现，二者是不同维度的：

• DefaultMQProducer是给调用方使用的，相当于门面
• DefaultMQProducerImpl：实现了MQProducerInner，真正实现消息发送机制

1. 指定SendCallback：当异步发送消息的时候，可以实现此接口，实现消息发送成功or失败后的回调
2. 指定MessageQueue：MessageQueue是由
   Topic
   ，
   broker
   ，
   queueId
   组成，一个topic可以分布在多个Broker上（横向扩展），一个broker上可以由多个queue（多个queue并行消费提升吞吐量），因此通过发送消息指定MessageQueue可以实现消息的局部有序（消费者使用
   MessageListenerOrderly
   单线程进行消费）

下面我们来看看消息发送的具体实现，这部分代码在
DefaultMQProducerImpl#sendDefaultImpl

1.获取路由信息

获取路由信息，即从
生产者向 NameServer 查询特定 Topic 的路由信息。这个路由信息包括了这个 Topic 有哪些 Broker 持有，以及这些 Broker 上各自的 Queue 数据

• NameServer 是 RocketMQ 中的一个关键组件，起到了服务注册中心的作用。所有的 Broker 启动时会向所有的 NameServer 注册，包括其 IP 地址、端口、存活状态以及所持有的 Topic 信息。NameServer 会持有整个消息系统的 Broker 服务器列表及其路由信息。
• 当生产者启动时，它会根据配置好的 NameServer 地址列表与 NameServer 集群建立连接。
• 生产者会在本地缓存从 NameServer 获取到的路由信息，以便快速选择目标 Queue 进行消息发送。为了确保路由信息的准确性，生产者会定期（如每隔30秒）或在发送消息时发现路由信息不可用时，重新从 NameServer 更新这些信息，并且生产者发送消息的时候根据本地缓存的路由信息选择一个 Queue 来发送消息。

通过这种方式，RocketMQ 确保生产者能够及时获取和更新路由信息，以及将消息发送到正确的 Broker 和 Queue。这个机制也使得 RocketMQ 能够在 Broker 或队列变化时动态适应，保证消息传输的高可用性和可扩展性。

生产者会优先从
ConcurrentMap<String/* topic */, TopicPublishInfo> topicPublishInfoTable
中获取路由，反之使用rpc请求nameServer获取路由信息

另外在生产者启动的时候，会触发MQClientInstance的start，其中会使用juc调度线程池进行路由信息的定期更新（默认30秒一次）。

这里居然没有使用长轮询，理论上长轮询相比于这种周期请求有更好的及时性，rocketmq可能是考虑到

• 长轮询的方式需nameServer维护连接状态，而周期轮询对于nameServer负担更小
• 周期请求可以让生产者设置周期频率
• 流量更加均匀：长轮询在路由信息发生变化的时候，nameServer需要立马将变化后的信息发送给hang住的producer，不如周期轮询来得流量均衡。
• 大部分情况下，路由信息不会频繁变化的，定期轮询可满足需要，不像配置中心配置变更是比较频繁的，并且配置中心对于配置变更及时性有比较高的要求。

2.负载均衡的选择一个MessageQueue

如下，如果是同步发送消息，一般会尝试3次，在获取到路由信息后会负载均衡的选择一个MessageQueue进行发送。

RocketMq支持三种选择MessageQueue的方式

• 发送消息的时候，传入MessageQueueSelector的实现选择队列；

• 未开启Broker故障延迟机制(sendLatencyFaultEnable:false)，会采用默认轮训机制(默认是此种实现方式)

  
  

• 开启Broker故障延迟机制(sendLatencyFaultEnable:true)，会根据brokerName的可用性选择队列发送（当需要顺序消息的时候不建议打开，会影响到消息的顺序性）

  
  
  

  其中是否可用，是否可达，依赖LatencyFaultTolerance进行实现：

  LatencyFaultTolerance 实现了一个基于延迟的容错策略。它记录了每个 Broker 的历史网络延迟记录和可用性状态，并根据这些信息智能选择最佳的 Broker 进行消息发送。原理包括以下几个关键点：

  
  
  • 延迟记录：每次发送消息时，LatencyFaultTolerance 都会记录下发送操作的延迟时间。如果发送成功，那么这次操作的延迟时间就会被记录下来。
  • 故障切换：如果发送消息时发生超时或异常，LatencyFaultTolerance 会将该 Broker 标记为不可用，并计算一个“不可用时长”。在该时长内，Broker 将不会被选中发送消息。
  • 动态容错：
    LatencyFaultTolerance 会根据之前记录的延迟时间，动态计算每个 Broker 的权重，并选择权重最小（表示网络状态最好）的 Broker 进行消息发送。
  • 自动恢复：
    被标记为不可用的 Broker 不是永久性的。随着时间的推移，Broker 的状态可以从不可用恢复到可用，这通常是通过“不可用时长”来确定的。一旦超过这个时长，Broker 将重新参与到Broker选择过程中。
  • Broker选择：
    生产者在发送消息前会从 LatencyFaultTolerance 中获取一个推荐的 Broker。选择过程排除了不可用的 Broker，并考虑了网络延迟和Broker的历史表现。

3.消息发送

至此，我们以及选择了一个MessageQueue接下来就是发送消息了。

在发生之前会从路由信息中获取发送的地址，这里只会选择master角色的broker进行发送

接下来会回调一些扩展性的钩子，如CheckForbiddenHook，SendMessageHook。
然后调用MQClientAPIImpl#sendMessage进行发送，最终调用RemotingClient进行消息发送，RemotingClient是rocketmq对网络通信的实现

无论是单向，还是异步，还是同步，最终都是使用tcp协议进行发送，这里rocketmq使用了netty提供高效的网络通信。源码如下:

netty的部分，不做过多赘述，详细学习：
Netty源码学习7——netty是如何发送数据的 - Cuzzz - 博客园 (cnblogs.com)

三丶总结

感觉学到了什么，又感觉什么都没学到

• NameServer：实现producor，broker，consumer的解耦合，互相不需要感知彼此的村子，本质是一个注册中心。
• 路由信息使用定期轮询，而不是长轮询
  
  • 长轮询的方式需nameServer维护连接状态，而周期轮询对于nameServer负担更小
  • 周期请求可以让生产者设置周期频率
  • 流量更加均匀：长轮询在路由信息发生变化的时候，nameServer需要立马将变化后的信息发送给hang住的producer，不如周期轮询来得流量均衡。
  • 大部分情况下，路由信息不会频繁变化的，定期轮询可满足需要，不像配置中心配置变更是比较频繁的，并且配置中心对于配置变更及时性有比较高的要求。
• 负载均衡：
  
  • rocketmq的负载均衡，大多实在客户端做的，在消息发送中的体现就是，producer自己实现负载均衡，而不是由一个中心化的网关实现，这样去中心化的设计，利于producer的横向扩展！
  • 默认情况下使用轮询，而且使用ThreadLocal记录轮询到的index，一定程度上减少大量消息发送时候的锁竞争
  • LatencyFaultTolerance：基于每一次发送消息的统计信息，如果发送消息时发生超时或异常，LatencyFaultTolerance 会将该 Broker 标记为不可用，并计算一个“不可用时长”。在该时长内，Broker 将不会被选中发送消息

故事

接二连三地背锅让小猫的内心受到了前所未有的打击。这也是他职业生涯中的第一次。感兴趣的伙伴们如果想了解一下小猫怎么了，可以看一下“
幂等事件
”以及“
缓存击穿事件
”。

这天组长找小猫来到了一间会议室。

“在这么短的时间内发生了这么多的事故，我想也你心里也不好受，也不怪你，毕竟刚接手项目。以前项目中可能本身存在一定问题。正好轮到你头上，我希望你也不要灰心......”，组长在一边balabala。

小猫在一旁小鸡啄米似的点着头。紧张的内心缓和了许多，“听组长这语气，貌似不扣我绩效啊”，小猫心里寻思着。

“但是呢，事情是发生了，系统中估计还有其他的问题，无论是业务上的还是代码上的亦或是设计上的，然后我希望你最近可以花时间整理一下，然后输出一份项目文档分析。等到下次月会的时候一起分享一下当前系统的情况。”

小猫连连点头，心里琢磨“看来还是躲不过当众批斗这关啊，罢了罢了，可是，这样的一份文档该从哪个口子开始呢，到底该怎么写呢”

此时的小猫内心又开始不安起来。

聊聊熟悉新项目

当接手到新的系统的时候，大家是如何进行熟悉的呢？其实老猫在上一篇“
缓存击穿事件
”的文末就问过大家了，不晓得大家是怎么处理的，欢迎大家留言。

接下来和大家聊一下老猫对一个新系统的熟悉过程。如果大家觉得有所帮助也可以借鉴一下。主要按照以下方法：

尝试画一下用例图

当接受到一个新的业务系统之后，首先咱们至少需要知道当前这个系统是干什么的，所以有时候就需要抽时间找到相关的产品经理了解一下业务，此时产品经理可能会和你聊一下现有的业务现状和背景，但是有可能也会丢给你一份V0-Vn版本的产品需求方案。如果是后者记住千万得忍住，不要用显示器砸产品的脸，因为你们的合作尚未开始。咳咳，开个玩笑，言归正传。

我们先了解一下什么是用例图。

用例图简析

用例是系统中的一个功能单元，可以被描述为执行者与主体之间的一次交互行为。执行者是与系统、子系统或类发生交互作用的外部用户、进程或其他系统的理想化角色。

用途：能够列出系统中的用例和执行者，并显示哪个执行者参与了哪个用例的执行。

针对之前小猫遇到的“下单付款的业务点”，咱们来画个用例图说明一下。如下图：

如上图其实就是一个简单的用例图。我们需要搞清楚的是各种线条的含义。

1. a线条表示的是关联即执行者与其参与的用例之间的通信路径。用实线表示。
2. b线条表示包含，在基用例上插入附加的行为，并且显式地描述了该插入。
3. c线条表示扩展，在基用例上插入附加的行为，基用例并不知道。
4. d线条表示用例泛化，一般用例和特殊用例之间地关系，其中特殊用例继承了一般用例的特征并增加了新的特性。

这样我们就可以很清晰地了解当前的业务现状。

后端模型梳理

当梳理完当前的系统功能点以及业务形态的时候，我们就可以介入去看一下现有系统的模型了即DB数据库的表。这样我们就能知道当前设计的系统是如何对业务进行抽象的。
那么在看相关表的时候，其实我们就可以慢慢地将ER图进行绘制出来了。

什么是ER图

E-R图即全称实体-联系图(Entity Relationship Diagram)，它提供了表示实体类型、属性和联系的方法，用来描述现实世界的概念模型。

通过其定义其实我们就知道了在ER图里面有三个比较重要的点，分别是实体类，属性，联系。当我们在整理DB表的时候其实对应的就是我们的表、表字段以及对应的表和表之间的关系。

看个例子，下面老猫绘制一下一般商城系统底层的商品逻辑。

解释一下每一块图的含义：

1. 方块表示一个模型即一个表，当然这个也是ER图中的实体类。
2. 椭圆形表示实体类包含的属性。
3. 菱形就表示两个类之间的动作行为关系，比如上图中商品上架到货架上。日常中老师给学生上课，那么菱形中可能就是上课。
4. 线条上的1和n就更清晰了，就是一对多，多对一，一对一的关系。

上图中其实我们就可以比较清晰地看到，在当前的这个系统中存在三个比较重要的实体概念，分别是商品、商品池、以及货架。从图中我们也可以大概地看到他们之间的关系。

当咱们梳理完ER图之后，其实上述的用例业务图如何在现有系统中的抽象大概就清楚了。

聊到这里，咱们从上帝视角去看一下，我们给当前这个系统赋予了骨架，接下来得让它的开始心脏跳动，血液奔腾起来，让整个系统赋予灵魂。
那么接下来，我们就把模型通过流程的方式串起来。

核心流程以及状态机流转

咱们直接看一下例子，其实老猫觉得流程图的梳理可能比较简单，但是难的是如何去把控整个流程中的环节。如果在绘制的时候想的比较细致，可能每一步的落库环节都会去记录。这样的话对于业务的专注度就会少一些。如果画粗了，模型的对应关系可能又把控不好。所以这个地方老猫觉得还是比较考验程序员的概括能力以及业务的理解能力的。

上述图中，老猫简单画了一个流程图，当然流程图中可能会存在纰漏，大家不要太过较真，在此是说明这么一个事情，咱们暂且不谈里面业务流程的准确度。
上面流程中我们看到有以下图形内容：

1. 起始节点，咱们用圆圈表示，当然可以选择自己喜欢的颜色，没有太多标准。
2. 流程进行流转的时候，我们用了相关的箭头线表示，涉及到核心业务操作的时候就是方块。
3. 遇到分支节点的时候，咱们用菱形去做路由。
4. 遇到一些异步操作的时候，老猫喜欢用虚线去表示。

上述这种流程的表示其实是比较简单的，我们不用去在意系统边界。只管绘制即可。

但是现在的开发体系咱们往往都是微服务化的，那么此时我们可能就要考虑到不同系统之间的交互流程。由此，咱们可能机会引入泳道的概念。见下图。

上述图中我们就可以看到各个系统应用之间的交互，每一个泳道就代表着其中的一个微服务系统。这是老猫日常熟悉业务中真实绘制的一张图，再次强调一下大家要看的是绘图的一些思路，不要太过纠结业务。

那么再细节一点，比如讨论到订单状态的一些流转的时候，此时我们为了更好的把控，会使用一些状态的流转图去梳理。

上述主要是阐述整个状态在流程中的流转，当然很多时候状态位比较简单的时候，咱们也可以不用画，可能当状态比较复杂多样的时候才会去考虑到画状态机。

到此，通过上面的几个步骤的梳理，其实我们已经能够对大部分的业务有个熟悉了。但是有的时候还不够，涉及一些复杂接口交互请求时序的时候，可能还是会翻车弄不清楚，那么此时咱们就要引入时序图进行梳理。

引入时序图

时序图通常为表示多个对象之间消息交互的序列。
其中时序图涉及：参与者、对象、生命线、激活、消息、片段。
当然消息中又包含：同步消息、异步消息、自关联消息、返回消息。

关于其中的解释，如下图例子。

上述简单绘制了下单扣减库存的时序。

在上述图中，老猫用到了其中的一种片段--抉择，其实关于片段还有很多种类，大家绘制时序的时候可以对照着去使用。

复杂代码的类图

完成上面这些梳理之后，这个老系统其实就已经被你搞定了，后续一些比较细节的地方，当然随着业务功能地迭代再慢慢吃透。

类图：是描述类、接口、协同以及他们之间关系的图，用来显示系统中这些概念的静态结构，类图主要由类、接口和各种关系组成。

如果老系统中存在一些屎山代码或者极端优秀的代码，你看不透里面的继承实现关系的，那么可以去绘制一下类图。这里老猫不作展开，先卖个关子，后续等咱们梳理到设计模式相关知识点的时候，咱们再慢慢聊。

写在最后

在日常开发的过程中，我们会遇到各种各样复杂的系统，老猫通过上述手段，基本就可以做一个比较好的拿捏。希望能够帮助到大家，大家如果还有其他比较好的办法，也欢迎在评论区留言。

背景

在feign中，一般是通过eureka、nacos等获取服务实例，但有时候调用一些服务时，人家给的是ip或域名，我们这时候还能用Feign这一套吗？

可以的。

有两种方式，一种是直接指定url：

这种是服务端自己会保证高可用、负载均衡那些。

但也可能对方给了多个url（一般不会这样，但是在app场景下，为了极致的高可用，可能会配置多个服务端地址），此时就需要咱们在客户端配置多个url，并且进行负载均衡。

此时应该怎么配置呢？前面的文章提到了，可以像下面这样配置：

spring:
  application:
    discovery:
      client:
        simple:
          instances:
            echo-service-provider:
              - uri: http://1.1.1.1:8082
                metadata:
                  my: instance1
              - uri: http://2.2.2.2:8082
                metadata:
                  my: instance2

但是，这第二种方式下，如果你同时使用了nacos，且打开了
spring.cloud.loadbalancer.nacos.enabled=true
这个选项，就会发现，调用报错了。

原因分析

从上面的错误堆栈可以看到，在执行Double.parseDouble的时候抛了空指针异常，为啥还涉及什么浮点数呢？

我们定位到报错的地方，原来是获取服务实例的权重值的时候，报错了：

很明显，是因为我们的服务实例里面的metadata字段，没有nacos.weight这个属性，所以是null，自然就空指针了。

这里的服务实例是ServiceInstance，这是个通用接口，定义在spring-cloud-commons中的，按理说，你作为一种实现，是需要考虑到传入的ServiceInstance不一定就有这个属性，比如可能是Eureka管理的。但是上面报错的地方又强制假设这个地方一定是metadata拥有nacos.weight。

这块就是个兼容性bug，看了下最新版本，也还是未修复：

https://github.com/alibaba/spring-cloud-alibaba/blob/2022.x/spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-nacos-discovery/src/main/java/com/alibaba/cloud/nacos/balancer/NacosBalancer.java#L58

接下来，我们看下，那如果是从nacos获取到的serviceInstance，是不是就没有这个问题？为啥配置静态ip地址的时候，就有这个问题。

nacos中获取到的serviceInstance

咱们先把前面的静态ip配置去掉，改为从nacos获取。

从上图看到，此时实例类型是
com.alibaba.cloud.nacos.NacosServiceInstance
:

此时自然就不会报错了。

静态ip时获取到的serviceInstance

在获取服务实例时，入口是
org.springframework.cloud.client.discovery.composite.CompositeDiscoveryClient#getInstances
,它内部聚合了两个discoveryClient，第一个是simpleDiscoveryClient，这个就是从静态ip获取服务实例，可以看到其order是-1，所以它排在了第一位；第二个是nacosDiscoveryClient，由于它的order值是0，所以排序靠后。

从simpleDiscoveryClient中获取到的serviceInstance的类型就是
org.springframework.cloud.client.DefaultServiceInstance
，它内部自然是没有配置nacos相关的metadata的，所以在前面的场景中才会报错。

解决办法一

既然nacos这个loadbalancer不兼容静态ip这种org.springframework.cloud.client.DefaultServiceInstance，那我不使用nacos的loadbalancer不就可以了。

是的，只要你不打开
spring.cloud.loadbalancer.nacos.enabled=true
这个选项，就不会用到nacos的这个loadbalancer。

我们搜了下这个选项：

这被弄成了一个条件注解。这个条件用于以下的自动装配类：

在之前的文章里，我们提到了，每个feign服务只要url没指定，就默认是走负载均衡，就会有一个loadbalancerClient。

每个loadbalancerClient都是通过一个spring容器来的，每个服务都有一个自己的用于创建loadbalancer的spring容器（比如这里的echo-service，就有一个自己的用于创建loadbalancer的spring容器）。这个容器里面默认有啥内容呢？

@LoadBalancerClients(defaultConfiguration = NacosLoadBalancerClientConfiguration.class)

这里的NacosLoadBalancerClientConfiguration.class就会被作为各个spring容器的默认配置类。

这里就会自动配置一个NacosLoadBalancer，一旦有了这个bean，spring-cloud-loadbalancer里的默认配置，就不会生效了：

最终获取bean的时候，就拿到了nacos的这个NacosLoadBalancer类型的bean，进行负载均衡。

这个办法的缺点：

这个选项是全局的，不能针对某一个服务来单独开启，这个选项一旦关了，那么其他的走nacos的服务，也就没法用nacosLoadBalancer了。

所以，我们想到了如下的方法。

解决办法二

我们上面提到，这个nacosLoadBalancer被自动装配进去的，那么，破解自动装配的办法就是你自己定义一个这种类型的bean，它就不会再自动装配了。

这样的话，echo-service-provider的spring容器创建时，就会优先把这个配置class注册到容器里：

这种办法的优势是，可以在
spring.cloud.loadbalancer.nacos.enabled=true
开启的情况下，解决本文的问题。就是，nacos的依然可以用nacosLoadBalancer来负载均衡；静态ip的服务，就可以用轮询这种loadbalancer。

总结

这个feign写得差不多了，后面写点别的。如果后续需要补充这块，再说。

参考

官网有类似bug：

https://github.com/alibaba/spring-cloud-alibaba/issues/3346

前言

对于.NET开发者而言Visual Studio是我们日常工作中比较常用的开发工具，掌握一些Visual Studio实用的搜索、查找、替换技巧可以帮助我们大大提高工作效率从而避免996。

Visual Studio更多实用技巧

• https://github.com/YSGStudyHards/DotNetGuide

代码和功能搜索（Ctrl + T）

Ctrl + T
是一个非常实用的代码和功能搜索技巧，代码搜索支持：文件、类型、成员的搜索，功能搜索支持查找Visual Studio中的所有功能。

代码搜索