RocketMQ搭建及整合springboot实例

作者: 任未然 | 来源:发表于2019-10-23 20:44 被阅读0次

一. 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可视化控制台安装

下载地址：https://github.com/apache/rocketmq-externals

进入到 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;">
                        <!--![](image/logo.png)-->
                        <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;
    }
}

RocketMQ搭建及整合springboot实例

一. JMS介绍

二. RocketMQ常见的概念

三. RocketMQ的环境准备

3.1 启动

3.2 rocketmq可视化控制台安装

四. RocketMQ整合Spring boot

4.1 消息生产者的编写

4.1.1 导包

4.1.2 RocketMQ配置

yml编写

RockmqConfig

SendMailComponent

4.1.3 entity层: EmailVo

4.1.4 Controller层 : UserRegsiterController

4.2 消息消费者的编写

4.2.1 导入依赖

4.2.2 配置

yml编写

RockmqConfig

4.2.3 消息接收

五. RocketMQ应用案例

5.1 依赖

5.2 rocketMQ 配置

yml编写

RockmqConfig

5.3 邮件模板代码

5.4 Service层(发送邮件方法)

5.5 消息生产类: Controller层(异步通知消费接收方进行发邮件)

5.6 消费接收类:(检查到订阅主题内容发生变化后触发发送邮件)

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Java技术升华

分布式&高可用