【Spring Cloud】RabbitMQ 消息队列服务入门_spring cloud rabiitmq

RabbitMQ 消息队列服务

RabbitMQ 使用场景

服务解耦

假设有这样一个场景, 服务A产生数据, 而服务B,C,D需要这些数据, 那么我们可以在A服务中直接调用B、C、D服务，把数据传递到下游服务即可。

但是，随着我们的应用规模不断扩大，会有更多的服务需要 A 的数据，如果有几十甚至几百个下游服务，而且会不断变更，再加上还要考虑下游服务出错的情况，那么A服务中调用代码的维护会极为困难。

这是由于服务之间耦合度过于紧密。

再来考虑用RabbitMQ解耦的情况，A服务只需要向消息服务器发送消息，而不用考虑谁需要这些数据。下游服务如果需要数据，自行从消息服务器订阅消息，不再需要数据时则取消订阅即可。

RabbitMQ是一个消息中间件，你可以想象它是一个邮局。当你把信件放到邮箱里时，能够确信邮递员会正确地递送你的信件。RabbitMq就是一个邮箱、一个邮局和一个邮递员。

RabbitMQ基本概念

RabbitMQ是一种消息中间件，用于处理来自客户端的异步消息。服务端将要发送的消息放入到队列池中。接收端可以根据RabbitMQ配置的转发机制接收服务端发来的消息。

RabbitMQ依据指定的转发规则进行消息的转发、缓冲和持久化操作，主要用在多服务器间或单服务器的子系统间进行通信，是分布式系统标准的配置。

Exchange

接受生产者发送的消息，并根据Binding规则将消息路由给服务器中的队列。

ExchangeType决定了Exchange路由消息的行为。

在RabbitMQ中，ExchangeType常用的有==Direct（直接）、Fanout（扇出）和Topic（主题）==三种。

Message Queue

消息队列。我们发送给RabbitMQ的消息最后都会到达各种queue，并且存储在其中(如果路由找不到相应的queue则数据会丢失)，等待消费者来取。

Binding Key

它表示的是Exchange与Message Queue是通过binding key进行联系的，这个关系是固定。

Routing Key

生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则。这个routing key需要与Exchange Type及binding key联合使用才能生，我们的生产者只需要通过指定routing key来决定消息流向哪里。

RabbitMQ 的安装与启动

Linux中搭建搭建 Docker 环境

更改虚拟网络

打开 VMware 虚拟机，配置虚拟网络

更改子网的IP

安装虚拟机

打开VMware虚拟机，将 centos-8 添加到VMware虚拟机中，点击下面文件

添加成功

打开 centos-8虚拟机，检查网络

关闭虚拟机，并克隆 centos-8 或 centos-7： docker-base，改名为docker-base

使用MobaXterm_Personal工具链接虚拟机系统，运行终端

安装 Docker

参考【Docker技术实践】01-Docker入门

终端输入docker指令查看是否安装成功

搭建Rabbitmq服务器

将 docker-base （安装了docker环境）克隆一份起名为 rabbitmq ,并运行虚拟机。

设置ip

进入rabbitmq 虚拟机后依次执行下面指令，更改并检查ip信息

./ip-static
1

ip: 192.168.64.140
1

ifconfig
1

使用MobaXterm_Personal工具链接虚拟机系统，运行终端

下载 RabbitMQ 镜像

方式一：Linux上网络下载

rabbitmq:management 镜像中已经安装了管理界面。

   docker pull rabbitmq:management
1

方式二： 本地拉取

从 code 下载 rabbit-image.gz，上传到服务器，然后执行镜像导入 。

• 上传离线安装包
• 安装

docker load -i rabbit-image.gz
1

Docker 启动 RabbitMQ 容器

安装完成后依次执行下面命令

• 停止并关闭防火墙

systemctl stop firewalld
1

systemctl disable firewalld
1

• 重启 docker 系统服务

systemctl restart docker
1

• 创建rabbitmq文件夹，在rabbitmq文件夹下创建 rabbitmq.conf 文件，编写配置。

创建文件夹

mkdir /etc/rabbitmq
1

rabbitmq 文件夹下创建 rabbitmq.conf 配置文件

vim /etc/rabbitmq/rabbitmq.conf

# 进入 rabbitmq.conf 添加两行配置：
default_user = admin
default_pass = admin
1
2
3
4
5

启动 rabbitmq 容器

docker run -d --name rabbit \
-p 5672:5672 \
-p 15672:15672 \
-v /etc/rabbitmq/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
-e RABBITMQ_CONFIG_FILE=/etc/rabbitmq/rabbitmq.conf \
rabbitmq:management
1
2
3
4
5
6

访问 RabbitMQ 管理控制台

启动 rabbitmq 容器后，访问管理控制台

http://192.168.64.140:15672
1

用户名密码是 admin

RabbitMQ 消息队列服务界面

用户管理参考
添加链接描述

其他

设置服务,开机自动启动

systemctl enable rabbitmq-server
1

启动服务

systemctl start rabbitmq-server
1

打开客户端连接端口

firewall-cmd --zone=public --add-port=5672/tcp --permanent
firewall-cmd --reload
1
2

主要端口介绍

• 4369 – erlang发现口
• 5672 – client端通信口
• 15672 – 管理界面ui端口
• 25672 – server间内部通信口

RabbitMQ 六种工作模式

简单模式

概述

• 发送消息的程序是生产者

• 队列就代表一个邮箱。虽然消息会流经RbbitMQ和你的应用程序，但消息只能被存储在队列里。队列存储空间只受服务器内存和磁盘限制，它本质上是一个大的消息缓冲区。多个生产者可以向同一个队列发送消息，多个消费者也可以从同一个队列接收消息。

