一、什么是RabbitMQ

消息队列（Message Queue，简称MQ）：是在消息的传输过程中保存消息的容器。用分布式系统之间进行通信。

1.2.选型和对比

	RabbitMQ	ActiveMQ	RocketMQ	Kafka
公司/社区	Rabbit	Apache	阿里	Apache
开发语言	Erlang	Java	Java	Scala&Java
协议	AMQP	OpenWire、AUTO、Stomp、MQTT	自定义	自定义
单机吞吐量	万级	万级(最差)	十万级	十万级
消息延迟	微妙级	毫秒级	毫秒级	毫秒以内
特性	并发能力很强，延时很低	老牌产品，文档较多	MQ功能比较完备，扩展性佳	只支持主要的MQ功能，毕竟是为大数据领域准备的。

综合上面的材料得出以下两点：

1.中小型软件公司，建议选择RabbitMQ 一方面，erlang语言天生具备高并发的特性，RabbitMQ的社区十分活跃，可以解决开发过程中遇到的bug，这点对于中小型软件公司不如互联网公司，数据量没那么大，选消息中间件，应首选功能比较完备的，所以kafka和rocketmq排除。

2.大型软件公司，根据具体使用在rocketMq和kafka之间二选一。一方面，大型软件公司，具备足够的资金搭建分布式环境，也具备足够大的数据量。针对rocketMQ，大型软件公司也可以抽出人手对rocketMQ进行定制化开发，毕竟国内有能力改JAVA源码的人还是相当多的。至于kafka，根据业务场景选择，如果有日志采集功能，肯定是首选kafka了。

1.3、什么是RabbitMQ？

AMQP,即Advanced Message Queuing Protocol（高级消息队列协议），是一个网络协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同产品，不同开发语言等条件的限制。2006年，AMQP规范发布。类比HTTP。

2007年，Rabbit技术公司基于AMQP标准开发的RabbitMQ1.0发布。RabbitMQ采用Erlang语言开发。

1.4为什么使用RabbitMQ？

1.4.1 解耦

传统模式：系统间耦合性太强，如图所示，系统A在代码中直接调用系统B和系统C的代码，如果将来D系统接入，系统A还需要修改代码。过于麻烦！

中间件模式：将消息写入消息队列，需要消息的系统自己从消息队列中订阅，而系统A不需要做任何修改。

1.4.2 异步

传统模式：一些非必要的业务逻辑以同步的方式运行，太耗费时间。

中间件模式：将消息写入消息队列，非必要的业务逻辑以异步的方式运行，加开响应速度。

1.4.3 削峰

传统模式：并发量大的时候，所有的请求直接怼到数据库，造成数据库连接异常

中间件模式：系统A慢慢的按照数据库能处理的并发量，从消息队列中慢慢拉去消息。

二、安装RabbitMQ

2.1、下载、上传及安装

2.1.1、先克隆一台虚拟机

2.1.2、下载RabbitMQ及其组件Erlang

官网下载地址：Downloading and Installing RabbitMQ — RabbitMQhttp://www.rabbitmq.com/download.html

2.1.3、上传到虚拟机

2.1.4、依次执行命令：

1）rpm -ivh esl-erlang-17.3-1.x86_64.rpm --force --nodeps

2）rpm -ivh esl-erlang_17.3-1~centos~6_amd64.rpm --force --nodeps3）rpm -ivh esl-erlang-compat-R14B-1.el6.noarch.rpm --force --nodeps

4）rpm -ivh rabbitmq-server-3.4.1-1.noarch.rpm

2.2、启动停止

service rabbitmq-server start

service rabbitmq-server stop

service rabbitmq-server restart

service rabbitmq-server status

2.3、设置开机自启

chkconfig rabbitmq-server on

2.4、防火墙开放15672端口

/sbin/iptables -I INPUT -p tcp --dport 15672 -j ACCEPT

/etc/rc.d/init.d/iptables save

2.5.开启web界面管理工具

rabbitmq-plugins enable rabbitmq_management

service rabbitmq-server restart

2.6.创建账户

