SpringBoot2 | 第十九篇：整合RabbitMQ（简单入门）

—— 基于点对点（自动 Ack）

SpringBoot 集成 RabbitMQ，如果只是简单的使用，配置非常少，SpringBoot 提供了 spring-boot-starter-amqp 项目对消息各种支持。本文以简单版（一个生产者和一个消费者）的形式入门，使用默认的消息交换机（Direct Exchange）

[TOC]

消息中间件

消息队列中间件指利用高效可靠的消息传递机制进行与平台无关的数据交流，并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型，它可以在分布式环境下扩展进程间的通信。 消息队列中间件 也称为 消息队列 和 消息中间件。它一般有两种传递模式：点对点模式 和 发布/订阅模式。

1562331552570

消息中间件的作用

解耦：消息中间件在处理过程中插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一层接口，这允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束即可。
存储：在某些情况下，处理数据的过程会失败。消息中间件可以把数据进行持久化处理直到它们已经被完全处理，通过这一方式规避了数据丢失的风险。在把一个消息从消息中间件中删除之前，需要你的处理系统明确指出该消息已经被处理完成，从而确保你的数据被安全的保存直到你使用完毕。
削峰：在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见。如果以处理这类峰值为标准投入资源，无疑是巨大的浪费。使用消息中间件能够使用关键组件支撑突发访问压力，不会因为突发的超负荷请求而完全崩溃。
可恢复性：当系统的一部分组件失效时，不会影响到整个系统。消息中间件降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入消息中间件的消息仍然可以在系统恢复后进行处理。
顺序保证：在大多数使用场景下，数据处理的顺序很重要，大部分消息中间件支持一定程度上的顺序性。
缓冲：在任何重要的系统中，都会存在需要不同处理时间的元素。消息中间件通过一个缓冲层来帮助任务最高效率的执行，写入消息中间件的处理会尽可能快速。该缓冲层有助于控制和优化数据流经过系统的速度。
异步通性：在很多时候应用不想也不需要立即处理消息。消息中间件提供了异步处理机制，允许应用把一些消息放入消息中间件中，但不立即处理它，在之后需要的时候再慢慢处理。

RabbitMQ介绍

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP 的主要特征是 面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ 是实现 AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在 易用性、扩展性、高可用性 等方面表现不俗。RabbitMQ 服务器端用 Erlang 语言 编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。支持延迟队列（这是一个非常有用的功能）….

JMS 是指一套规范，它是有 Sun 公司提出来的，实际就是指 JMS API，旨在为 java 应用提供一套统一的 API（类似 JPA 吧）。AMQP 则是一种协议，RabbitMQ 则是基于这种协议产生的消息中间件。RabbitMQ 还能作为 JMS 的实现，因为它支持 JMS 这种客户端（类似 Spring Cache，可以有很多实现，其中之一就是 Spring Data Redis，前面的整合笔记用过呢）。

环境/版本一览：

开发工具：Intellij IDEA 2018.2.2
springboot： 2.0.6.RELEASE
jdk：1.8.0_171
maven：3.3.9
spring-boot-starter-amqp： 2.0.6.RELEASE

1、pom.xml

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!-- RabbitMQ -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
</dependencies>

2、application.yml

spring:
  rabbitmq:
    username: guest    # 默认为`guest`
    password: guest    # 默认为`guest`
    host: localhost    # 默认为`localhost`
    port: 5672  # 默认为 5672
    virtual-host: /
    # 开启自动ACK，默认为`none`（no ack）
    listener:
      simple:
        acknowledge-mode: auto

3、config

package com.fatal.config;

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

/**
 * RabbitMQ 配置类
 * @author: Fatal
 * @date: 2018/10/22 0022 10:41
 */
@Configuration
public class RabbitMQConfig {

    public static final String QUEUE_NAME = "fatal";

    /**
     * 配置一个队列`QUEUE_NAME`
     */
    @Bean
    public Queue queue() {
        /*
          在 RabbitMQ 服务器上创建一个 `QUEUE_NAME` 队列
          由于没有指定交换机，所以会与`RabbitMQ 服务器`的默认交换机(AMQP default)绑定，
          路由键为该队列的名称
         */
        return new Queue(QUEUE_NAME);
    }

}

4、sender

package com.fatal.sender;

import com.fatal.config.RabbitMQConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * 发送者
 * @author: Fatal
 * @date: 2018/10/22 0022 10:44
 */
@Slf4j
@Component
public class FatalSender {

    private RabbitTemplate rabbitTemplate;

    @Autowired
    public FatalSender(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void send() {
        String context = "现在时刻" + new Date();
        log.info("【FatalSender】发布de消息：[{}]", context);
        rabbitTemplate.convertAndSend(RabbitMQConfig.QUEUE_NAME, context);
    }

}

5、receiver

package com.fatal.receiver;

import com.fatal.config.RabbitMQConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * 接收者
 * @author: Fatal
 * @date: 2018/10/22 0022 10:45
 */
@Slf4j
@Component
public class FatalReceiver {

