Spring Boot整合RocketMQ
-
pom.xml导入RocketMQ依赖
<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.2</version> </dependency> -
application.yml中添加配置
rocketmq:name-server: 127.0.0.1:9876producer:group: xaccess-key: myaccesskeysecret-key: mysecretKeysend-message-timeout: 10000tls-enable: trueconsumer:group: xaccess-key: myaccesskeysecret-key: mysecretKeytls-enable: true -
创建MQ工具类
public class MqUtil {private final RocketMQTemplate rocketMQTemplate;public MqUtil(RocketMQTemplate rocketMQTemplate) {this.rocketMQTemplate = rocketMQTemplate;}/*** 单条通知发送** @param topic 主题* @param message 消息*/public void convertAndSend(String topic, Object message) {rocketMQTemplate.convertAndSend(topic, message);}/*** 批量通知发送** @param topic 主题* @param messages 消息集合*/public <T extends Message<?>> SendResult syncSend(String topic, Collection<T> messages) {return rocketMQTemplate.syncSend(topic, messages);}/*** 批量通知发送** @param topic 主题* @param messages 消息集合* @param sendCallback 回调函数*/public <T extends Message<?>> void asyncSend(String topic, Collection<T> messages, SendCallback sendCallback) {rocketMQTemplate.asyncSend(topic, messages, sendCallback);} } -
注入工具类Bean
@Bean public MqUtil mqUtil(RocketMQTemplate rocketMQTemplate) {return new MqUtil(rocketMQTemplate); } -
测试发消息
@Resource private MqUtil mqUtil;@Test public void test() {mqUtil.convertAndSend(TopicConstant.TOPIC_B, "123456"); } -
订阅接收消息
@Slf4j @Service @RequiredArgsConstructor(onConstructor = @__({@Autowired})) @RocketMQMessageListener(consumerGroup = GroupConstant.GROUP_A, topic = TopicConstant.TOPIC_B) public class TopicAConsumer implements RocketMQListener<Message> {@Overridepublic void onMessage(Message message) {log.info(JsonUtil.toJsonStr(message));} }