这里我们以创建个admin帐号，密码1111为例，创建一个账号并支持远程ip访问。

1.创建账号

rabbitmqctl add_user admin 1111

2.设置用户角色

rabbitmqctl set_user_tags admin administrator

3.设置用户权限

rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"

4.设置完成后可以查看当前用户和角色(需要开启服务)

rabbitmqctl list_users

账号guest具有所有的操作权限，并且又是默认账号，出于安全因素的考虑，guest用户只能通过localhost登陆

5.测试

浏览器输入：serverip:15672。其中serverip是RabbitMQ-Server所在主机的ip，15672是RabbitMQ-Server的端口号

2.7、管理界面

2.7.1、主页总览

connections：无论生产者还是消费者，都需要与RabbitMQ建立连接后才可以完成消息的生产和消费，在这里可以查看连接情况

channels：通道，建立连接后，会形成通道，消息的投递获取依赖通道。

Exchanges：交换机，用来实现消息的路由

Queues：队列，即消息队列，消息存放在队列中，等待消费，消费后被移除队列。

端口：

5672: rabbitMq的编程语言客户端连接端口

15672：rabbitMq管理界面端口

25672：rabbitMq集群的端口

2.7.2、添加用户

如果不使用guest，我们也可以自己创建一个用户：

1、超级管理员(administrator)

可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。

2、监控者(monitoring)

可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)

3、策略制定者(policymaker)

可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。

4、普通管理者(management)

仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。

5、其他

无法登陆管理控制台，通常就是普通的生产者和消费者。

2.7.3、创建Virtual Hosts

虚拟主机：类似于mysql中的database。他们都是以“/”开头

2.7.4、设置权限

1.点击用户名

2.设置权限

3.查看效果

三、创建RabbitMQ

3.1、创建工程

3.2、pom.xml

<?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>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.2.RELEASE</version>
    </parent>
    <groupId>com.exercise</groupId>
    <artifactId>springboot_rabbitmq</artifactId>
    <version>1.0-SNAPSHOT</version>

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

</project>

3.3.util

抽取一个建立RbbitMQ连接的工具类，方便其他程序获取连接：

package com.exercise.util;

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

public class ConnectionUtil {
    /**
     * 建立与RabbitMQ的连接
     * @return
     * @throws Exception
     */
    public static Connection getConnection() throws Exception {
        //定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置服务地址
        factory.setHost("192.168.233.132");
        //端口
        factory.setPort(5672);
        //设置账号信息，用户名、密码、vhost
        factory.setUsername("admin");
        factory.setPassword("1111");
        factory.setVirtualHost("/");
        // 通过工程获取连接
        Connection connection = factory.newConnection();
        return connection;
    }
}

四、五种消息模型

RabbitMQ提供了6种消息模型，但是第6种其实是RPC，并不是MQ，因此不予学习。那么也就剩下5种。

但是其实3、4、5这三种都属于订阅模型，只不过进行路由的方式不同。

4.1、simple-简单模型

官方介绍

RabbitMQ是一个消息代理：它接收和转发消息。你可以把它想像成一个邮政信箱

RabbitMQ与邮局的主要区别是它不处理纸张，而是接受，存储和转发数据消息的二进制数据块。

P（producer/ publisher）：生产者，一个发送消息的用户应用程序。

C（consumer）：消费者，消费和接收有类似的意思，消费者是一个主要用来等待接收消息的用户应用程序

队列（红色区域）：rabbitmq内部类似于邮箱的一个概念。虽然消息流经rabbitmq和你的应用程序，但是它们只能存储在队列中。队列只受主机的内存和磁盘限制，实质上是一个大的消息缓冲区。许多生产者可以发送消息到一个队列，许多消费者可以尝试从一个队列接收数据。

总之：

生产者将消息发送到队列，消费者从对列表中获取消息，队列是存储消息的缓冲区。

我们将java编写两个程序；发送单个消息的生产者，以及接受消息并将打印出来的消费者。我们将详细介绍java API中的一些细节，这是一个消费传递的"Hello World"。

我们将调用我们的消息发布者（发送者）Send和我们的消息消费者（接受者）Recv。发布者将连接到RabbitMQ，发送一条消息，然后退出。

