(给ImportNew加星标,提高Java技能)
转自:WilsonHe
链接:juejin.im/post/5ea159e4f265da47f0794da5
rocketmq-spring-boot-starter
相关类:
RocketMQListener
接口:消费者都需实现该接口的消费方法
onMessage(msg)
。
RocketMQPushConsumerLifecycleListener
接口:当
@RocketMQMessageListener
中的配置不足以满足我们的需求时,可以实现该接口直接更改消费者类
DefaultMQPushConsumer
配置
@RocketMQMessageListener
:被该注解标注并实现了接口
RocketMQListener
的bean为一个消费者并监听指定topic队列中的消息,该注解中包含消费者的一些常用配置(大部分按默认即可),一般只需更改consumerGroup(消费组)与topic。
RocketMQMessageListener
中的属性配置是可以使用Placeholder(占位符)从配置文件或配置中心获取的,如下图:
文章例子环境:1NameServer + 2Broker + 1Consumer
添加maven依赖<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
</dependency>
复制代码
application.yml配置
rocketmq:
name-server: 127.0.0.1:9876
producer:
group: praise-group
server:
port: 10000
spring:
datasource:
driver-class-name: com.mysql.cj.jdbc.Driver
username: root
password: tiger
url: jdbc:mysql://localhost:3306/wilson
swagger:
docket:
base-package: io.rocket.consumer.controller
复制代码
点赞接口
PraiseRecord(点赞记录):
@Data
public class PraiseRecord implements Serializable {
private Long id;
private Long uid;
private Long liveId;
private LocalDateTime createTime;
}
复制代码
MessageController(简单的测试接口):
RestController
@RequestMapping("/message")
public class MessageController {
@Resource
private RocketMQTemplate rocketMQTemplate;
@PostMapping("/praise")
public ServerResponse praise(@RequestBody PraiseRecordVO vo) {
rocketMQTemplate.sendOneWay(RocketConstant.Topic.PRAISE_TOPIC, MessageBuilder.withPayload(vo).build());
return ServerResponse.success();
}
// ......
}
复制代码
sendOneyWay()
进行消息发送。
@Service
@RocketMQMessageListener(topic = RocketConstant.Topic.PRAISE_TOPIC, consumerGroup = RocketConstant.ConsumerGroup.PRAISE_CONSUMER)
@Slf4j
public class PraiseListener implements RocketMQListener<PraiseRecordVO>, RocketMQPushConsumerLifecycleListener {
@Resource
private PraiseRecordService praiseRecordService;
@Override
public void onMessage(PraiseRecordVO vo) {
praiseRecordService.insert(vo.copyProperties(PraiseRecord::new));
}
@Override
public void prepareStart(DefaultMQPushConsumer consumer) {
// 每次拉取的间隔,单位为毫秒
consumer.setPullInterval(2000);
// 设置每次从队列中拉取的消息数为16
consumer.setPullBatchSize(16);
}
}
MessageStoreConfig.maxTransferCountOnMessageInMemory
(默认为32)值限制,即若想要消费者从队列拉取的消息数大于32有效(pullBatchSize>32)则需更改Broker的启动参数
maxTransferCountOnMessageInMemory
值。在MQ削峰的配置参数里,以下几个
DefaultMQPushConsumer
的参数是需要注意一下的:
EachPullTotal=所有Broker上的写队列数和(writeQueueNums=readQueueNums) * pullBatchSize
DefaultMQPushConsumer
并配置其
consumeMessageBatchMaxSize
属性。
consumeMessageBatchMaxSize
属性默认值为1,即每次只消费一条消息,需要注意的是该属性也会受
pullBatchSize
影响,如果
consumeMessageBatchMaxSize
为32但
pullBatchSize
只为12,那么每次批量消费的最大消息数也就只有12。如下为个人测试批量消费Consumer的测试bean:
@Bean
public DefaultMQPushConsumer userMQPushConsumer() throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketConstant.ConsumerGroup.SPRING_BOOT_USER_CONSUMER);
consumer.setNamesrvAddr(nameServer);
consumer.subscribe(RocketConstant.Topic.SPRING_BOOT_USER_TOPIC, "*");
// 设置每次消息拉取的时间间隔,单位毫秒
consumer.setPullInterval(1000);
// 设置每个队列每次拉取的最大消息数
consumer.setPullBatchSize(24);
// 设置消费者单次批量消费的消息数目上限
consumer.setConsumeMessageBatchMaxSize(12);
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context)
-> {
List<UserInfo> userInfos = new ArrayList<>(msgs.size());
Map<Integer, Integer> queueMsgMap = new HashMap<>(8);
msgs.forEach(msg -> {
userInfos.add(JSONObject.parseObject(msg.getBody(), UserInfo.class));
queueMsgMap.compute(msg.getQueueId(), (key, val) -> val == null ? 1 : ++val);
});
log.info("userInfo size: {}, content: {}", userInfos.size(), userInfos);
/*
处理批量消息,如批量插入:userInfoMapper.insertBatch(userInfos);
*/
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();
return consumer;
}
consumeMessageBatchMaxSize
与
pullBatchSize
,且
pullBatchSize
较小,所以每次消费的消息数最大值为12,如下图:
mqnamesrv -n 127.0.0.1:9876
mqbroker -c E:\RocketMQ\rocketmq-all-4.5.2-bin-release\bin\2m-noslave\broker-a.properties
mqbroker -c E:\RocketMQ\rocketmq-all-4.5.2-bin-release\bin\2m-noslave\broker-b.properties
一个基于 RabbitMQ 的可复用的分布式事务消息架构方案!
哥们五年多的开发经验,面试栽在了MQ上,但面试官的回答真的牛逼!
RabbitMQ 消费端限流、TTL、死信队列
看完本文有收获?请转发分享给更多人
关注「ImportNew」,提升Java技能
好文章,我在看❤️