【开发篇】二十、SpringBoot整合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