一. JMS介绍
JMS(Java Message Service)，是Java平台中关于面向消息中间件的接口，是一种与厂商无关的API,用来访问消息收发。它的使用场景如下：
- 实现业务的解耦、削峰，异步。
- 跨平台，多语言
- 分布式事务，最终一致性
二. RocketMQ常见的概念
| 名词 | 说明 |
|---|---|
| Producer： | 生产消息的服务。 |
| Producer Group: | 消息生产者组，发送同类消息的一个消息生产组。 |
| Consumer: | 消费消息的服务。 |
| Consumber Group: | 消费同类消息的多个实例。 |
| Topic: | 主题，某类消息，queue是消息的物理管理单位，而topic是逻辑管理单位，一个topic下有多个queue，默 认4个。 |
| Tag: | 标签，子主题(二级分类)，对Topic的进一步细化，用于区分同个主题下不同的业务消息。 |
| Message: | 消息，每个message必须指定一个topic |
| Broker: | MQ程序，接收生产的消息，提供给消费者的程序。 |
| Name Server: | 给生产者和消费者提供路由信息，提供轻量级的服务发现、路由、元数据信息 |
| Offset: | 偏移量，可以理解为消息进度 |
| Commit log: | 消息存储会写在 commit log文件中 |
三. RocketMQ的环境准备
RocketMQ的下载地址：http://rocketmq.apache.org/release_notes/release-notes-4.4.0/
3.1 启动
rocketmq是用纯粹的Java语言编写，所以首先需要安装Jdk.
- windows系统下启动：先启动
mqnamesrv.cmd, 然后再启动mqbroker.cmd -n localhost:9876 - Linux/Mac操作系统下启动：先切换到root用户，启动
sh mqnamesrv.sh, 然后再启动sh mqbroker -n localhost:9876
3.2 rocketmq可视化控制台安装
进入到 rocketmq-console 文件夹下，执行如下命令：
mvn clean package -Dmaven.test.skip=true
java -jar target/rocketmq-console-ng-1.0.1.jar
在浏览器输入：http://localhost:8080进行访问。
四. RocketMQ整合Spring boot
4.1 消息生产者的编写
4.1.1 导包
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.4.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
4.1.2 RocketMQ配置
yml编写
rocketmq:
#指定ip
nameserverAddr: localhost:9876
#主题名
topicName: send-email-topic
#标签名
topicTag: tag1
#消息生产组名
producerGroupName: sendMailProducer
#消息消费组名
consumerGroupName: sendMailConsumer
RockmqConfig
@Configuration
@ConfigurationProperties(prefix = "rocketmq")
public class RockmqConfig {
private String nameserverAddr;
private String topicName;
private String topicTag;
private String producerGroupName;
private String consumerGroupName;
public String getNameserverAddr() {
return nameserverAddr;
}
public void setNameserverAddr(String nameserverAddr) {
this.nameserverAddr = nameserverAddr;
}
public String getTopicName() {
return topicName;
}
public void setTopicName(String topicName) {
this.topicName = topicName;
}
public String getTopicTag() {
return topicTag;
}
public void setTopicTag(String topicTag) {
this.topicTag = topicTag;
}
public String getProducerGroupName() {
return producerGroupName;
}
public void setProducerGroupName(String producerGroupName) {
this.producerGroupName = producerGroupName;
}
public String getConsumerGroupName() {
return consumerGroupName;
}
public void setConsumerGroupName(String consumerGroupName) {
this.consumerGroupName = consumerGroupName;
}
}
SendMailComponent
@Configuration
@Import(RockmqConfig.class)
public class SendMailComponent {
@Autowired
private RockmqConfig rockmqConfig;
// 消息的生产者
@Bean
public DefaultMQProducer defaultMQProducer() {
//消息生成者
DefaultMQProducer defaultMQProducer = new DefaultMQProducer();
// 设置NameServerw
defaultMQProducer.setNamesrvAddr(rockmqConfig.getNameserverAddr());
//设置生产者组
defaultMQProducer.setProducerGroup(rockmqConfig.getProducerGroupName());
try {
defaultMQProducer.start(); //开启生产者组
} catch (MQClientException e) {
e.printStackTrace();
}
return defaultMQProducer;
}
}
4.1.3 entity层: EmailVo
public class EmailVo {
private String name; // 名字
private String email; // 邮箱
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getEmail() {
return email;
}
public void setEmail(String email) {
this.email = email;
}
}
4.1.4 Controller层 : UserRegsiterController
//import org.apache.rocketmq.client.producer.DefaultMQProducer;
//import org.apache.rocketmq.client.producer.MessageQueueSelector;
//import org.apache.rocketmq.common.message.Message;
//import org.apache.rocketmq.common.message.MessageQueue;
@RestController
@RequestMapping("/reg")
@Import(RockmqConfig.class)
public class UserRegsiterController {
@Autowired
private DefaultMQProducer defaultMQProducer;
@Autowired
private ObjectMapper objectMapper;
@Autowired
private RockmqConfig rockmqConfig;
@RequestMapping
public Object register(String email, String name) throws Exception {
EmailVo emailVo = new EmailVo();//实体类
// 获取唯一ID
String uuid = UUID.randomUUID().toString();
emailVo.setName(uuid); // 设置名字
emailVo.setEmail(email);// 设置邮箱账号
Message message = new Message();//消息对象
message.setBody(objectMapper.writeValueAsString(emailVo).getBytes());//设置主题内容
message.setTopic(rockmqConfig.getTopicName());//设置主题名
message.setTags(rockmqConfig.getTopicTag());// 设置标签
// defaultMQProducer.send(message); //消息的发布, 进入到不同的队列，不能保证有序
// 声明单号
String sFId = "SF109387733";
defaultMQProducer.send(message, new MessageQueueSelector() {
/**
* mqs: 表示队列，同一个主题下所有的队列
*/
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
System.out.println(mqs.size()); // 4, 表示同一个主题下有4个队列
System.out.println(arg); // SF109387733
int maxIndex = mqs.size() - 1;
int index = maxIndex & arg.hashCode(); // 按位与，或、异或、非
return mqs.get(index);
}
}, sFId);
return "success";
}
}
4.2 消息消费者的编写
4.2.1 导入依赖
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.4.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
4.2.2 配置
yml编写
crocketmq:
nameserverAddr: localhost:9876
topicName: send-email-topic
topicTag: tag1
producerGroupName: sendMailProducer
consumerGroupName: sendMailConsumer
RockmqConfig
@Configuration
@ConfigurationProperties(prefix = "rocketmq")
public class RockmqConfig {
private String nameserverAddr;
private String topicName;
private String topicTag;
private String producerGroupName;
private String consumerGroupName;
public String getNameserverAddr() {
return nameserverAddr;
}
public void setNameserverAddr(String nameserverAddr) {
this.nameserverAddr = nameserverAddr;
}
public String getTopicName() {
return topicName;
}
public void setTopicName(String topicName) {
this.topicName = topicName;
}
public String getTopicTag() {
return topicTag;
}
public void setTopicTag(String topicTag) {
this.topicTag = topicTag;
}
public String getProducerGroupName() {
return producerGroupName;
}
public void setProducerGroupName(String producerGroupName) {
this.producerGroupName = producerGroupName;
}
public String getConsumerGroupName() {
return consumerGroupName;
}
public void setConsumerGroupName(String consumerGroupName) {
this.consumerGroupName = consumerGroupName;
}
}
4.2.3 消息接收
@Configuration
@Import(com.qf.config.RockmqConfig.class)
public class MessageConsumerCompoment {
@Autowired
private RockmqConfig rockmqConfig;
@Autowired
private ObjectMapper objectMapper;
@Bean
public DefaultMQPushConsumer defaultMQPushConsumer() throws MQClientException {
//消息接受者
DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
//设置ConsumerGroup
defaultMQPushConsumer.setConsumerGroup(rockmqConfig.getConsumerGroupName());
//设置Nameserve
defaultMQPushConsumer.setNamesrvAddr(rockmqConfig.getNameserverAddr());
//设置主题与主题下的标签
defaultMQPushConsumer.subscribe(rockmqConfig.getTopicName(), "tag1");
// 开始接收消息，当订阅主题内容发生变化，本方法就会执行
defaultMQPushConsumer.registerMessageListener((List<MessageExt> msgs, ConsumeConcurrentlyContext context) -> {
//遍历消息队列
msgs.forEach(mt -> {
String body = new String(mt.getBody());
System.out.println(body);
try {
//把消息内容取出转换为实体对象
EmailVo emailVo = objectMapper.readValue(body, EmailVo.class);
String uuid = emailVo.getName();
// 到数据库查询是否有该uuid, 如果有, 证明该消息已处理过, 直接return ConsumeConcurrentlyStatus.CONSUME_SUCCESS
// 没有接着进行其他的操作。
// 可能会有并发问题。读写锁()
} catch (IOException e) {
e.printStackTrace();
}
});
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
defaultMQPushConsumer.start();
return defaultMQPushConsumer;
}
}
五. RocketMQ应用案例
mq最典型的应用就是邮件系统，当用户注册的时候，给用户发送注册成功的邮件，接下来就演示该案例。在该案例中使用的是thymeleaf模板引擎。
5.1 依赖
<!-- rocketmq -->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.4.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- thymeleaf依赖 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-thymeleaf</artifactId>
</dependency>
<!-- 邮件依赖 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-mail</artifactId>
</dependency>
5.2 rocketMQ 配置
yml编写
spring:
thymeleaf:
mode: html #指开发的thymeleaf的文件格式
enabled: true #开启thymeleaf
prefix: classpath:/templates/ #前缀
suffix: .html #后缀
mail:
password: XXXXXXXX
host: smtp.126.com
default-encoding: utf-8
username: XXXXXX@126.com
rocketmq:
nameserverAddr: localhost:9876
topicName: send-email-topic
topicTag: tag1
producerGroupName: sendMailProducer
consumerGroupName: sendMailConsumer
RockmqConfig
@Configuration
@ConfigurationProperties(prefix = "rocketmq")
public class RockmqConfig {
private String nameserverAddr;
private String topicName;
private String topicTag;
private String producerGroupName;
private String consumerGroupName;
public String getNameserverAddr() {
return nameserverAddr;
}
public void setNameserverAddr(String nameserverAddr) {
this.nameserverAddr = nameserverAddr;
}
public String getTopicName() {
return topicName;
}
public void setTopicName(String topicName) {
this.topicName = topicName;
}
public String getTopicTag() {
return topicTag;
}
public void setTopicTag(String topicTag) {
this.topicTag = topicTag;
}
public String getProducerGroupName() {
return producerGroupName;
}
public void setProducerGroupName(String producerGroupName) {
this.producerGroupName = producerGroupName;
}
public String getConsumerGroupName() {
return consumerGroupName;
}
public void setConsumerGroupName(String consumerGroupName) {
this.consumerGroupName = consumerGroupName;
}
}
5.3 邮件模板代码
<!DOCTYPE html>
<html lang="en"
xmlns="http://www.w3.org/1999/xhtml"
xmlns:th="http://www.thymeleaf.org">
<head>
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8"/>
<meta name="viewport" content="width=device-width, initial-scale=1.0"/>
<title>Siemens Sinnovation</title>
<style>
.button {
background-color: #4CAF50;
border-radius: 12px;
border: none;
color: white;
padding: 10px 25px;
text-align: center;
text-decoration: none;
display: inline-block;
font-size: 18px;
margin: 4px 2px;
cursor: pointer;
box-shadow: 0 8px 16px 0 rgba(0, 0, 0, 0.2), 0 6px 20px 0 rgba(0, 0, 0, 0.19);
}
.button:hover {
box-shadow: 0 12px 16px 0 rgba(0, 0, 0, 0.24), 0 17px 50px 0 rgba(0, 0, 0, 0.19);
}
</style>
</head>
<body style="margin: 0;padding: 0;">
<table align="center" border="1" cellpadding="0" cellspacing="0" width="600px">
<tr>
<td>
<table align="center" border="0" cellpadding="0" cellspacing="0" width="600"
style="border-collapse: collapse;">
<tr>
<td align="center" style="padding: 40px 0 30px 0;">
<!---->
<img th:src="@{'' + ${imageResourceName}}">
</td>
</tr>
<tr>
<td bgcolor="#ffffff" style="padding: 20px 30px 20px 30px">
<h4>The following message was created by <span th:text="${username}"></span> in the
Siemens DFFA group: </h4>
<table border="0" cellpadding="0" cellspacing="0" width="100%">
<tr>
<td><span th:text="${title}">title</span></td>
</tr>
<tr>
<td style="padding: 20px 0 30px 0">
<span th:text="${description}">description</span>
</td>
</tr>
<tr>
<table border="0" cellpadding="0" cellspacing="0" width="100%">
<tr>
<td align="center" style="padding: 5px 0 3px 0">Status:<span
th:text="${status}"></span></td>
<td align="center" style="padding: 5px 0 3px 0">Date submitted: 28/08/2017 <span
th:text="${createDate}"></span></td>
<td align="center" style="padding: 5px 0 3px 0">Days left to join: 10h<span
th:text="${leftTime}"></span></td>
</tr>
</table>
</tr>
<tr>
<td align="center" style="padding: 5px 40px 5px 40px">
<button class="button">View Details</button>
</td>
</tr>
</table>
</td>
</tr>
</table>
</td>
</tr>
</table>
</body>
</html>
5.4 Service层(发送邮件方法)
/**
* 处理发送邮件
*/
@Service
public class SendEmail {
// 模板引擎, 目的是在邮件模板中取值
@Autowired
private TemplateEngine templateEngine;
// 发送邮件
@Autowired
private JavaMailSender javaMailSenderImpl;
/**
* @param from 邮件的发送方
* @param to 邮件发送给谁
* @param subject 主题
* @param map 数据部分
*/
public void sendMail(String from, String to, String subject, Map<String, Object> map) {
MimeMessage mimeMessage = javaMailSenderImpl.createMimeMessage();
try {
// 设置信息
MimeMessageHelper messageHelper = new MimeMessageHelper(mimeMessage, true);
messageHelper.setFrom(from);
messageHelper.setTo(to);
messageHelper.setSubject(subject);
Context context = new Context();
context.setVariables(map); //要填充到模板的数据
String emailContent = templateEngine.process("email", context);
// 第二个参数，是否为html
messageHelper.setText(emailContent, true);
javaMailSenderImpl.send(mimeMessage);
} catch (Exception ex) {
ex.printStackTrace();
}
}
}
5.5 消息生产类: Controller层(异步通知消费接收方进行发邮件)
说明: 用户注册成功后, 通过消息队列发送邮件
@RestController
@RequestMapping("/register")
public class RegisterController {
@Autowired
private DefaultMQProducer defaultMQProducer;
@Autowired
private ProducerConfig producerConfig;
@RequestMapping
public Object register(String email) {
/**
* 构建消息，第一个参数是主题，第二个参数是tag, 第三个参数是消息体部分
*/
Message message = new Message(producerConfig.getTopic(), producerConfig.getTag(), email.getBytes());
//Message message = new Message("email-test-topic", "notice", email.getBytes());
try {
defaultMQProducer.send(message); // 消息的生产者生产的消息
} catch (MQClientException e) {
e.printStackTrace();
} catch (RemotingException e) {
e.printStackTrace();
} catch (MQBrokerException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
return "注册成功";
}
}
5.6 消费接收类:(检查到订阅主题内容发生变化后触发发送邮件)
@Configuration
@Import(com.qf.config.RockmqConfig.class)
public class MessageConsumerCompoment {
@Autowired
private RockmqConfig rockmqConfig;
@Autowired
private ObjectMapper objectMapper;
@Autowired
private SendEmail sendEmail;
@Bean
public DefaultMQPushConsumer defaultMQPushConsumer() throws MQClientException {
//消息接受者
DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
//设置ConsumerGroup
defaultMQPushConsumer.setConsumerGroup(rockmqConfig.getConsumerGroupName());
//设置Nameserve
defaultMQPushConsumer.setNamesrvAddr(rockmqConfig.getNameserverAddr());
//设置主题与主题下的标签
defaultMQPushConsumer.subscribe(rockmqConfig.getTopicName(), "tag1");
// 开始接收消息，当订阅主题内容发生变化，本方法就会执行
defaultMQPushConsumer.registerMessageListener((List<MessageExt> msgs, ConsumeConcurrentlyContext context) -> {
//遍历消息队列
msgs.forEach(mt -> {
String body = new String(mt.getBody());
System.out.println(body);
try {
//把消息内容取出转换为实体对象
//EmailVo emailVo = objectMapper.readValue(body, EmailVo.class);
//String uuid = emailVo.getName();
// 到数据库查询是否有该uuid, 如果有, 证明该消息已处理过, 直接return ConsumeConcurrentlyStatus.CONSUME_SUCCESS
// 没有接着进行发送邮件操作
//sendEmail. sendEmail();(略)
} catch (IOException e) {
e.printStackTrace();
}
});
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
defaultMQPushConsumer.start();
return defaultMQPushConsumer;
}
}
