Spring Cloud Alibaba RocketMQ架构与源码分析

矫情吗；* 2023-01-21 10:28 22阅读 0赞

### 文章目录 ###

*   *   *  （一）Spring Cloud Stream架构
         *   *   *  【Spring Messaging】
                 *  【Spring Integration】
                 *  【Binders】
                 *  【Bindings】
         *  （二）Spring Cloud Alibaba RocketMQ架构
         *  （三）Spring Cloud Stream消息发送流程
         *  （四）RocketMQ Binder集成消息发送
         *  （五）Spring Cloud Stream消息订阅流程

### （一）Spring Cloud Stream架构 ###

Spring Cloud Stream是Spring Cloud体系内的一个框架，用于构建与共享消息传递系统连接的高度可伸缩的事件驱动微服务，其目的是简化消息业务在Spring Cloud应用程序中的开发。

Spring Cloud Stream的架构图如图所示，应用程序通过Spring Cloud Stream注入的输入通道inputs和输出通道outputs与消息中间件Middleware通信，消息通道通过特定的中间件绑定器Binder实现连接到外部代理  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70]  
Spring Cloud Stream的实现基于发布/订阅机制，核心由四部分组成  
1：Spring Messaging  
2：Spring Integration  
3：Spring Cloud Stream中的Binders  
4：Spring Cloud Stream中的Bindings

##### 【Spring Messaging】 #####

Spring Messaging是Spring中的统一消息编程模型，主要包括如下对象  
1：Message:消息对象，包含消息头Header和消息体Payload  
2：MessageChannel：消息通道接口，用于接收消息，提供send方法将消息发送至消息通道  
3：MessageHandler：消息处理器接口，用于处理消息逻辑

##### 【Spring Integration】 #####

Spring Integration是用于支持企业集成的一种扩展机制，作用是提供一个简单的模型来构建企业集成解决方案，对Spring Messaging进行了扩展  
1：MessageDispatcher:消息分发接口，用于分发消息和添加删除消息处理器  
2：MessageRouter:消息路由接口，定义默认的输出消息通道  
3：Filter：消息的过滤注解，用于配置消息过滤表达式  
4：Aggregator：消息的聚合注解，用于将多条消息聚合成一条  
5：Splitter：消息的分割，用于将一条消息拆分成多条

##### 【Binders】 #####

目标绑定器，负责与外部消息中间件系统集成的组件  
1：doBindProducer：绑定消息中间件客户端发送消息模块  
2：doBindConsumer：绑定消息中间件客户端接收消息模块

##### 【Bindings】 #####

外部消息中间件系统与应用程序提供的消息生产者和消费者，它是由Binders创建，是之间的桥梁

Spring Cloud Stream官方提供了Kafka Binder和RabbitMQ Binder，用于集成Kafka和RabbitMQ, Spring Cloud Alibaba中加入了RocketMQ Binder，用于将RocketMQ集成到Spring Cloud Stream.

### （二）Spring Cloud Alibaba RocketMQ架构 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 1]  
MessageChannel ( output):消息通道，用于发送消息，Spring Cloud Stream的标准接口  
MessageChannel ( input）:消息通道，用于订阅消息，Spring Cloud Stream的标准接口  
Binder bindProducer:目标绑定器，将发送通道发过来的消息发送到RocketMQ消息服务器，由Spring Cloud Alibaba团队按照Spring Cloud Stream的标准协议实现。  
Binder bindConsumer:目标绑定器，将接收到RocketMQ消息服务器的消息推送给订阅通道，由Spring Cloud Alibaba团队按照Spring Cloud Stream的标准协议实现。

### （三）Spring Cloud Stream消息发送流程 ###

Spring Cloud Stream消息发送流程如图所示，包括发送、订阅、分发、委派、消息处理等，具体实现如下。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 2]

1：在业务代码中调用MessageChannel接口的Send ()方法，例如source.output ( ).send(message)  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 3]  
2：AbstractMessageChannel是消息通道的基本实现类，提供发送消息和接收消息的公用方法。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 4]  
3：在发送前会先执行消息发送前置拦截器，ChannelInterceptor对象  
![在这里插入图片描述][20210515171818612.png]  
4：消息发送到AbstractSubscribableChannel类实现的doSend()方法  
org.springframework.integration.channel.AbstractSubscribableChannel\#doSend

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 5]  
5：通过消息分发类MessageDispatcher把消息分发给MessageHandler.  
![在这里插入图片描述][20210515171840974.png]  
6：从AbstractSubscribableChannel的实现类DirectChannel得到MessageDispatcher的实现类UnicastingDispatcher。