    @RabbitListener(queues = RabbitMQConfig.QUEUE_NAME)
    public void process(String context) {
        log.info("【FatalReceiver】处理了：[{}]", context);
        // 模拟消费者业务异常，观察自动 ack 的弊端。
        // 当消费者出现异常导致消费失败，这会导致消费者循环消费。
        // int i = 1/0;
    }

}

6、controller

package com.fatal.controller;

import com.fatal.sender.FatalSender;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author: Fatal
 * @date: 2019/7/5 0005 16:09
 */
@RestController
@RequestMapping("/send")
public class SenderController {

    private FatalSender sender;

    @Autowired
    public SenderController(FatalSender sender) {
        this.sender = sender;
    }

    @GetMapping
    public void send() {
        // 发布消息
        sender.send();
    }

}

7、测试

7.1、测试抛正常的 `autoAck`

注掉FatalReceiver类 22 行的int i = 1/0;的注释，然后运行项目

访问http://localhost:8080/send

RabbitMQ 客户端：

1562325158518

控制台：

1562325158518

7.2、测试抛异常的 `autoAck`

打开FatalReceiver类 22 行的int i = 1/0;，然后运行项目

访问http://localhost:8080/send

你会发现循环异常，这时候终止程序。。。

RabbitMQ 客户端：

1562325449677

控制台：

1562330795400

这是简单的入门，使用自动 Ack 的方式。当然咯，缺陷如 7.2 的测试结果一样，所以不推荐使用 自动 Ack

笔记

基础概念

通常我们谈到队列服务, 会有三个概念：发消息者、队列、收消息者，RabbitMQ 在这个基本概念之上, 多做了一层抽象, 在发消息者和队列之间, 加入了交换器 (Exchange). 这样发消息者和队列就没有直接联系, 转而变成发消息者把消息给交换器, 交换器根据调度策略再把消息再给队列。

1558714445588

左侧 P 代表生产者，也就是往 RabbitMQ 发消息的程序。
中间即是 RabbitMQ Broker，其中包括了交换机和队列。
右侧 C 代表消费者，也就是往 RabbitMQ 拿消息的程序。

Broker：简单来说就是消息队列服务器实体
Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列
Queue：消息队列载体，每个消息都会被投入到一个或多个队列
Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来，在绑定的时候一定会指定一个绑定键“Binding Key”
Routing Key：路由关键字，exchange根据这个关键字进行消息投递
vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离
producer：消息生产者，就是投递消息的程序
consumer：消息消费者，就是接受消息的程序
channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务

常见应用场景

邮箱发送：用户注册后投递消息到rabbitmq中，由消息的消费方异步的发送邮件，提升系统响应速度
流量削峰：一般在 秒杀活动 中应用广泛，秒杀会因为流量过大，导致应用挂掉，为了解决这个问题，一般在应用前端加入消息队列。用于控制活动人数，将超过此一定阀值的订单直接丢弃。缓解短时间的高流量压垮应用。
订单超时：利用rabbitmq的 延迟队列，可以很简单的实现 订单超时 的功能，比如用户在下单后30分钟未支付取消订单
还有更多应用场景就不一一列举了…..

怎么知道没有配置交换机的时候是使用`Direct`作为默认交换机呢？

1540200585317

源码

点开 RabbitTemplate 的源码来看一下

它这里定义了一个常量 DEFAULT_EXCHANGE

1540028823076

然后，当我们使用 RabbitTemplate 发布消息的时候，如果我们没有指定交换机，那么调用的方法类似下面的 convertAndSend(String routingKey, final Object object) ，可以看出，它帮我们指定了默认的交换机了。

1540028970290

RabbitMQ 客户端

RabbitMQ 已经帮我们初始化了一些交换机，其中就包括默认交换机

1562373829677

点进去查看它的 Bindings，你可以看到一串描述

1562373897956

The default exchange is implicitly bound to every queue, with a routing key equal to the queue name. It is not possible to explicitly bind to, or unbind from the default exchange. It also cannot be deleted.
翻译为：
默认交换机隐式的绑定到每个队列，路由键等于队列的名称。不能显示绑定到默认交换机，或者从默认交换机中解绑，也不能删除。
发送到默认交换机的方式
rabbitTemplate.convertAndSend(队列名称(路由键), 数据);
第三篇会做个 Direct 交换机，然后用上面的方式，发送消息到默认交换机。也就是说，参数指定了交换机，那么就发送到该交换机，没有指定交换机的话，那么拿队列名作为路由键，并发送到默认交换机

参考资料

springboot(八)：RabbitMQ详解

一起来学SpringBoot | 第十二篇：初探RabbitMQ消息队列

RabbitMQ 实战指南

总结

SpringBoot的知识已经有前辈在我们之前探索了。比较喜欢的博主有：唐亚峰 | Battcn、方志朋的专栏、程序猿DD、纯洁的微笑。对这门技术感兴趣的可以去他们的博客逛逛。谢谢他们的分享~~

以上文章是我用来学习的Demo，都是基于 SpringBoot2.x 版本。

源码地址: https://github.com/ynfatal/springboot2-learning/tree/master/chapter19_1