4.1.1、生产者发送消息

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.bjpowernode.util.ConnectionUtil;

public class Send {

    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接以及mq通道
        Connection connection = ConnectionUtil.getConnection();
        // 轻量级的 Connection，这是完成大部分API的地方。
        Channel channel = connection.createChannel();

        // 声明（创建）队列，必须声明队列才能够发送消息，我们可以把消息发送到队列中。
        // 声明一个队列是幂等的 - 只有当它不存在时才会被创建
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 消息内容
        String message = "Hello World!";
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");

        //关闭通道和连接
        channel.close();
        connection.close();
    }
}

控制台：

4.1.2、管理工具中查看消息

进入队列页面，可以看到新建的一个队列：simple_queue

点击队列名称，进入详情页，可以查看消息：

在控制台查看消息并不会将消息消费，所以消息还在。

4.1.3、消费者获取消息

import java.io.IOException;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.bjpowernode.util.ConnectionUtil;

public class Recv {
    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 
                          BasicProperties properties,byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [x] received : " + msg + "!");
            }
        };
        // 监听队列，第二个参数：是否自动进行消息确认。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

控制台：

这个时候，队列中的消息就没了：

我们发现，消费者可以获取了消息，但是程序没有停止，一直在监听队列中是否有新的消息。一旦有新的消息进入队列，就会立即打印。

4.1.4消息确认机制（ACK）

通过刚才的案例可以看出消息一旦被消费者接收，队列中的消息就会被删除。

那么问题来了：RabbitMQ怎么知道消息被接受了呢？

如果消费者领取到消息后，还没有执行操作就挂掉了呢？或者抛出了异常？消费者消费失败，但是RabbitMQ无从可知，这样消息就丢失了！

因此，RabbitMQ有一个ACK机制。当消费者获取消息后，会向RabbitMQ发送回执ACK，告知消息已经被接收。不过这种回执ACK分两种情况：

1.自动回执ACK：消息一旦被接受，消费者自动发送ACK

2.手动ACK：消费接收后，不会发送ACK，需要手动调用

大家觉得哪种更好呢？

这需要看消息的重要性：

如果消息不太重要，丢失也没有影响，那么自动ACK比较方便

果消息非常重要，不容丢失。那么最好在消费完成后手动ACK，否则接收消息后就自动ACK，RabbitMQ就会把消息从队列中删除。如果此时消费者宕机，那么消息就丢失了。

我们之前的测试都是自动ACK的，如果要手动ACK，需要改动我们的代码：

public class Recv2 {
    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建通道
        final Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 
                          BasicProperties properties,byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [x] received : " + msg + "!");
                /**
                 * 手动进行ACK
                 *  deliveryTag：该消息的index
                 *  multiple：是否批量处理.true:将一次性ack所有消息。
                 */
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        // 监听队列，第二个参数false，手动进行ACK
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

注意到最后一行代码：

channel.basicConsume(QUEUE_NAME, false, consumer);

如果第二个参数为true，则会自动进行ACK；如果为false，则需要手动ACK。方法的声明：

4.1.1、自动ACK存在的问题

修改消费者，添加异常，如下：

生产者不做任何修改，直接运行，消息发送成功：

运行消费者，程序抛出异常。但是消息依然被消费：

管理界面：

4.1.2、演示手动ACK

修改消费者，把自动改成手动（去掉之前制造的异常）

生产者不变，再次运行：

运行消费者

但是，查看管理界面，发现：

停掉消费者的程序，发现：

这是因为虽然我们设置了手动ACK，但是代码中并没有进行消息确认！所以消息并未被真正消费掉。

当我们关掉这个消费者，消息的状态再次称为Ready

修改代码手动ACK：

执行：

消息消费成功！

4.2、work-工作模型

工作队列或者竞争消费者模式

在第一篇教程中，我们编写了一个程序，从一个命名队列中发送并接受消息。在这里，我们将创建一个工作队列，在多个工作者之间分配耗时任务。

工作队列，又称任务队列。主要思想就是避免执行资源密集型任务时，必须等待它执行完成。相反我们稍后完成任务，我们将任务封装为消息并将其发送到队列。在后台运行的工作进程将获取任务并最终执行作业。当你运行许多消费者时，任务将在他们之间共享，但是一个消息只能被一个消费者获取。

这个概念在Web应用程序中特别有用，因为在短的HTTP请求窗口中无法处理复杂的任务。

接下来我们来模拟这个流程：

P：生产者：任务的发布者

C1：消费者，领取任务并且完成任务，假设完成速度较快

C2：消费者2：领取任务并完成任务，假设完成速度慢

面试题：避免消息堆积？面试题：避免消息堆积？

1）采用workqueue，多个消费者监听同一队列。

2）接收到消息以后，而是通过线程池，异步消费。

4.2.1、生产者

生产者与案例1中的几乎一样：

public class Send {
    private final static String QUEUE_NAME = "test_work_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 循环发布任务
        for (int i = 0; i < 50; i++) {
            // 消息内容
            String message = "task .. " + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");

            Thread.sleep(i * 2);
        }
        // 关闭通道和连接
        channel.close();
        connection.close();
    }
}

