RocketMQ整合Spring后，可以很方便的使用。那么这里rocketmq-spring做了哪些事情，可以使得我们可以很方便使用呢？
一、rocketmq-spring对消息的处理过程
rocketmq-spring-boot整合后的入口：org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration，通过这个类，我们可以它的一系列动作操作，会首先将相关配置MessageConverterConfiguration、ListenerContainerConfiguration、ExtProducerResetConfiguration、ExtConsumerResetConfiguration、RocketMQTransactionConfiguration、RocketMQListenerConfiguration进行导入。在自动注入前处理事务配置RocketMQTransactionConfiguration，在自动注入后处理消息转换器配置MessageConverterConfiguration。
完成之后会填充两个bean:
defaultMQProducer
defaultLitePullConsumer
根据当前启动的生产和消费者，进行填充。完成后，就可以根据注册的bean容器，此时会执行bean后置处理器操作，完成对带RocketMQMessageListener的注解进行增强，然后注册。注册完成后，便可以对默认rocketmq监听器容器DefaultRocketMQListenerContainer进行启动，启动的过程其实是启动DefaultMQPushConsumer的过程，也即此时会调用rocketmq的this.defaultMQPushConsumerImpl.start();完成对消费者的启动。启动消费者完成后，等待消费。根据消息是否需要顺序消费，分为顺序消费和并发消费。此时都会调用同一个 handleMessage(messageExt)来处理消息，此时可以看到分为两种：
rocketMQListener
rocketMQReplyListener
此时根据当前选择的监听器来做对应业务实现处理，执行消费。
这里可以看到rocketmq-spring对litepull模式的消息没有demo。需要去rocketmq项目找例子看具体的过程。
1.首先我们可以从rocketmq-spring-boot-sample中看到整合还是很方便的。
下面我们来看一下example:
二、生产者
生产者demo:
@SpringBootApplication
publicclassProducerACLApplicationimplementsCommandLineRunner {
@Override
publicvoidrun(String... args) throwsException {
// 发送消息
SendResultsendResult=rocketMQTemplate.syncSend(springTopic+":acl", "Hello, ACL Msg!");
System.out.printf("syncSend1 to topic %s sendResult=%s %n", springTopic, sendResult);
// 使用spring message发送消息
sendResult=rocketMQTemplate.syncSend(springTopic, MessageBuilder.withPayload("Hello, World! I'm from spring message & ACL Msg").build());
System.out.printf("syncSend2 to topic %s sendResult=%s %n", springTopic, sendResult);
//发送事务消息
testTransaction();
}
从demo中，我们可以看到通常分为三类：
使用rocketmq模板发送消费
使用springMessage发送消息
发送事务消息
同时spring自动注入后，其实是可以通过RocketMQProperties拿到的topic信息有限，因此这里采用了值注入的方式，方便放入多个topic，从而实现对多个业务系统对接消息。
三、消费者
同样，也是配置好需要消费的信息后，执行监听器或者使用定时任务
@SpringBootApplication
publicclassConsumerACLApplicationimplementsCommandLineRunner {
@Resource
privateRocketMQTemplaterocketMQTemplate;
@Override
publicvoidrun(String... args) throwsException {
List<String>messages=rocketMQTemplate.receive(String.class);
System.out.printf("receive from rocketMQTemplate, messages=%s %n", messages);
}
}
或者采用监听的方式：
@Service
@RocketMQMessageListener(topic="normal_topic_define_in_cloud_MQ",consumerGroup="group_define_in_cloud_MQ",accessKey="AK", secretKey="SK")
publicclassACLStringConsumerimplementsRocketMQListener<String> {
@Override
publicvoidonMessage(Stringmessage) {
System.out.printf("------- ACL StringConsumer received: %s \n", message);
}
}
配置和RocketMQProperties里面的配置名称一样，它就可以帮你读到对应的配置，并进行相关bean的注入，可以在RocketMQAutoConfiguration可以看到对应的配置信息。比如此时业务系统是消费端的时候，此时会将对应的配置信息注入到defaultLitePullConsumer,相关参数,来源于rocketMQProperties和consumerConfig中：
pullConsumer拉取消费者
nameServernameServer地址
group消费组
topic主题
accessChannel通道
messageModel消息模式，分为广播、集群
selectorType筛选器类型，分为sql92、tag
accessKeykey
secretKey密钥
pullBatchSize拉取批次大小
tlsEnable是否开启tls
enableMsgTrace是否开启msgTrace
customizedTraceTopic自定义链路主题
namespace命名空间
instanceName实例名称
RocketMQListenerConfiguration会注册相关的监听器后置处理器。可以看到后置处理器RocketMQMessageListenerBeanPostProcessor中会执行后置处理方法postProcessAfterInitialization对带RocketMQMessageListener注解的class进行增强，同时在监听容器配置中注册容器：
@Override
publicObjectpostProcessAfterInitialization(Objectbean, StringbeanName) throwsBeansException {
Class<?>targetClass=AopUtils.getTargetClass(bean);
RocketMQMessageListenerann=targetClass.getAnnotation(RocketMQMessageListener.class);
if (ann!=null) {
// 执行注解增强
RocketMQMessageListenerenhance=enhance(targetClass, ann);
//如果监听容器配置不为空，则注册容器
if (listenerContainerConfiguration!=null) {
listenerContainerConfiguration.registerContainer(beanName, bean, enhance);
}
}
returnbean;
}
注册完容器后，就可以启动容器了，因此可以在org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration#registerContainer方法中看到容器启动：
container.start();
也即此时我们可以看到容器是默认rocketmq监听器容器DefaultRocketMQListenerContainer启动的是默认推模式的消费DefaultMQPushConsumer，而启动的时候其实就是调用rocketmq中的消费者方法：
this.defaultMQPushConsumerImpl.start();
启动消费者，等待消费。
推消费分为两种，可以从消费模式种看到：
在默认mq监听容器DefaultRocketMQListenerContainer中可以看到在初始化rocketmq推模式消费 initRocketMQPushConsumer，这个方法很重要，会通过onMessage执行我们写的监听方法
switch (consumeMode) {
caseORDERLY:
consumer.setMessageListener(newDefaultMessageListenerOrderly());
break;
caseCONCURRENTLY:
consumer.setMessageListener(newDefaultMessageListenerConcurrently());
break;
default:
thrownewIllegalArgumentException("Property 'consumeMode' was wrong.");
}
根据消费模式分为两种，一种是按照顺序消费，一种是并发消费，可以看到里面有一个重要方法：
handleMessage(messageExt);
我们可以看到这个方法：
privatevoidhandleMessage(
MessageExtmessageExt) throwsMQClientException, RemotingException, InterruptedException {
//不带响应
if (rocketMQListener!=null) {
//执行rocketmq监听 处理消息
rocketMQListener.onMessage(doConvertMessage(messageExt));
} elseif (rocketMQReplyListener!=null) {
//有响应，消费消息
ObjectreplyContent=rocketMQReplyListener.onMessage(doConvertMessage(messageExt));
Message<?>message=MessageBuilder.withPayload(replyContent).build();
org.apache.rocketmq.common.message.MessagereplyMessage=MessageUtil.createReplyMessage(messageExt, convertToBytes(message));
//获取生产者，执行生产者发送回调
DefaultMQProducerproducer=consumer.getDefaultMQPushConsumerImpl().getmQClientFactory().getDefaultMQProducer();
producer.setSendMsgTimeout(replyTimeout);
producer.send(replyMessage, newSendCallback() {
@Override
publicvoidonSuccess(SendResultsendResult) {
if (sendResult.getSendStatus() !=SendStatus.SEND_OK) {
log.error("Consumer replies message failed. SendStatus: {}", sendResult.getSendStatus());
} else {
log.debug("Consumer replies message success.");
}
}
@Override
publicvoidonException(Throwablee) {
log.error("Consumer replies message failed. error: {}", e.getLocalizedMessage());
}
});
}
}
如果监听器rocketMQListener存在，则执行消费消息，也即onMessage的操作，执行业务系统的onMessage接口的实现。
如果响应监听器rocketMQReplyListener存在，则获取响应信息，MessageBuilder构建Message对象信息，此时的消息类型是Spring Message类型的，需要转成Rocketmq自己的类型的。从默认mq推消费者实现中获取默认的mq生产者，执行生产者发送回调。
从这个接口的注释可以看到：
/**
* The consumer supporting request-reply should implement this interface.
*
* @param <T> the type of data received by the listener
* @param <R> the type of data replying to producer
*/
publicinterfaceRocketMQReplyListener<T, R> {
/**
* @param message data received by the listener
* @return data replying to producer
*/
RonMessage(Tmessage);
}
四、ExtRocketMQTemplate
在提供的demo中，出现了ExtRocketMQTemplate这个类，用原来的RocketMQTemplate不行吗，为啥会有这个类呢？
@ExtRocketMQConsumerConfiguration(topic = "${demo.rocketmq.topic}", group = "string_consumer", tlsEnable = "${demo.ext.consumer.tlsEnable}")
public class ExtRocketMQTemplate extends RocketMQTemplate {
}
可以看到这个类上面的配置@ExtRocketMQConsumerConfiguration
StringNAME_SERVER_PLACEHOLDER="${rocketmq.name-server:}";
StringGROUP_PLACEHOLDER="${rocketmq.pull-consumer.group:}";
StringTOPIC_PLACEHOLDER="${rocketmq.pull-consumer.topic:}";
StringACCESS_CHANNEL_PLACEHOLDER="${rocketmq.access-channel:}";
StringACCESS_KEY_PLACEHOLDER="${rocketmq.pull-consumer.access-key:}";
StringSECRET_KEY_PLACEHOLDER="${rocketmq.pull-consumer.secret-key:}";
StringTRACE_TOPIC_PLACEHOLDER="${rocketmq.pull-consumer.customized-trace-topic:}";
可以看到如果有多个消息集群的时候，这里可以方便我们替换rocketmq的地址信息，从而实现对多服务的消费。也即业务系统可能出现对接多个消息，其消息的topic可能不同，方便对接，因此其注解上面可以可以方便替换成想要的rocketmq的地址信息。
可以看到三个重置方法：
ExtProducerResetConfiguration
ExtConsumerResetConfiguration
都实现了SmartInitializingSingleton，其会在bean加载完成后执行。
五、RocketMQTemplate
在RocketMQTemplate中，我们看到的是大量的发送方法，主要包括这几类方法：
同步发送
异步发送
只发送不管消费
延迟消息 发送
事务消息发送
根据顺序、延迟程度、是否批量进一步分为下面这些方法：
send
sendAndReceive
syncSend
syncSendDelayTimeSeconds
syncSendDelayTimeMills
syncSendDeliverTimeMills
syncSendOrderly
asyncSend
asyncSendOrderly
sendOneWay
sendOneWayOrderly
sendMessageInTransaction
