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

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

#### 文章目录 ####

*  1、整合
 *  2、消息的生产
 *  3、消费
 *  4、发送异步消息
 *  5、补充：安装RocketMQ

![在这里插入图片描述][b1608333385e41a3a3de5277faf660ef.png]

## 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层来演示：

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