不过这里我们是循环发送50条消息。

4.2.2消费者1

4.2.3消费者2

与消费者1基本类似，就是没有设置消费耗时时间。

这里是模拟有些消费者快，有些比较慢。

接下来，两个消费者一同启动，然后发送50条消息：

可以发现，两个消费者各自消费了25条消息，而且各不相同，这就实现了任务的分发。

4.2.4、能者多劳

刚才的实现有问题吗？

消费者1比消费者2的效率要低，一次任务的耗时较长
然而两人最终消费的消息数量是一样的
消费者2大量时间处于空闲状态，消费者1一直忙碌

现在的状态属于是把任务平均分配，正确的做法应该是消费越快的人，消费的越多。

怎么实现呢？

我们可以使用basicQos方法和prefetchCount = 1设置。这告诉RabbitMQ一次不要向工作人员发送多于一条消息。或者换句话说，不要向工作人员发送新消息，直到它处理并确认了前一个消息。相反，它会将其分派给不是仍然忙碌的下一个工作人员。

再次测试：

4.3、Fanout-广播模型

Fanout，也称为广播。

流程图：

在广播模式下，消息发送流程是这样的：

1）可以有多个消费者
2）每个消费者有自己的queue（队列）
3）每个队列都要绑定到Exchange（交换机）
4） 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
5）交换机把消息发送给绑定过的所有队列
6）队列的消费者都能拿到消息。实现一条消息被多个消费者消费

4.3.1、生产者

两个变化：

1）声明Exchange，不再声明Queue
2）发送消息到Exchange，不再发送到Queue

public class Send {

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        
        // 声明exchange，指定类型为fanout
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        
        // 消息内容
        String message = "Hello everyone";
        // 发布消息到Exchange
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        System.out.println(" [生产者] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}

4.3.2、消费者1

public class Recv {
    private final static String QUEUE_NAME = "fanout_exchange_queue_1";

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 
                          BasicProperties properties,byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者1] received : " + msg + "!");
            }
        };
        // 监听队列，自动返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

要注意代码中：队列需要和交换机绑定

4.3.3、消费者2

public class Recv2 {
    private final static String QUEUE_NAME = "fanout_exchange_queue_2";

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 
                          BasicProperties properties,byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者2] received : " + msg + "!");
            }
        };
         // 监听队列，自动返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

4.3.4、测试

我们运行两个消费者，然后发送1条消息：

4.4.Direct-定向模型

有选择性的接收消息

在订阅模式中，生产者发布消息，所有消费者都可以获取所有消息。

在路由模式中，我们将添加一个功能 - 我们将只能订阅一部分消息。例如，我们只能将重要的错误消息引导到日志文件（以节省磁盘空间），同时仍然能够在控制台上打印所有日志消息。

但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

在Direct模型下，队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）

消息的发送方在向Exchange发送消息时，也必须指定消息的routing key。

P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。

X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列

