【开发篇】二十、SpringBoot整合RocketMQ-蒲公英云

【开发篇】二十、SpringBoot整合RocketMQ

港控/mmm° 2023-10-16 18:11 83阅读 0赞

文章目录

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);      
    }
}