4.6、rocketmq
作用：
1、削峰填谷
2、功能解耦
3、顺序执行
-
4.6.1、生产消息
-
4.6.2、消费消息
-
4.6.3、场景使用说明
4.6.1、生产消息
4.6.1.1、同步生产消息
// SendResult syncSend("TOPIC:TAG", 数据对象),例如：
rocketMQTemplate.syncSend("Hsp_Topic_Manager_Order:Hsp_Tag_Manager_Order_Create", order);
4.6.1.2、异步生产消息
// void asyncSend("TOPIC:TAG", 数据对象, 回调函数)，例如：
rocketMQTemplate.asyncSend("Hsp_Topic_Manager_Order:Hsp_Tag_Manager_Order_Create", order, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
log.info("消息发送成功：{}", sendResult);
}
@Override
public void onException(Throwable e) {
log.info("消息发送失败", e);
}});
4.6.2、消费消息
注解解释：
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RocketMQMessageListener {
String consumerGroup();// 指定consumerGroup
String topic();// 指定消费的topic
SelectorType selectorType() default SelectorType.TAG; // 指定消费过滤方式: TAG, SQL92，默认TAG
String selectorExpress() default "*"; // 根据过滤方式，定义选择表达式，默认所有
ConsumeMode consumeMode() default ConsumeMode.CONCURRENTLY; // 消费方式：并发，顺序，默认并发
MessageModel messageModel() default MessageModel.CLUSTERING; // 消费模式: 集群, 广播，默认集群
int consumeThreadMax() default 64; //消费的并发线程数
}
4.6.2.1、集群消费
@Component
@Slf4j
@RocketMQMessageListener(topic = "Hsp_Topic_Manager_Order", selectorExpression = "Hsp_Tag_Manager_Order_Create",
consumerGroup = "Hsp_Tag_Manager_Order_Create", consumeMode = ConsumeMode.CONCURRENTLY, messageModel = MessageModel.CLUSTERING)
public class ThreeConsumer implements RocketMQListener<MQQuery>{
@Override
public void onMessage(MQQuery message) {
log.info(message.getPayload());
}
}
// 指定为集群消费MessageModel.CLUSTERING，并发消费ConsumeMode.CONCURRENTLY，可以不指定，默认就是这样。
4.6.2.2、广播消费
@Component
@Slf4j
@RocketMQMessageListener(topic = "Hsp_Topic_Manager_Order", selectorExpression = "Hsp_Tag_Manager_Order_Create",
consumerGroup = "Hsp_Tag_Manager_Order_Create", consumeMode = ConsumeMode.CONCURRENTLY,messageModel = MessageModel.BROADCASTING)
public class BroadCastingTwoConsumer implements RocketMQListener<MQQuery>{
@Override
public void onMessage(MQQuery message) {
log.info("========广播消费节点========");
log.info(message.toString());
}
}
4.6.2.3、顺序消费
@Component
@Slf4j
@RocketMQMessageListener(topic = "Hsp_Topic_Manager_Order", selectorExpression = "Hsp_Tag_Manager_Order_Create",
consumerGroup = "Hsp_Tag_Manager_Order_Create", consumeMode = ConsumeMode.ORDERLY, messageModel = MessageModel.CLUSTERING)
public class ThreeConsumer implements RocketMQListener<MQQuery>{
@Override
public void onMessage(MQQuery message) {
log.info(message.getPayload());
}
}
4.6.3、场景使用说明
1、对于需要实时了解调用结果的情况使用同步发送。比如短信通知。（同步生产消息+集群消费）
2、对实时要求不高的接口使用异步发送，如日志记录。（异步生产消息+集群消费）
3、对数据消费有顺序要求时，采用顺序消费，比如：订单的多状态变更 待支付-》支付中-》支付完成。（同步生产消息+顺序消费）
4、对于需要广播通知的消息，采用广播消费，如修改了门店信息，需要通知到各个模块检查逻辑。（同步生产消息+广播消费）
4.6.4、使用前提
-
4.6.4.1、在业务模块bootstrap.yml中添加rocketmq配置
-
4.6.4.2、注入RocketMQTemplate即可使用
-
4.6.4.3、group,topic,tag的定义
4.6.4.1、在业务模块bootstrap.yml中添加rocketmq配置
hsp-rocketmq.yml配置如下
rocketmq:
name-server: http://localhost:9876
access-channel: CLOUD
producer:
group: GID_HSP_LOG_GROUP
send-message-timeout: 5000
access-key: key
secret-key: secret
consumer:
access-key: key
secret-key: secret
4.6.4.2、注入RocketMQTemplate即可使用
4.6.4.3、group,topic,tag的定义
group: 实现消息消费的负载均衡和消息容错，主要用于集群环境下。
topic: 消息主题。比如订单：
tag: 主题标签
命名规则：
TOPIC:
HSP_TOPIC_模块_主题如 HSP_TOPIC_MANAGER_ORDERTAG:
HSP_TAG_模块_标签如 HSP_TAG_MANAGER_ORDER_CREATE, HSP_TAG_MANAGER_ORDER_UPDATEGROUP: 建议与TAG名称相同 如 HSP_TAG_MANAGER_ORDER_CREATE