RocketMQ Binder集成消息订阅

RocketMQ Binder集成消息订阅

AbstractMessageChannelBinder类中提供了创建MessageProducer的协议，在初始化Binder的时候加载createConsumerEndpoint方法

RocketMQMessageChannelBinder完成RocketMQInboundChannelAdapter的创建和初始化

RocketMQMessageChannelBinder的createConsumerEndpoint方法：

RocketMQInboundChannelAdapter是适配器，需要适配Spring Framework的重试和回调机制，用来订阅消息和转化消息格式。RocketMQListenerBindingContainer是对RocketMQ客户端API的封装，适配器中持有它的对象。

RocketMQ提供两种消费模式：顺序消费和并发消费。RocketMQ客户端API中顺序消费的默认监听器是DefaultMessageListenerOrderly，并发消费的默认监听器是DefaultMessageListenerConcurrently类，无论哪个消费模式，监听器收到的消息都会回调RocketMQListener

RocketMQInboundChannelAdapter中创建和初始化RocketMQListener的实现类

RocketMQInboundChannelAdapter

DefaultMessageListenerOrderly收到RocketMQ消息后，先回调BindingRocketMQListener的onMessage方法，再调用RocketMQInboundChannelAdapter父类的sendMessage方法将消息发送到DirectChannel

Spring Cloud Stream的接收消息和发送消息的消息模型是一致的，Binder中接收的消息先发送到MessageChannel，由订阅的MessageChannel通过Dispatcher转发到对应的MessageHandler进行处理。

RocketMQInboundChannelAdapter的父类MessageProducerSupport的getOutputChannel()得到的MessageChannel是在初始化RocketMQ Binder时传入的DirectChannel

MessageProducerSupport的getOutputChannel方法：

MessagingTemplate继承GenericMessagingTemplate类，实际执行doSend()方法发送消息

MessageChannel的实例是DirectChannel对象，复用前面消息发送流程，通过消息分发类MessageDispatcher把消息分发给MessageHandler

DirectChannel对应的消息处理器是StreamListenerMessageHandler

InvocableHandlerMethod使用java反射机制完成回调，StreamListenerMessageHandler与@

StreamListenerAnnotationBeanPostProcessor的afterSingletonsInstantiated方法：

在Spring容器管理的所有单例对象初始化完成之后，遍历StreamListenerHandlerMethodMapping，进行InvocableHandlerMethod和StreamListenerMessageHandler的创建和初始化

StreamListenerHandlerMethodMapping保存了StreamListener和InvocableHandlerMethod的映射关系，映射关系的创建是在StreamListenerAnnotationBeanPostProcessor的postProcessAfterInitialization()方法

@Override
public final Object postProcessAfterInitialization(Object bean, final String beanName)
      throws BeansException {
   Class<?> targetClass = AopUtils.isAopProxy(bean) ? AopUtils.getTargetClass(bean)
         : bean.getClass();
   Method[] uniqueDeclaredMethods = ReflectionUtils
         .getUniqueDeclaredMethods(targetClass);
   for (Method method : uniqueDeclaredMethods) {
      StreamListener streamListener = AnnotatedElementUtils
            .findMergedAnnotation(method, StreamListener.class);
      if (streamListener != null && !method.isBridge()) {
         this.streamListenerCallbacks.add(() -> {
            Assert.isTrue(method.getAnnotation(Input.class) == null,
                  StreamListenerErrorMessages.INPUT_AT_STREAM_LISTENER);
            this.doPostProcess(streamListener, method, bean);
         });
      }
   }
   return bean;
}
private void doPostProcess(StreamListener streamListener, Method method,
            Object bean) {
        streamListener = postProcessAnnotation(streamListener, method);
        Optional<StreamListenerSetupMethodOrchestrator> orchestratorOptional;
        orchestratorOptional = this.streamListenerSetupMethodOrchestrators.stream()
                .filter(t -> t.supports(method)).findFirst();
        Assert.isTrue(orchestratorOptional.isPresent(),
                "A matching StreamListenerSetupMethodOrchestrator must be present");
        StreamListenerSetupMethodOrchestrator streamListenerSetupMethodOrchestrator = orchestratorOptional
                .get();
        streamListenerSetupMethodOrchestrator
                .orchestrateStreamListenerSetupMethod(streamListener, method, bean);
    }

@Override
        public void orchestrateStreamListenerSetupMethod(StreamListener streamListener,
                Method method, Object bean) {
            String methodAnnotatedInboundName = streamListener.value();

            String methodAnnotatedOutboundName = StreamListenerMethodUtils
                    .getOutboundBindingTargetName(method);
            int inputAnnotationCount = StreamListenerMethodUtils
                    .inputAnnotationCount(method);
            int outputAnnotationCount = StreamListenerMethodUtils
                    .outputAnnotationCount(method);
            boolean isDeclarative = checkDeclarativeMethod(method,
                    methodAnnotatedInboundName, methodAnnotatedOutboundName);
            StreamListenerMethodUtils.validateStreamListenerMethod(method,
                    inputAnnotationCount, outputAnnotationCount,
                    methodAnnotatedInboundName, methodAnnotatedOutboundName,
                    isDeclarative, streamListener.condition());
            if (isDeclarative) {
                StreamListenerParameterAdapter[] toSlpaArray;
                toSlpaArray = new StreamListenerParameterAdapter[this.streamListenerParameterAdapters
                        .size()];
                Object[] adaptedInboundArguments = adaptAndRetrieveInboundArguments(
                        method, methodAnnotatedInboundName, this.applicationContext,
                        this.streamListenerParameterAdapters.toArray(toSlpaArray));
                invokeStreamListenerResultAdapter(method, bean,
                        methodAnnotatedOutboundName, adaptedInboundArguments);
            }
            else {
                registerHandlerMethodOnListenedChannel(method, streamListener, bean);
            }
        }

private void registerHandlerMethodOnListenedChannel(Method method,
                StreamListener streamListener, Object bean) {
            Assert.hasText(streamListener.value(), "The binding name cannot be null");
            if (!StringUtils.hasText(streamListener.value())) {
                throw new BeanInitializationException(
                        "A bound component name must be specified");
            }
            final String defaultOutputChannel = StreamListenerMethodUtils
                    .getOutboundBindingTargetName(method);
            if (Void.TYPE.equals(method.getReturnType())) {
                Assert.isTrue(StringUtils.isEmpty(defaultOutputChannel),
                        "An output channel cannot be specified for a method that does not return a value");
            }
            else {
                Assert.isTrue(!StringUtils.isEmpty(defaultOutputChannel),
                        "An output channel must be specified for a method that can return a value");
            }
            StreamListenerMethodUtils.validateStreamListenerMessageHandler(method);
            StreamListenerAnnotationBeanPostProcessor.this.mappedListenerMethods.add(
                    streamListener.value(),
                    new StreamListenerHandlerMethodMapping(bean, method,
                            streamListener.condition(), defaultOutputChannel,
                            streamListener.copyHeaders()));
        }

StreamListenerAnnotationBeanPostProcessor.this.mappedListenerMethods.add来创建并保存StreamListenerHandlerMethodMapping

这是使用Spring Cloud Stream的消息模型来使用RocketMQ，也可以使用SpringBoot集成的RocketMQ组件。