Rocketmq 是由阿里巴巴团队开发并捐赠给apache团队的优秀消息中间件,承受过历年双十一大促的考验。
你可以通过本项目轻松的集成Rocketmq到你的SpringBoot项目中。 本项目主要包含以下特性
- 同步发送消息
- 异步发送消息
- 广播发送消息
- 有序发送和消费消息
- 发送延时消息
- 消息tag和key支持
- 自动序列化和反序列化消息体
- 消息的实际消费方IP追溯
- ...
-
发送即忘消息(可能由于直接抛弃所有异常导致消息静默丢失,弃用) -
拉取方式消费(配置方式复杂,位点可能发生偏移,弃用)
<dependency>
<groupId>com.maihaoche</groupId>
<artifactId>spring-boot-starter-rocketmq</artifactId>
<version>0.0.7</version>
</dependency>
spring:
rocketmq:
name-server-address: 172.21.10.111:9876
# 可选, 如果无需发送消息则忽略该配置
producer-group: local_pufang_producer
# 发送超时配置毫秒数, 可选, 默认3000
send-msg-timeout: 5000
# 追溯消息具体消费情况的开关,默认打开
#trace-enabled: false
# 是否启用VIP通道,默认打开
#vip-channel-enabled: false
在springboot应用主入口添加@EnableMQConfiguration
注解开启自动装配:
@SpringBootApplication
@EnableMQConfiguration
class DemoApplication {
}
通过我们提供的Builder
类创建消息对象,详见wiki
MessageBuilder.of(new MSG_POJO()).topic("some-msg-topic").build();
详见wiki:
@MQProducer
public class DemoProducer extends AbstractMQProducer{
}
详见wiki:
支持配置项解析,如存在suclogger-test-cluster
配置项,会优先将topic解析为配置项对应的值。
@MQConsumer(topic = "suclogger-test-cluster", consumerGroup = "local_sucloger_dev")
public class DemoConsumer extends AbstractMQPushConsumer {
@Override
public boolean process(Object message, Map extMap) {
// extMap 中包含messageExt中的属性和message.properties中的属性
System.out.println(message);
return true;
}
}
// 注入发送者
@Autowired
private DemoProducer demoProducer;
...
// 发送
demoProducer.syncSend(msg)