C1：消费者，其所在队列指定了需要routing key 为 error 的消息

C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

4.4.1.生产者

此处我们模拟商品的增删改，发送消息的RoutingKey分别是：insert、update、delete

public class Send {
    private final static String EXCHANGE_NAME = "direct_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为direct
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        // 消息内容
        String message = "商品新增了， id = 1001";
        // 发送消息，并且指定routing key 为：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());
        System.out.println(" [商品服务：] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}

4.4.2.消费者1

我们此处假设消费者1只接收两种类型的消息：更新商品和删除商品。

public class Recv {
    private final static String QUEUE_NAME = "direct_exchange_queue_1";
    private final static String EXCHANGE_NAME = "direct_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 绑定队列到交换机，同时指定需要订阅的routing key。假设此处需要update和delete消息
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 
                                                    BasicProperties properties,
                                                byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者1] received : " + msg + "!");
            }
        };
        // 监听队列，自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

4.4.3.消费者2

我们此处假设消费者2接收所有类型的消息：新增商品，更新商品和删除商品。

public class Recv2 {
    private final static String QUEUE_NAME = "direct_exchange_queue_2";
    private final static String EXCHANGE_NAME = "direct_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 绑定队列到交换机，同时指定需要订阅的routing key。订阅 insert、update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 
                                                    BasicProperties properties,
                                                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者2] received : " + msg + "!");
            }
        };
        // 监听队列，自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

4.4.4.测试

我们分别发送增、删、改的RoutingKey，发现结果：

4.5.Topic-主题模型

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：

`#`：匹配一个或多个词
`*`：匹配不多不少恰好1个词

举例：

`audit.#`：能够匹配`audit.irs.corporate` 或者 `audit.irs`
`audit.*`：只能匹配`audit.irs`

在这个例子中，我们将发送所有描述动物的消息。消息将使用由三个字（两个点）组成的routing key发送。路由关键字中的第一个单词将描述速度，第二个颜色和第三个种类：“<speed>.<color>.<species>”。

我们创建了三个绑定：Q1绑定了绑定键“*.orange.*”，Q2绑定了“*.*.rabbit”和“lazy.＃”。

Q1匹配所有的橙色动物。

Q2匹配关于兔子以及懒惰动物的消息。

练习，生产者发送如下消息，会进入那个队列：

quick.orange.rabbit

lazy.orange.elephant

quick.orange.fox

lazy.pink.rabbit

quick.brown.fox

quick.orange.male.rabbit

orange

4.5.1.生产者

使用topic类型的Exchange，发送消息的routing key有3种： item.isnert、item.update、item.delete：

public class Send {
    private final static String EXCHANGE_NAME = "topic_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为topic
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        // 消息内容
        String message = "新增商品 : id = 1001";
        // 发送消息，并且指定routing key 为：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "item.insert", null, message.getBytes());
        System.out.println(" [商品服务：] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}

4.5.2.消费者1

我们此处假设消费者1只接收两种类型的消息：更新商品和删除商品

public class Recv {
    private final static String QUEUE_NAME = "topic_exchange_queue_1";
    private final static String EXCHANGE_NAME = "topic_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 绑定队列到交换机，同时指定需要订阅的routing key。需要 update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");

        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 
                                       BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者1] received : " + msg + "!");
            }
        };
        // 监听队列，自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

4.5.3.消费者2

我们此处假设消费者2接收所有类型的消息：新增商品，更新商品和删除商品。

/**
 * 消费者2
 */
public class Recv2 {
    private final static String QUEUE_NAME = "topic_exchange_queue_2";
    private final static String EXCHANGE_NAME = "topic_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 绑定队列到交换机，同时指定需要订阅的routing key。订阅 insert、update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.*");

        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者2] received : " + msg + "!");
            }
        };
        // 监听队列，自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

五、持久化

如何避免消息丢失？

1）消费者的手动ACK机制。可以防止业务处理失败。

2）但是，如果在消费者消费之前，MQ就宕机了，消息就没了。

是可以将消息进行持久化呢？

要将消息持久化，前提是：队列、Exchange都持久化

5.1.交换机持久化

