RocketMQ-消息队列

消息队列概述

消息队列中间件是分布式系统中重要的组件，主要解决应用解耦，异步消息，流量削锋等问题，实现高性能，高可用，可伸缩和最终一致性架构。目前使用较多的消息队列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ

核心概念

Topic：消息主题，一级消息类型，生产者向其发送消息。
生产者：也称为消息发布者，负责生产并发送消息至Topic。
消费者：也称为消息订阅者，负责从Topic接收并消费消息。
消息：生产者向Topic发送并最终传送给消费者的数据和（可选）属性的组合。
消息属性：生产者可以为消息定义的属性，包含Message Key和Tag。
Group：一类生产者或消费者，这类生产者或消费者通常生产或消费同一类消息，且消息发布或订阅的逻辑一致。

应用场景

消息队列作为高并发系统的核心组件之一，能够帮助业务系统解构提升开发效率和系统稳定性。主要具有以下优势：

削峰填谷（主要解决瞬时写压力大于应用服务能力导致消息丢失、系统奔溃等问题）
诸如秒杀、抢红包、企业开门红等大型活动时皆会带来较高的流量脉冲，或因没做相应的保护而导致系统超负荷甚至崩溃，或因限制太过导致请求大量失败而影响用户体验，消息队列RocketMQ版可提供削峰填谷的服务来解决该问题。
异步解耦（解决不同重要程度、不同能力级别系统之间依赖导致一死全死）
交易系统作为淘宝和天猫主站最核心的系统，每笔交易订单数据的产生会引起几百个下游业务系统的关注，包括物流、购物车、积分、流计算分析等等，整体业务系统庞大而且复杂，消息队列RocketMQ版可实现异步通信和应用解耦，确保主站业务的连续性。
提升性能（当存在一对多调用时，可以发一条消息给消息系统，让消息系统通知相关系统）
蓄流压测（线上有些链路不好压测，可以通过堆积一定量消息再放开来压测）

适用场景

消息队列RocketMQ版作为分布式系统中的重要组件，可用于应对这些挑战，例如解决应用的异步解耦。

异步解耦

传统处理

最常见的一个场景是用户注册后，需要发送注册邮件和短信通知，以告知用户注册成功。传统的做法有以下两种：

串行方式
串行方式下的注册流程如下图所示。

数据流动如下所述：
1. 您在注册页面填写账号和密码并提交注册信息，这些注册信息首先会被写入数据库。
2. 注册信息写入数据库成功后，再发送请求至邮件注册。邮件注册收到请求后向用户发送邮件通知。
3. 注册邮件接收注册请求后再向下游的短信注册发送请求。短信注册收到请求后向用户发送短信通知。
以上三个任务全部完成后，才返回注册结果到客户端，用户才能使用账号登录。
假设每个任务耗时分别为50 ms，则用户需要在注册页面等待总共150 ms才能登录。
并行方式
并行方式下的注册流程如下图所示。

数据流动如下所述：

用户在注册页面填写账号和密码并提交注册信息，这些注册信息首先会被写入数据库。
注册信息写入数据库成功后，再同时发送请求至邮件注册和短信注册。邮件和短信注册收到请求后分别向用户发送邮件和短信通知。

以上两个任务全部完成后，才返回注册结果到客户端，用户才能使用账号登录。

假设每个任务耗时分别为50 ms，其中，邮件和短信通知并行完成，则用户需要在注册页面等待总共100 ms才能登录。

以下就注册场景中使用了消息队列RocketMQ版的效果进行说明。

异步解耦

对于用户来说，注册功能实际只需要注册系统存储用户的账户信息后，该用户便可以登录，后续的注册短信和邮件不是即时需要关注的步骤。

对于注册系统而言，发送注册成功的短信和邮件通知并不一定要绑定在一起同步完成，所以实际当数据写入注册系统后，注册系统就可以把其他的操作放入对应的消息队列RocketMQ版中然后马上返回用户结果，由消息队列RocketMQ版异步地进行这些操作。

数据流动如下所述：

用户在注册页面填写账号和密码并提交注册信息，这些注册信息首先会被写入数据库。
注册信息写入数据库成功后，再发送消息至消息队列。消息队列会马上返回响应给注册系统，注册完成。用户可立即登录。
下游的邮件和短信注册订阅消息队列的此类注册请求消息，即可向用户发送邮件和短信通知，完成所有的注册流程。

用户只需在注册页面等待注册数据写入注册系统和消息队列RocketMQ版的时间，即等待55 ms即可登录。