• 消费者等待从队列接收消息

实战案列

创建工程

打开Idea，新建empty project 工程，命名为rabbitmq

pom.xml

在 rabbitmq 工程中添加 Maven 模块，命名为rabitmq-api，添加com.rabbitmq-amqp-client 依赖。

<?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>cn.tedu</groupId>
    <artifactId>rabbitmq-api</artifactId>
    <version>1.0-SNAPSHOT</version>
    
    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.4.3</version>
        </dependency>
    </dependencies>

</project>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

并在java包下创建m1包，创建Producer类实现消息的发送，Consumer类实现消息的接收。

业务代码

Producer类 生产者发送消息

package m1;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author deven
 * @version 1.0
 * @description: RabbitMQ 的生产者连接测试类
 * @date 2021/10/21 16:50
 */
public class Producer {

    public static void main(String[] args) throws IOException, TimeoutException {
        //连接RabbitMQ
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.64.140"); // wht6.cn
        f.setPort(5672);
        f.setUsername("admin");
        f.setPassword("admin");
        Connection con = f.newConnection();
        Channel c = con.createChannel();//通信通道

        /**
         * @description: 创建队列 hello-world
         * 如果队列已经存在，不会重复创建
         * queueDeclare (String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5)
         * var1 队列名
         * var2 是否是持久队列
         * var3 是否是排他队列、独占队列
         * var4 是否自动删除（没有消费者时自动删除）
         */
        c.queueDeclare("hello-world", false, false, false, null);

        /**
         * @description: 向 helloworld 队列发送消息
         * basicPublish (String var1, String var2, BasicProperties var3, byte[] var4)
         * var1 交换机，""是一个默认的交换机
         * var3 消息的其他参数属性
         */
        c.basicPublish("", "hello-world", null, "Hello world!".getBytes());
        System.out.println("消息已发送");
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

Consumer类 消费者接收消息

package m1;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author deven
 * @version 1.0
 * @description: RabbitMQ 的消费者连接测试类
 * @date 2021/10/21 17:12
 */
public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 连接RabbitMQ
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.64.140"); // wht6.cn
        f.setPort(5672);
        f.setUsername("admin");
        f.setPassword("admin");
        Connection con = f.newConnection();
        Channel c = con.createChannel();//通信通道

        // 1.创建队列 hello-world，如果队列已经存在，不会重复创建
        c.queueDeclare("hello-world", false, false, false, null);

        // 3. 创建回调对象
        DeliverCallback deliverCallback =
                (consumerTag, message) -> {
                    byte[] a = message.getBody();
                    String s = new String(a);
                    System.out.println("收到： " + s);
                };

        CancelCallback cancelCallback = (consumerTag) ->{ };

        /**
         * @description:  2.接收消息，收到的消息会传递到一个回调对象去处理
         * @param: c.basicConsume("hello-world", true, 处理消息的回调对象, 取消接收消息的回调对象);
         *  第二个参数： autoAck - auto acknowledgment 自动确认
         *  true    - 自动确认
         *      消息一向消费者发送，服务器立即自动确认消息，删除消息
         *  false   - 手动确认
         *      消息发出后，服务器会缓存消息，不删除，等待消费者发回一个确认消息（回执）才删除，
         *      如果消费者处理消息过程中崩溃或离线，服务器会回滚消息，等待重新发送
         */
        c.basicConsume("hello-world", true, deliverCallback, cancelCallback);

    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

测试 RabbitMQ

启动 Producer 类 发送消息。

查看 RabbitMQ 显示一条消息在队列中。

启动 Consumer 类 接收消息。

查看 RabbitMQ 发现队列中的消息已被取出，队列清空。

工作模式

工作队列(即任务队列)背后的主要思想是避免立即执行资源密集型任务，并且必须等待它完成。相反，我们将任务安排在稍后完成。

我们将任务封装为消息并将其发送到队列。后台运行的工作进程将获取任务并最终执行任务。当运行多个消费者时，任务将在它们之间分发。

使用任务队列的一个优点是能够轻松地并行工作。如果我们正在积压工作任务，我们可以添加更多工作进程，这样就可以轻松扩展。

业务代码

Producer类 生产者发送消息

Consumer类 消费者接收消息

运行:

• 一个生产者
• 两个消费者

生产者发送多条消息，如: 1,2,3,4,5. 两个消费者分别收到:

• 消费者一: 1,3,5
• 消费者二: 2,4

rabbitmq在所有消费者中轮询分发消息，把消息均匀地发送给所有消费者

消息确认

一个消费者接收消息后，在消息没有完全处理完时就挂掉了,那么这时会发生什么呢?

就现在的代码来说,rabbitmq把消息发送给消费者后,会立即删除消息,那么消费者挂掉后,它没来得及处理的消息就会丢失

如果生产者发送以下消息:

1…

2

3

4

5

两个消费者分别收到:

消费者一: 1…, 3, 5
消费者二: 2, 4
当消费者一收到所有消息后,要话费7秒时间来处理第一条消息,这期间如果关闭该消费者,那么1未处理完成,3,5则没有被处理

订单流量削峰

导入pd-web工程

执行下面语句，修改sqlyog配置。

SET GLOBAL max_allowed_packet=100000000;
SET GLOBAL net_buffer_length=100000;
SET GLOBAL  interactive_timeout=28800000;
SET GLOBAL  wait_timeout=28800000
1
2
3
4

将 pd.sql 导入数据库