我们整合boot项目的时候都是引入 xxx-start 依赖，但是现在大多数的整合RocketMQ都还不是这样。
我花了一天时间使用rocketmq-spring-boot-starter整合，使得操作起来更加简单。
1、说明
1-1：rocketmq-spring-boot-starter 提供了一个 rocketMQTemplate 使得发消息更加简单，它底层也还是基于DefaultMQProducer 但是封装之后使用更加方便了。
1-2：rocketmq-spring-boot-starter使用监听器的方式获取消息更加的简单
RocketMQ_10">1-3：下面代码提供三种基础的消息发送测试【同步、异步、单向】你只需要修改yml文件里面的RocketMQ地址就好了
1-4：更多内容可以查看官网文档 https://github.com/apache/rocketmq-spring
2、代码
2-1：pom
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>www.xdx97.com</groupId>
<artifactId>RocketMQDemo</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<!-- Spring Boot 依赖 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot</artifactId>
<version>2.2.0.RELEASE</version>
</dependency>
<!-- SpringBoot Web -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<version>2.2.2.RELEASE</version>
</dependency>
<!-- RocketMQ -->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.2</version>
</dependency>
<!-- RocketMQ -->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.2</version>
</dependency>
</dependencies>
</project>
2-1：ProducerController
java">import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* 生产者
*/
@RestController
public class ProducerController {
@Autowired
private RocketMQTemplate rocketMQTemplate;
/**
* 同步消息
*
* @author 小道仙
* @date 2020年3月3日
*/
@GetMapping("/syncNews")
public String syncNews(){
Message message = new Message();
message.setBody("同步消息".getBytes());
SendResult sendResult = rocketMQTemplate.syncSend("sync-topic", message);
// 同步消息发送成功会有一个返回值，我们可以用这个返回值进行判断和获取一些信息
System.out.println(sendResult);
return "success";
}
/**
* 异步消息
*
* @author 小道仙
* @date 2020年3月3日
*/
@GetMapping("/asynNews")
public String asynNews(){
Message message = new Message();
message.setBody("异步消息".getBytes());
rocketMQTemplate.asyncSend("asyn-topic", message, new SendCallback() {
public void onSuccess(SendResult sendResult) {
// 成功回调
}
public void onException(Throwable throwable) {
// 失败回调
}
});
return "success";
}
/**
* 单向消息
*
* @author 小道仙
* @date 2020年3月3日
*/
@GetMapping("/sendOneWay")
public String sendOneWay(){
Message message = new Message();
message.setBody("单向消息".getBytes());
rocketMQTemplate.sendOneWay("oneWay-topic",message);
return "success";
}
}
2-1：AsynConsumer
java">import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;
/**
* 异步消息
*/
@Service
@RocketMQMessageListener(consumerGroup = "my-consumer_asyn-topic", topic = "asyn-topic")
public class AsynConsumer implements RocketMQListener<MessageExt> {
public void onMessage(MessageExt messageExt) {
byte[] body = messageExt.getBody();
System.out.println(new String(body));
}
}
2-1：OneWayConsumer
java">import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;
/**
* 单向消息
*/
@Service
@RocketMQMessageListener(consumerGroup = "my-consumer_oneWay-topic", topic = "oneWay-topic")
public class OneWayConsumer implements RocketMQListener<MessageExt> {
public void onMessage(MessageExt messageExt) {
byte[] body = messageExt.getBody();
System.out.println(new String(body));
}
}
2-1：SyncConsumer
java">import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;
/**
* 同步消费
*/
@Service
@RocketMQMessageListener(consumerGroup = "my-consumer_sync-topic", topic = "sync-topic")
public class SyncConsumer implements RocketMQListener<MessageExt> {
public void onMessage(MessageExt messageExt) {
byte[] body = messageExt.getBody();
System.out.println(new String(body));
}
}
RocketMQApp_216">2-2：RocketMQApp
java">import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class RocketMQApp {
public static void main(String[] args) {
SpringApplication.run(RocketMQApp.class, args);
}
}
2-2：application
server:
port: 80
rocketmq:
name-server: xxxxxxx:9876
producer:
send-message-timeout: 300000
group: rocketmq-group
3、测试
3-1：http://localhost/syncNews
3-2：http://localhost/asynNews
3-3：http://localhost/sendOneWay
ps：我们会看到打印的message信息，你可以使用JSON去格式化获取自己想要的信息。消息有短暂的延迟。
RocketMQDemo_258">4、如果需要源码可以关注微信公众号回复: RocketMQDemo