![在这里插入图片描述][20210515171848244.png]  
7：调用dispatch()方法把消息分发给各个MessageHandler。遍历所有MessageHandler，调用handleMessage()处理消息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 6]  
8：AbstractMessageChannelBinder在初始化Binding时，会创建并初始化SendingHandler,调用subscribe()添加到handlers列表。  
producer的MessageHandler是由消息中间件Binder来完成的，Spring Cloud Stream提供了创建MessageHandler的规范。  
AbstractMessageChannelBinder的初始化由AbstractBindingLifecycle在Spring容器加载所有Bean并完成初始化之后完成。

### （四）RocketMQ Binder集成消息发送 ###

AbstractMessageChannelBinder类提供了创建MessageHandler的规范,createProducerMessageHandler方法在初始化Binder的时候会加载。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 7]  
RocketMQMessageChannelBinder类根据规范完成RocketMQMessageHandler的创建和初始化，RocketMQMessageHandler是消息处理器MessageHandler的具体实现，RocketMQMessageHandler在RocketMQBinder中的作用是转化消息格式并发送消息。

RocketMQMessageHandler中持有RocketMQTemplate对象，RocketMQTemplate是对RocketMQ客户端API的封装，Spring Boot中已经支持RocketMQTemplate,Spring Cloud Stream对其兼容。  
DefaultMQProducer是由RocketMQ客户端提供的API，发送消息到RocketMQ消息服务器都是由它来完成的。

RocketMQMessageHandler是消息发送的处理逻辑，解析Message对象头中的参数，调用RocketMQTemplate中不同的发送消息接口。

org.springframework.cloud.stream.binder.rocketmq.integration.RocketMQMessageHandler\#handleMessageInternal

{ 
            try { 
    			//获取消息目的地
                String tags = Optional.ofNullable(message.getHeaders().get("TAGS")).orElse("").toString();
                SendResult sendRes = null;
    			//发送事务消息
                if (this.transactional) { 
                    sendRes = this.rocketMQTemplate.sendMessageInTransaction(this.groupName, topicWithTags.toString(), message, message.getHeaders().get("TRANSACTIONAL_ARG"));
                    log.debug("transactional send to topic " + topicWithTags + " " + sendRes);
                } else { 
    				//设置定时消息的参数
                    int delayLevel = 0;
                    try { 
                        Object delayLevelObj = message.getHeaders().getOrDefault("DELAY", 0);
                        if (delayLevelObj instanceof Number) { 
                            delayLevel = ((Number)delayLevelObj).intValue();
                        } else if (delayLevelObj instanceof String) { 
                            delayLevel = Integer.parseInt((String)delayLevelObj);
                        }
                    } catch (Exception var7) { 
                    }
                    //同步消息发送
                    if (this.sync) { 
    					//顺序消息
                        sendRes = this.rocketMQTemplate.syncSend(topicWithTags.toString(), message, (long)this.rocketMQTemplate.getProducer().getSendMsgTimeout(), delayLevel);
                        log.debug("sync send to topic " + topicWithTags + " " + sendRes);
                    } else { 
    					//异步消息，即普通消息
                        this.rocketMQTemplate.asyncSend(topicWithTags.toString(), message, new SendCallback() { 
    						//异步消息回调方法
    						//异步发送结果成功
                            public void onSuccess(SendResult sendResult) { 
                                RocketMQMessageHandler.log.debug("async send to topic " + topicWithTags + " " + sendResult);
                            }
    	                    //异步发送结果失败
                            public void onException(Throwable e) { 
                                RocketMQMessageHandler.log.error("RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
                                if (RocketMQMessageHandler.this.getSendFailureChannel() != null) { 
                                    RocketMQMessageHandler.this.getSendFailureChannel().send(RocketMQMessageHandler.this.errorMessageStrategy.buildErrorMessage(new MessagingException(message, e), (AttributeAccessor)null));
                                }
    
                            }
                        });
                    }
                }
                if (sendRes != null && !((SendResult)sendRes).getSendStatus().equals(SendStatus.SEND_OK)) { 
                    if (this.getSendFailureChannel() == null) { 
                        throw new MessagingException(message, new MQClientException("message hasn't been sent", (Throwable)null));
                    }
    
                    this.getSendFailureChannel().send(message);
                }
            } 
            }
    
        }

发送普通消息、事务消息、定时消息还是顺序消息，由Message对象的消息头Header中的属性决定，在业务代码创建Message对象时设置。

### （五）Spring Cloud Stream消息订阅流程 ###

在Spring Cloud Stream中接收消息和发送消息的消息模型是一致的，Binder中接收到的消息先发送到MessageChannel ,由订阅的MessageChannel通过Dispatcher转发到对应的MessageHandler进行处理。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 8]  
1：RocketMQInboundChannelAdapter调用sendMessage( )发送消息。  
org.springframework.cloud.stream.binder.rocketmq.integration.RocketMQInboundChannelAdapter.BindingRocketMQListener\#onMessage  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 9]  
org.springframework.integration.endpoint.MessageProducerSupport\#sendMessage  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 10]  
2：通过调用this.getOutputChannel()获取匹配的通道名称，getOutputChannel(）得到的MessageChannel是在初始化RocketMQ Binder时传入的DirectChannel

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 11]

