文章目录
- 1、整合
- 2、消息的生产
- 3、消费
- 4、发送异步消息
- 5、补充:安装RocketMQ
1、整合
首先导入起步依赖,RocketMQ的starter不是Spring维护的,这一点从starter的命名可以看出来(不是spring-boot-starter-xxx,而是xxx-spring-boot-starter,和MyBatisPlus、Druid一样),因此version值得自己加:
<dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-spring-boot-starter</artifactId> <version>2.2.1</version>
</dependency>
添加相关配置:
rocketmq: name-server: localhost:9876 producer: group: group_rocketmq # 设置一个自定义的生产者默认组名,省掉这个启动会报错
在需要的地方注入RocketMQTemplate操作对象:
@Autowired
private RocketMQTemplate rocketMQTemplate;
2、消息的生产
发送消息继续convertAndSend方法,接着上篇在Service层来演示:
@Service
@Slf4j
public class MessageServiceRocketmqImpl implements MessageService { @Autowired private RocketMQTemplate rocketMQTemplate; @Override public void sendMessage(String id) { rocketMQTemplate.convertAndSend("order_sm_id",id); log.info("使用Rabbitmq将待发送短信的订单纳入处理队列,id:"+id); }
}
convertAndSend方法依旧重载,可以直接传一个Object,也可以先传一个destination参数,即发到哪儿,再传要发的message
3、消费
这里不演示手动receive方法拿消息,直接用监听器自动拿来消费:实现RocketMQListener接口,泛型为Message类型
,重写onMessage方法,加@RocketMQMessageListener
注解,两个属性为主题名称和消费者组
@Component
@Slf4j
@RocketMQMessageListener(topic="order_sm_id",consumerGroup = "group_rocketmq")
public class RocketmqMessageListener implements RocketMQListener<String> { @Override public void onMessage(String id) { log.info("已完成短信发送业务,id:"+id); }
}
4、发送异步消息
@Service
@Slf4j
public class MessageServiceRocketmqImpl implements MessageService { @Autowired private RocketMQTemplate rocketMQTemplate; @Override public void sendMessage(String id) { //回调逻辑SendCallback callback = new SendCallback() { @Override public void onSuccess(SendResult sendResult) { //消息发送成功后你要做的业务//...log.info("消息发送成功"); } @Override public void onException(Throwable throwable) { log.info("消息发送失败!!!!!!!!!!!"); } }; //异步发送rocketMQTemplate.asyncSend("order_sm_id",id,callback); log.info("使用Rabbitmq将待发送短信的订单纳入处理队列,id:"+id); }
}
asyncSend异步发消息,有个参数是callback回调方法,类型是一个接口,创建这个对象的时候重写onSuccess和OnException
方法,即消息发送成功以后的逻辑和消息发送失败以后的逻辑(异步的体现,不用等,来个回调)。
5、补充:安装RocketMQ
建议以Docker方式启动,下面备份下在Windows的安装(安装为一个系统服务):
- 下载
下载地址:https://rocketmq.apache.org/
- 安装:解压缩即可
默认服务端口:9876
- 环境变量配置
ROCKETMQ_HOME
PATH
NAMESRV_ADDR (建议): 127.0.0.1:9876
- 启动命名服务器:
mqnamesrv
- 启动Broker
mqbroker
- 服务器功能测试:生产数据
tools org.apache.rocketmq.example.quickstart.Producer
- 服务器功能测试:消费数据
tools org.apache.rocketmq.example.quickstart.Consumer