pmq再学习三
前面我们已经了解了在做好基础数据的准备工作后，启动测试的时候，会做一个注册消费组的工作，完成后，我们就可以执行生产者发消息操作了。发消息的操作是：
@GetMapping("/test1")
public void test1(@RequestParam String topicName, @RequestParam int count) {
if (Util.isEmpty(topicName))
return;
Executors.newSingleThreadExecutor().submit(new Runnable() {
@Override
public void run() {
for (int i = 1; i < count; i++) {
try {
MqClient.publish(topicName, "", new ProducerDataDto(String.valueOf(i)));
} catch (MqNotInitException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (ContentExceed65535Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
Util.sleep(10);
}
}
});
}发送消息完成后，由于其采用的拉模式，我们可以看到消息在经过发送，存储到数据库之后，会做一个通知拉取数据操作，然后执行拉取。拉取完成后，进行响应。此时会进行消费操作，而这个过程的处理关键是handleData操作，从代码中，我们可以看到其是执行的线程操作是一个batchExcute批量执行操作，可以看到其里面有一个重要方法：threadExcute方法，从而进一步看到我们想看到的方法doMessageReceived，这个方法会调用我们自定义的方法实现消费。
// 执行消费操作
@Override
public List<Long> onMessageReceived(List<MessageDto> messages) {
try {
// 执行消息消费
Transaction catTransaction = null;
System.out.println("开始接收生产者发送过来的消息");
for (MessageDto messageDto : messages) {
// 执行消息消费
System.out.println("当前接收到的消息是messageDto消费组:" + messageDto.getConsumerGroupName() + " " + "当前接收的消费主题：" + messageDto.getTopicName()
+ " " + "当前接收到的消息：" + messageDto.getBody());
}
System.out.println("接收成功");
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return null;
}而从消息的结果可以看到其打印出来的消费消息。
本文参与 腾讯云自媒体分享计划，分享自微信公众号。
原始发表：2021-07-08，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读