3：MessagingTemplate继承了GenericMessagingTemplate类，实际执行了doSend()方法发送消息。  
org.springframework.messaging.core.GenericMessagingTemplate\#doSend

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 12]4：由于MessageChannel的实例是DirectChannel对象，就复用了Spring Cloud Stream消息发送流程中提到的流程，通过消息分发类MessageDispatcher把消息分发给MessageHandler。  
DirectChannel对应的消息处理器是StreamListenerMessageHandler，在消息处理器中回调使用了@StreamListener注解的业务方法。

org.springframework.cloud.stream.binding.StreamListenerMessageHandler\#handleRequestMessage

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 13]  
5：InvocableHandlerMethod中持有BeanFactory、Method、MethodParameter等对象，使用Java反射机制完成回调。  
我们都知道可以通过定义注解@StreamListener来完成消息的监听，那么这个是在什么时候完成注解的扫描和对象注入呢，这个主要的处理逻辑在StreamListenerAnnotationBeanPostProcessor类上  
org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor\#afterSingletonsInstantiated

在Spring容器管理的所有单例对象初始化完成之后，遍历StreamListenerHandlerMethodMapping,进行StreamListenerMessageHandler和InvocableHandlerMethod的创建和初始化  
从类名看显而易见，StreamListenerHandlerMethodMapping保存了StreamListener和HandlerMethod的映射关系。根据代码逐渐往上找，创建映射关系也是在StreamListenerAnnotationBeanPostProcessor类中完成的。

6：这其中一个非常重要的容器mappedListenerMethods 就保存了相关的映射信息，那么现在我们在往前走下，看看在什么时候这些映射关系对象被添加到当前集合中，通过IDEA的搜索，可以发现如下代码进行了@StreamListener注解扫描获取对象

private final MultiValueMap<String, StreamListenerHandlerMethodMapping> mappedListenerMethods = new LinkedMultiValueMap<>();

org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor\#postProcessAfterInitialization

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 14]

在上面代码中会获取所有的方法，然后通过下面的代码查找被注解StreamListener 修饰的类以及方法

StreamListener streamListener = AnnotatedElementUtils
    					.findMergedAnnotation(method, StreamListener.class);

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 15]  
StreamListenerAnnotationBeanPostProcessor找到所有使用@StreamListener的Method ,并创建StreamListenerHandlerMethodMapping对象，将映射关系保存到集合中。后面消息转发的时候就是通过这个集合定位具体接收消息的方法，通过反射调用对应的方法最终完成消息的消费。

【简要总结】  
Spring Cloud Stream提供了简单易用的消息编程模型，内部基于发布/订阅模型实现。  
Spring Cloud Stream的Binder提供标准协议，不同的消息中间件都可以按照标准协议接入。  
Binder提供bindConsumer和bindProducer接口协议，分别用于构造生产者和消费者。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70]: /images/20221020/b561fdf78dba4883aa8571cc1a2971fd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 1]: /images/20221020/689a905009e24e45ac37785ea272e488.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 2]: /images/20221020/618524f68313494ab5d251d8509142c3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 3]: /images/20221020/399d5338053447a4a68e6d0e33e44c25.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 4]: /images/20221020/3739eec1bf834317840c892da23e752e.png
[20210515171818612.png]: /images/20221020/47a2df06136c471f8b978650ed6f2956.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 5]: /images/20221020/04070166d32740d3a8b2d26a1505c806.png
[20210515171840974.png]: /images/20221020/a44765051388407c9396ebe5ad6c02a0.png
[20210515171848244.png]: /images/20221020/8288b5958e9f45e3b1a12da44ae641c8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 6]: /images/20221020/bb2636763ca14d2daf6e65afe0123bd5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 7]: /images/20221020/7e56c8ea2b824492a51c9b5e56f7fe46.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 8]: /images/20221020/acc7ac996ac34ce6b41885e77201d19f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 9]: /images/20221020/805897f635744bd7af0423085567c885.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 10]: /images/20221020/bc3f29fe22c94e4584d9dfb58ed7352c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 11]: /images/20221020/9727b1ea7674498db781589c2cb05e73.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 12]: /images/20221020/a1245d1fca934ecd83e46ece99d1c2b1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 13]: /images/20221020/b5902301633240daa6427f3026c0b4d2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 14]: /images/20221020/d4c681f7ed1e4cabb09e38d19de002df.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 15]: /images/20221020/9e80783d70444575b5283ac0b0905dc7.png