异步解耦是消息队列RocketMQ版的主要特点，主要目的是减少请求响应时间和解耦。主要的适用场景就是将比较耗时而且不需要即时（同步）返回结果的操作作为消息放入消息队列。同时，由于使用了消息队列RocketMQ版，只要保证消息格式不变，消息的发送方和接收方并不需要彼此联系，也不需要受对方的影响，即解耦。

削峰填谷

流量削峰也是消息队列RocketMQ版的常用场景，一般在秒杀或团队抢购活动中使用广泛。

在秒杀或团队抢购活动中，由于用户请求量较大，导致流量暴增，秒杀的应用在处理如此大量的访问流量后，下游的通知系统无法承载海量的调用量，甚至会导致系统崩溃等问题而发生漏通知的情况。为解决这些问题，可在应用和下游通知系统之间加入消息队列RocketMQ版。

秒杀处理流程如下所述：

用户发起海量秒杀请求到秒杀业务处理系统。
秒杀处理系统按照秒杀处理逻辑将满足秒杀条件的请求发送至消息队列。
假如消息队列长度超过最大数量，则直接抛弃用户请求或跳转到错误页面。
用户收到秒杀成功的通知。

Linux 安装和部署RocketMQ

Linux 配置环境：

JDK: jdk-8u271-linux-x64.tar.gz

RocketMQ: rocketmq-all-4.7.1-bin-release

RocketMQ控制台: rocketmq-externals

安装RocketMQ

传入 Linux 服务器
解压 rocketmq-all-4.7.1-bin-release
```
unzip rocketmq-all-4.7.1-bin-release.zip
```
注意：解压文件的时候说没有zip的命令
安装命令： yum install zip #提示输入时，请输入y；
安装命令：yum install unzip #提示输入时，请输入y；

配置 jvm 参数

cd bin
vim runserver.sh

cd bin
vim runbroker.sh

启动 NameServer
在bin目录下启动
启动之前需要编辑配置文件，修改 JVM 内存设置，默认给的内存 4 GB，超过我们的 JVM 了。
上面已经修改过了
```
#打印日志启动 ./mqnamesrv
#后台启动
nohup ./mqnamesrv &
```
检查是否启动成功
```
netstat -an | grep 9876
```
netstat 命令无效

启动 Broker

在bin目录下启动

nohup ./mqbroker -n [虚拟机端口/服务端口]:9876 &

查看结果
```
netstat -nlet
```

测试 RocketMQ

消息发送

在 bin 目录下进行操作

export NAMESRV_ADDR=localhost:9876
./tools.sh org.apache.rocketmq.example.quickstart.Producer

消息接收

在 bin 目录下进行操作

export NAMESRV_ADDR=localhost:9876
./tools.sh org.apache.rocketmq.example.quickstart.Consumer

关闭 RocketMQ
在 bin 目录下进行操作
```
./mqshutdown broker
./mqshutdown namesrv
```

安装 RocketMQ 控制台

解压缩包
进入rocketmq-externals-rocketmq-console-1.0.0\rocketmq-externals-rocketmq-console-1.0.0\rocketmq-console\src\main\resources 目录修改 application.properties 配置

打包项目

#查看是否配置maven
mvn -v

配置环境变量

配置成功

打包项目

mvn clean package -Dmaven.test.skip=true

进入 target 启动 jar

java -jar rocketmq-console-ng-1.0.0.jar

打开浏览器访问 localhost:9898，如果报错

这是因为我们的 RocketMQ 安装在 Linux 中，控制台在 windows，Linux 需要开放端口才能访问，开放 10909 和 9876 端口

#放行
firewall-cmd --zone=public --add-port=10909/tcp --permanent
firewall-cmd --zone=public --add-port=9876/tcp --permanent
firewall-cmd --zone=public --add-port=10911/tcp --permanent
#重启
systemctl restart firewalld.service
firewall-cmd --reload

成功运行

Java 实现消息发送

在 provider 提供者导入依赖

<!-- Spring Boot Rocketmq -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.1.0</version>
</dependency>

生产消息

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class Provider {
    public static void main(String[] args) throws Exception {
        //创建消息生产者
        DefaultMQProducer producer = new DefaultMQProducer("myproducer-group");
        //设置NameServer
        producer.setNamesrvAddr("[服务器端口/虚拟机端口]:9876");
        //启动生产者
        producer.start();
        //构建消息对象  主题   标签  内容
        Message message = new Message("myTopic","myTag",("鹏之徒于南冥也；水击三千里；抟扶摇而上者九万里。").getBytes());
        //发送消息
        SendResult result = producer.send(message, 6000);
        System.out.println(result);
        //关闭生产者
        producer.shutdown();
    }
}

