互联网分布式应用之RabbitMQ

RabbitMQ

Java 是第一大编程语言和开发平台。它有助于企业降低成本、缩短开发周期、推动创新以及改善应用服务。如今全球有数百万开发人员运行着超过 51 亿个 Java 虚拟机，Java 仍是企业和开发人员的首选开发平台。

课程内容的介绍

1. RabbitMQ介绍安装

2. RabbitMQ交换器

3. RabbitMQ高级

一、RabbitMQ介绍安装

1. AMQP 简介

AMQP (Advanced Message Queuing Protocol ，高级消息队列协议）是个线路层的协议规范，而不是 API 规范（例如 JMS ）。由于AMQP 是一个线路层协议规范，因此它天然就是跨平台的，就像 SMTP HTTP 等协议样，只要开发者按照规范的格式发送数据，任何平台都可以通过 AMQP进行消息交互。像目前流行的 StormMQ RabbitMQ 等都实现了 AMQP。

2. RabbitMQ简介

RabbitMQ 一个实现了 AMQP 的开源消息中间件，使用高性能的 Erlang 编写。 RabbitMQ有可靠性、支持多种协议、高可用、支持消息集群以及多语言客户端等特点，在分布式系统中存储转发消息，具有不错的性能表现。

为什么要使用 RabbitMQ？他解决了什么问题？

现在的市面上有很多MQ可以选择，比如ActiveMQ、ZeroMQ、Appche Qpid，那问题来了为什么要选择RabbitMQ？

除了Qpid，RabbitMQ是唯一一个实现了AMQP标准的消息服务器；

可靠性，RabbitMQ的持久化支持，保证了消息的稳定性；

高并发，RabbitMQ使用了Erlang开发语言，Erlang是为电话交换机开发的语言，天生自带高并发光环，和高可用特性；

集群部署简单，正是应为Erlang使得RabbitMQ集群部署变的超级简单；

社区活跃度高，根据网上资料来看，RabbitMQ也是首选；

3. 消息队列基础知识

3.1 Provider

消息生产者，就是投递消息的程序。

3.2 Consumer

消息消费者，就是接受消息的程序。

3.3 非消息队列

3.4 使用消息队列

3.5 什么是队列？

队列就像存放了商品的仓库或者商店，是生产商品的工厂和购买商品的用户之间的中转站。

3.6 队列里存储了什么？

在 rabbitMQ 中，信息流从你的应用程序出发，来到 Rabbitmq 的队列，所有信息可以只存储在一个队列中。队列可以存储很多信息，因为它基本上是一个无限制的缓冲区，前提是你的机器有足够的存储空间。

3.7 队列和应用程序的关系？

多个生产者可以将消息发送到同一个队列中，多个消息者也可以只从同一个队列接收数据。

4. RabbitMQ安装

因为在Linux上面直接安装RabbitMQ比较复杂，而且容易出错，所以我们通过Docker来快速的安装我们的RabbitMQ。

4.1 查找镜像

docker search rabbitmq:management

4.2 拉取镜像

docker pull macintoshplus/rabbitmq-management

4.3 查看镜像

docker images

4.4 创建容器

docker run -d --hostname bobo01 --name rabbitmq -e RABBITMQ_DEFAULT_USER=guest -e RABBITMQ_DEFAULT_PASS=guest -p 15672:15672 -p 5672:5672 c20

4.5 查看容器

docker ps -a

4.6 访问测试

http://192.168.100.120:15672/

这就表示RabbitMQ安装成功并且访问成功了！

5.入门案例

5.1 创建项目

创建一个SpringBoot项目，并引入相关的依赖即可。

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

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
    </dependencies>

5.2 配置文件

我们需要在application.properties中添加RabbitMQ的相关的配置信息。

spring.application.name=rabbitmq-demo01
spring.rabbitmq.host=192.168.100.120
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

# 自定义一个属性 设置 队列的名称
mq.queue.name=hello-queue

5.3 队列配置文件

package com.bobo.config;

import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class QueueConfig {

    @Value("${mq.queue.name}")
    private String queueName;

    @Bean
    public Queue createQueue(){
        return new Queue(queueName);
    }
}

5.4 消费者

package com.bobo.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * 消费者
 */
@Component
public class Receiver {

    /**
     * 接收消息，然后处理消息
     */
    @RabbitListener(queues = {"${mq.queue.name}"})
    public void process(String msg){
        // 处理消息
        System.out.println("recevier: " + msg);
    }
}

5.5 生产者

package com.bobo.provider;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * 消息的提供者
 *
 */
@Component
public class Sender {

    @Autowired
    private AmqpTemplate template;

    @Value("${mq.queue.name}")
    private String queueName;

    /**
     * 发送消息的方法
     */
    public void send(String msg){
        // 队列名称  消息内容
        template.convertAndSend(queueName,msg);
    }

}

5.6 测试

首先启动服务，消费者处于监听状态。

启动后可以在RabbitMQ的服务中看到对应的队列和消费者信息。

通过单元测试来发送消息。

package com.bobo;

import com.bobo.provider.Sender;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
class RabbitmqDemo01ApplicationTests {

    @Autowired
    private Sender sender;

    @Test
    void contextLoads() {
        sender.send("你好啊....");
    }

}

6. RabbitMQ原理介绍

6.1 原理图

6.2 概念介绍

1.Message

消息。消息是不具名的，它由消息头消息体组成。消息体是不透明的，而消息头则由一系列可选属性组成，这些属性包括：routing-key(路由键)、priority(相对于其他消息的优先权)、delivery-mode(指出消息可能持久性存储)等。

2.Publisher

消息的生产者。也是一个向交换器发布消息的客户端应用程序。

3.Consumer

消息的消费者。表示一个从消息队列中取得消息的客户端应用程序。

4.Exchange

交换器。用来接收生产者发送的消息并将这些消息路由给服务器中的队列。

三种常用的交换器类型

1. direct(发布与订阅完全匹配)

2. fanout(广播)

3. topic(主题，规则匹配)

5.Binding

绑定。用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。

6.Queue

消息队列。用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者链接到这个队列将其取走。

7.Routing-key

路由键。RabbitMQ 决定消息该投递到哪个队列的规则。队列通过路由键绑定到交换器。消息发送到 MQ 服务器时，消息将拥有一个路由键，即便是空的 ,RabbitMQ 也会将其和绑定使用的路由键进行匹配。如果相匹配，消息将会投递到该队列。如果不匹配，消息将会进入黑洞。

8.Connection

链接。指 rabbit 服务器和服务建立的 TCP 链接。

9.Channel

1.Channel 中文叫做信道，是 TCP 里面的虚拟链接。例如：电缆相当于 TCP，信道是一个独立光纤束，一条 TCP 连接上创建多条信道是没有问题的。

2.TCP 一旦打开，就会创建 AMQP 信道。

3.无论是发布消息、接收消息、订阅队列，这些动作都是通过信道完成的。

10.Virtual Host

虚拟主机。表示一批交换器，消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是AMQP 概念的基础，必须在链接时指定，RabbitMQ 默认的 vhost 是/

11.Borker

表示消息队列服务器实体。

交换器和队列的关系

交换器是通过路由键和队列绑定在一起的，如果消息拥有的路由键跟队列和交换器的路由键匹配，那么消息就会被路由到该绑定的队列中。也就是说，消息到队列的过程中，消息首先会经过交换器，接下来交换器在通过路由键匹配分发消息到具体的队列中。路由键可以理解为匹配的规则。

RabbitMQ 为什么需要信道？为什么不是 TCP 直接通信？

1. TCP 的创建和销毁开销特别大。创建需要 3 次握手，销毁需要 4 次分手。

2. 如果不用信道，那应用程序就会以 TCP 链接 Rabbit，高峰时每秒成千上万条链接会造成资源巨大的浪费，而且操作系统每秒处理 TCP 链接数也是有限制的，必定造成性能瓶颈。

3. 信道的原理是一条线程一条通道，多条线程多条通道同用一条 TCP 链接。一条 TCP链接可以容纳无限的信道，即使每秒成千上万的请求也不会成为性能的瓶颈。

二、RabbitMQ交换器

1.Direct案例

DirectExchange 路由策略是将消息队列绑定到 DirectExchange 上，当一条消息到达DirectExchange 时会被转发到与该条消息 routing key 相同的 Queue 上，例如消息队列名为“hello-queue ”，则 routingkey 为“hello-queue ”的消息会被该消息队列接收。

1.1 创建消费者

创建项目，并添加依赖。

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

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
    </dependencies>

配置文件

spring.application.name=rabbitmq-demo02
spring.rabbitmq.password=guest
spring.rabbitmq.username=guest
spring.rabbitmq.port=5672
spring.rabbitmq.host=192.168.100.120

# 设置交换器名称
mq.config.exchange=log.direct

# info 队列名称
mq.config.queue.info=log.info
# info 路由键
mq.config.queue.info.routing.key=log.info.routing.key

# error 队列名称
mq.config.queue.error=log.error
# error 路由键
mq.config.queue.error.routing.key=log.error.routing.key

消费者

package com.bobo.consumer;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

/**
 * info 日志的消费者
 *    @QueueBinding value 绑定的队列名称
 *
 *    autoDelete：是否是一个可删除的临时队列
 *    @Exchange：交换器名称和类型
 *    key：路由键
 */
@Component
@RabbitListener(
        bindings = @QueueBinding(
                value = @Queue(value = "${mq.config.queue.error}",autoDelete = "false")
                ,exchange = @Exchange(value = "${mq.config.exchange}"
                                    ,type = ExchangeTypes.DIRECT)
                ,key = "${mq.config.queue.error.routing.key}"
        )
)
public class ErrorRecevier {

    @RabbitHandler
    public void process(String msg) {
        System.out.println("error....recevier:" + msg);
        boolean flag = true;
        if(flag){
            System.out.println(1/0);
        }

    }
}

package com.bobo.consumer;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

/**
 * info 日志的消费者
 *    @QueueBinding value 绑定的队列名称
 *
 *    autoDelete：是否是一个可删除的临时队列
 *    @Exchange：交换器名称和类型
 *    key：路由键
 */
@Component
@RabbitListener(
        bindings = @QueueBinding(
                value = @Queue(value = "${mq.config.queue.info}",autoDelete = "false")
                ,exchange = @Exchange(value = "${mq.config.exchange}"
                                    ,type = ExchangeTypes.DIRECT)
                ,key = "${mq.config.queue.info.routing.key}"
        )
)
public class InfoRecevier {

    @RabbitHandler
    public void process(String msg){
        System.out.println("info....recevier:" + msg);
    }
}

1.2 创建生产者

创建一个SpringBoot项目，添加和上面一样的依赖。

配置文件有区别，不需要添加队列的配置信息。

spring.application.name=rabbitmq-demo03
spring.rabbitmq.password=guest
spring.rabbitmq.username=guest
spring.rabbitmq.port=5672
spring.rabbitmq.host=192.168.100.120

# 设置交换器名称
mq.config.exchange=log.direct

# info 路由键
mq.config.queue.info.routing.key=log.info.routing.key

# error 路由键
mq.config.queue.error.routing.key=log.error.routing.key

添加生产者的类

package com.bobo.provider;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
public class Sender {

    @Autowired
    private AmqpTemplate template;

    @Value("${mq.config.exchange}")
    private String exchange;


    @Value("${mq.config.queue.error.routing.key}")
    private String routingKey;

    public void send(String msg){
        // 发送消息
        template.convertAndSend(exchange,routingKey,msg);
    }
}

package com.bobo;

import com.bobo.provider.Sender;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
class RabbitmqDemo03ApplicationTests {

    @Autowired
    private Sender sender;

    @Test
    void contextLoads() throws Exception{
        //

            sender.send("Hello RabbitMQ .... " + 666);

    }

}

测试效果

2.Topic案例

TopicExchange 是比较复杂也比较灵活的种路由策略,在TopicExchange 中,Queue 通过routingkey 绑定到 TopicExchange 上,当消息到达 TopicExchange 后,TopicExchange 根据消息的routingkey 消息路由到一个或者多 Queue上，相比direct模式topic会更加的灵活些。

package com.bobo.consumer;


import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(
        bindings = @QueueBinding(
                    value = @Queue(value = "${mq.config.queue.error}",autoDelete = "true")
                    ,exchange = @Exchange(
                            value = "${mq.config.exchange}"
                            ,type = ExchangeTypes.TOPIC)
                ,key = "*.log.error"
        )

)
public class ErrorRecevier {

    @RabbitHandler
    public void process(String msg){
        System.out.println("error ... recevier:" + msg);
    }
}

package com.bobo.provider;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
public class OrderSender {

    @Autowired
    private AmqpTemplate template;

    @Value("${mq.config.exchange}")
    private String exchange;

    /**
     * 发送消息
     * @param msg
     */
    public void send(String msg){
        template.convertAndSend(exchange,"Order.log.debug","Order log debug"+msg);
        template.convertAndSend(exchange,"Order.log.info","Order log info"+msg);
        template.convertAndSend(exchange,"Order.log.error","Order log error"+msg);
        template.convertAndSend(exchange,"Order.log.warn","Order log warn"+msg);
    }
}

3.Fanout案例

FanoutExchange 的数据交换策略是把所有到达 FanoutExchang 的消息转发给所有与它绑定的Queue ，在这种策略中， routingkey 将不起任何作用。

package com.bobo.consumer;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(
        bindings = @QueueBinding(
                value = @Queue(value = "${mq.config.queue.sms}",autoDelete = "true")
                ,exchange = @Exchange(value = "${mq.config.exchange}",type = ExchangeTypes.FANOUT)
        )
)
public class SmsRecevier {

    @RabbitHandler
    public void process(String msg){
        System.out.println("Sms .... recevider:" + msg);
    }
}

package com.bobo.provider;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
public class Sender {


    @Autowired
    private AmqpTemplate template;

    @Value("${mq.config.exchange}")
    private String exchange;

    /**
     * 发送消息
     * @param msg
     */
    public void send(String msg){
        template.convertAndSend(exchange,"",msg);
    }
}

三、RabbitMQ高级

1. 持久化

消息的可靠性是RabbitMQ的一大特色，RabbitMQ是如何保证消息的可靠性的呢？--> 消息的持久化。

创建消费者

注意，此时我们需要设置autoDelete=false。

创建服务提供者

package com.bobo.provider;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
public class Sender {
    @Autowired
    private AmqpTemplate template;
    @Value("${mq.config.exchange}")
    private String exchange;
    @Value("${mq.config.queue.error.routing.key}")
    private String routingKey;
    public void send(String msg){
        // 发送消息
        template.convertAndSend(exchange,routingKey,msg);
    }
}

单元测试

package com.bobo;

import com.bobo.provider.Sender;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
class RabbitmqDemo03ApplicationTests {
    @Autowired
    private Sender sender;
    @Test
    void contextLoads() throws Exception{
        //
        for (int i = 0; i < 10000; i++) {
            Thread.sleep(2000);
            sender.send("Hello RabbitMQ .... " + i);
        }
    }
}

当消费者处理了一段时间的消息之后，断开连接，然后消费者再上线我们发现消费者又能够处理掉下线后提供者发送的消息，保证了消息的完整性。

autoDelete属性

@Queue：当所有的消费者客户端连接断开后，是否自定删除队列

true：删除，false：不删除

@Exchange:当所有的绑定队列都不再使用时，是否自动删除交换器

true：删除，false：不删除

2. ACK确认机制

2.1 什么是ACK

如果消息在处理过程中，消费者的服务器在处理消息时出现了异常，那么可能这条正在处理的消息就没有完成消息消费，数据就会丢失，为了确保数据不会丢失，RabbitMQ支持消息确认机制-ACK。

2.2 ACK消息确认机制

ACK(Acknowledge Character)是消费者从RabbitMQ收到消息并处理完成后，反馈给RabbitMQ的，RabbitMQ接收到反馈信息后才会将消息从队列中删除。

1. 如果一个消费者在处理消息出现了网络不稳定，福区群异常等现象，会将消息重新放入队列中。

2. 如果在集群的情况下，RabbitMQ会立即将这个消息推送给这个在线的其他的消费者，这种机制保障了消费者在服务端故障的时候不会丢失任何的数据和任务。

3. 消息永远不会从RabbitMQ中删除：只有当消费者正确发送ACK反馈后，RabbitMQ收到确认后，消息才会从RabbitMQ的服务中删除。

4. 消息的ACK机制默认就是打开的。

ACK的验证

在服务端我们给出一个错误。

然后我们再去掉错误，发现消息会被正常的消费。

ACK的注意事项

如果忘记掉ACK，那么后果会比较严重，当Consumer退出时，Message会一直重复分发，然后RabbitMQ会占用越来越多的内存，由于RabbitMQ会长时间的运行，因此这个内存泄漏是致命的,我们可以通过设置重试次数来防止这个问题，在Consumer的application.properties中设置如下参数。

spring.rabbitmq.listener.simple.retry.enabled=true
## 设置重试次数
spring.rabbitmq.listener.simple.retry.max-attempts=5
## 重试的间隔时间
spring.rabbitmq.listener.simple.retry.initial-interval=5000