SpringBoot集成RocketMQ
添加pom.xml依赖
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.3</version>
</dependency>创建消息消费者
@Component
@Slf4j
public class MessageConsumerService {
@Component
@RocketMQMessageListener(topic = RocketMQConstant.TOPIC, consumerGroup = "consumer-group1")
public class Consumer1 implements RocketMQListener<UserChange> {
@Override
public void onMessage(UserChange message) {
log.info("收到信息：{}", JSON.toJSONString(message));
}
}
@Component
@RocketMQMessageListener(topic = RocketMQConstant.TOPIC, consumerGroup = "consumer-group2")
public class Consumer2 implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
log.info("收到信息：{}", message);
}
}
@Component
@RocketMQMessageListener(topic = RocketMQConstant.TOPIC, consumerGroup = "consumer-group3")
public class Consumer3 implements RocketMQListener<MessageExt> {
@Override
public void onMessage(MessageExt message) {
log.info("收到信息：{}", new String(message.getBody()));
}
}
}发送消息
@RestController
public class TestController {
/**
* destination包括2个部分信息，topic和tags，可以只有topic
*/
private final String destination = RocketMQConstant.TOPIC + ":" + RocketMQConstant.TAGS;
@Autowired
private RocketMQTemplate rocketMQTemplate;
/**
* 同步消息发送
*
* @return
*/
@GetMapping("send")
public SendResult send() {
UserChange change = UserChange.builder().userName("张三").remark("密码变更").build();
Message<UserChange> message = MessageBuilder.withPayload(change).setHeader(RocketMQHeaders.KEYS, "key").build();
return rocketMQTemplate.syncSend(destination, message);
}
}日志输出
2023-05-06 19:45:15.042 [ConsumeMessageThread_consumer-group2_1] INFO com.zhi.demo.consumer.MessageConsumerService$Consumer2.onMessage 38 -- 收到信息：{"userName":"张三","remark":"密码变更"}
2023-05-06 19:45:15.042 [ConsumeMessageThread_consumer-group3_1] INFO com.zhi.demo.consumer.MessageConsumerService$Consumer3.onMessage 47 -- 收到信息：{"userName":"张三","remark":"密码变更"}
2023-05-06 19:45:15.080 [ConsumeMessageThread_consumer-group1_1] INFO com.zhi.demo.consumer.MessageConsumerService$Consumer1.onMessage 29 -- 收到信息：{"remark":"密码变更","userName":"张三"}
常见错误：
1、connect to 172.17.183.41:10911 failed
防火墙需要开启10911端口
firewall-cmd --zone=public --add-port=10911/tcp --permanent
firewall-cmd --reload2、sendDefaultImpl call timeout
消息发送超时，可调整rocketmq.producer.send-message-timeout参数，默认3秒