打开 RocketMQ 控制台，可查看消息。

Java 实现消息消费

在 consumer 消费者导入依赖

<!-- Spring Boot Rocketmq -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.1.0</version>
</dependency>

<!-- Lombok -->
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
</dependency>

消费消息


import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

@Slf4j
public class Consumer {

    public static void main(String[] args) throws MQClientException {
        //创建消息消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myconsumer-group");
        //设置NameServer
        consumer.setNamesrvAddr("[服务器端口/虚拟机端口]:9876");
        //指定订阅的主题和标签
        consumer.subscribe("myTopic","*");
        //回调函数  创建了一个消息监听器,一直监听消息队列中的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                log.info("Message=>{}",new String(list.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //启动消费者
        consumer.start();
    }
}

打开 RocketMQ 控制台，可查看消息。

Spring Boot 整合 RocketMQ

创建 dao 模块统一公共的依赖

依赖配置

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

<!-- Lombok -->
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
</dependency>

实体类

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.util.Date;

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Order {
    private Integer id;
    private String buyerName;
    private String buyerTel;
    private String address;
    private Date createDate;
}

提供者继承 dao 模块

provider 提供者依赖配置

<dependency>
    <groupId>com.cp</groupId>
    <artifactId>dao</artifactId>
    <version>0.0.1-SNAPSHOT</version>
</dependency>

<!--nacos-->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>


<!-- Spring Boot Rocketmq -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.1.0</version>
</dependency>

<!-- Rocketmq 可视化连接 -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.7.1</version>
</dependency>

application.yml 配置

spring:
  application:
    name: provider
  cloud:
    nacos:
      discovery:
        server-addr: [服务器端口/虚拟机端口]:8848

server:
  port: 9000

rocketmq:
  name-server: [服务器端口/虚拟机端口]:9876
  producer:
    group: myprovider

Controller

import com.cp.enitiy.Order;
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;

import java.util.Date;

/**
 * @author : CP
 * @version: 1.0
 * @program : spring-cloud-alibaba
 * @description :
 * @date : 2021-03-04 11:39
 **/
@RestController
public class ProviderController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @GetMapping("/create")
    public Order create(){
        Order order = new Order(
                1,
                "张三",
                "123123",
                "软件园",
                new Date()
        );
        this.rocketMQTemplate.convertAndSend("orderTopic",order);
        return order;
    }
}

运行前先启动 nacos 服务

启动访问 localhost:9000/create 进行消息发送

先启动提供者或消费者都无所谓

消费者继承 dao 模块

consumer 消费者依赖配置

<dependency>
    <groupId>com.cp</groupId>
    <artifactId>dao</artifactId>
    <version>0.0.1-SNAPSHOT</version>
</dependency>

<!--nacos-->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>


<!-- Spring Boot Rocketmq -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.1.0</version>
</dependency>

<!-- Rocketmq 可视化连接 -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.7.1</version>
</dependency>

application.yml 配置

spring:
  application:
    name: consumer
  cloud:
    nacos:
      discovery:
        server-addr: [服务器端口/虚拟机端口]:8848

server:
  port: 9001

rocketmq:
  name-server: [服务器端口/虚拟机端口]:9876

ConsumerService


import com.cp.enitiy.Order;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

/**
 * @author : CP
 * @version: 1.0
 * @program : spring-cloud-alibaba
 * @description :
 * @date : 2021-03-16 20:13
 **/
@Slf4j
@Service
@RocketMQMessageListener(consumerGroup = "myConsumer",topic = "orderTopic")
public class ConsumerService implements RocketMQListener<Order> {
    @Override
    public void onMessage(Order order) {
        log.info("新订单{}，发短信",order);
    }
}

启动消费者消费消息队列

消费者会实时监控提供者是否产生消息来通过消费

原文链接：https://blog.csdn.net/m0_46210779/article/details/114899059

RocketMQ-消息队列

核心概念

应用场景

适用场景

异步解耦

削峰填谷

Linux 安装和部署RocketMQ

安装RocketMQ

安装 RocketMQ 控制台

Java 实现消息发送

Java 实现消息消费

Spring Boot 整合 RocketMQ

创建 dao 模块统一公共的依赖

依赖配置

实体类

提供者 继承 dao 模块

provider 提供者 依赖配置

application.yml 配置

Controller

消费者 继承 dao 模块

consumer 消费者 依赖配置

application.yml 配置

ConsumerService

提供者继承 dao 模块

provider 提供者依赖配置

消费者继承 dao 模块

consumer 消费者依赖配置