5.2.队列持久化

5.3.消息持久化

5.4.测试

分别测试持久化和非持久化：

1、Send给Recv发送50条消息

2、Recv收到一条消息sleep1秒钟，收到前几条消息后立即关闭

3、重启RabbitMQ观察消息是否丢失

六、Spring AMQP

6.1.简介

Spring有很多不同的项目，其中就有对AMQP的支持：

Spring AMQP的页面：Spring AMQPLevel up your Java code and explore what Spring can do for you.http://spring.io/projects/spring-amqp

注意这里一段描述：

Spring-amqp是对AMQP协议的抽象实现，而spring-rabbit 是对协议的具体实现，也是目前的唯一实现。底层使用的就是RabbitMQ。

6.2.依赖和配置

添加AMQP的启动器：

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

在application.properties中添加RabbitMQ地址：

#主机 
spring.rabbitmq.host=192.168.233.132 
#端口 
spring.rabbitmq.port=5672 
#用户名 
spring.rabbitmq.username=admin 
#密码 
spring.rabbitmq.password=1111 #虚拟分组 spring.rabbitmq.virtual-host=/

6.3.监听者

在SpringAmqp中，对消息的消费者进行了封装和抽象，一个普通的JavaBean中的普通方法，只要通过简单的注解，就可以成为一个消费者。

@Component
public class Listener {

    /**
     * 监听者接收消息三要素：
     *  1、queue
     *  2、exchange
     *  3、routing key
     */
    @RabbitListener(bindings = @QueueBinding(
        value = @Queue(value="springboot_queue",durable = "true"),
        exchange = @Exchange(value="springboot_exchage",type= ExchangeTypes.TOPIC),
        key= {"*.*"}
    ))
    public void listen(String msg){
        System.out.println("接收到消息：" + msg);
    }
}

@Componet：类上的注解，注册到Spring容器
@RabbitListener：方法上的注解，声明这个方法是一个消费者方法，需要指定下面的属性：
- bindings：指定绑定关系，可以有多个。值是@QueueBinding的数组。@QueueBinding包含下面属性：
  - value：这个消费者关联的队列。值是@Queue，代表一个队列
  - exchange：队列所绑定的交换机，值是@Exchange类型
  - key：队列和交换机绑定的RoutingKey

类似listen这样的方法在一个类中可以写多个，就代表多个消费者。

6.4.AmqpTemplate

Spring最擅长的事情就是封装，把他人的框架进行封装和整合。

Spring为AMQP提供了统一的消息处理模板：AmqpTemplate，非常方便的发送消息，其发送方法：

红框圈起来的是比较常用的3个方法，分别是：

指定交换机、RoutingKey和消息体
指定消息
指定RoutingKey和消息，会向默认的交换机发送消息

6.5.发送者

@RunWith(SpringRunner.class)
@SpringBootTest(classes = App.class)
public class MqDemo {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Test
    public void testSend() throws InterruptedException {
        String msg = "hello, Spring boot amqp";
        this.amqpTemplate.convertAndSend("spring.test.exchange","a.b", msg);
        // 等待10秒后再结束
        Thread.sleep(10000);
    }
}

运行后查看日志：

6.6.手动ack

6.6.1.application.properties

#设置三种订阅模式手动
ack spring.rabbitmq.listener.direct.acknowledge-mode=manual 
#设置work消息类型手动
ack spring.rabbitmq.listener.simple.acknowledge-mode=manual

6.6.2.监听者

注意：监听方法添加Channel channel, Message message两个参数


import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
public class Recv {
    /**
     * 监听者接收消息三要素：
     *  1、queue
     *  2、exchange
     *  3、routing key
     */
    @RabbitListener(bindings = @QueueBinding(
        value = @Queue(value="springboot_queue",durable = "true"),
        exchange = @Exchange(value="springboot_exchage",type= ExchangeTypes.TOPIC),
        key= {"*.*"}
    ))
    public void listen(String msg, Channel channel, Message message) throws IOException {
        System.out.println("接收到消息：" + msg);
        //int a = 6/0;
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }
}

RabbitMQ介绍