RabbitMQ详解（上） - 蚂蚁小哥

①：生产者（Producer）
    产生数据发送消息的程序是生产者
②：交换机（Exchange）
    交换机是 RabbitMQ 非常重要的一个部件，一方面它接收来自生产者的消息，另一方面它将消息推送到队列中。交换机必须确切知道如何
　　处理它接收到的消息，是将这些消息推送到特定队列还是推送到多个队列，亦或者是把消息丢弃，这个得有交换机类型决定
③：队列（Queue）
    队列是 RabbitMQ 内部使用的一种数据结构，尽管消息流经 RabbitMQ 和应用程序，但它们只能存储在队列中。队列仅受主机的内存和
　　磁盘限制的约束，本质上是一个大的消息缓冲区。许多生产者可以将消息发送到一个队列，许多消费者可以尝试从一个队列接收数据。这就是
　　我们使用队列的方式
④：消费者（Consumer）
    消费与接收具有相似的含义。消费者大多时候是一个等待接收消息的程序。请注意生产者，消费者和消息中间件很多时候并不在同一机器上。
　　同一个应用程序既可以是生产者又是可以是消费者。

Broker：
    接收和分发消息的应用，RabbitMQ Server就是Message Broker；简单来说就是消息队列服务器实体Virtual host：出于多租户和
　　 安全因素设计的，把AMQP的基本组件划分到一个虚拟的分组中，类似于网络中的namespace概念。当多个不同的用户使用同一个
　　 RabbitMQ server提供的服务时，可以划分出多个vhost，每个用户在自己的vhost创建exchange／queue等
Connection：
    publisher／consumer和broker之间的TCP连接
Channel：
    消息通道，如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCPConnection的开销将是巨大的，效率也较
　　 低。Channel是在connection内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的channel进行通讯，AMQP
　　 method包含了channel id帮助客户端和message broker识别channel，所以channel之间是完全隔离的。Channel作为轻量级的
　　 Connection极大减少了操作系统建立TCP connection的开销。
Exchange：
　　message到达broker的第一站，根据分发规则，匹配查询表中的routing key，分发消息到queue中去。
　　常用的类型有：direct (point-to-point), topic (publish-subscribe) and fanout(multicast)
Queue：
    消息队列载体，每个消息都会被投入到一个或多个队列;消息最终被送到这里等待consumer取走
Routing Key：
    路由关键字，exchange根据这个关键字进行消息投递。
Binding：
　　exchange和queue之间的虚拟连接，binding中可以包含routing key，Binding信息被保存到exchange中的查询表中，用于message
　　的分发依据;它的作用就是把exchange和queue按照路由规则绑定起来。
producer：
    消息生产者，就是投递消息的程序。
consumer：
    消息消费者，就是接受消息的程序。

# 安装 erlang 环境
rpm -ivh erlang-23.3.4.11-1.el7.x86_64.rpm
# 安装 socat 环境
yum -y install socat
# 安装 RabbitMQ 服务
rpm -ivh rabbitmq-server-3.9.13-1.el7.noarch.rpm
# 检查是否安装
yum list | grep rabbitmq
yum list | grep erlang

补充命令：
# 查看所以的已开启的端口
firewall-cmd --zone=public --list-ports
# 开启15672端口（--permanent代表永久生效，重启系统不会失效）
firewall-cmd --zone=public --add-port=15672/tcp --permanent

# 防火墙关闭
systemctl stop firewalld
# 启动RabbitMQ服务
systemctl start rabbitmq-server
    或 /sbin/service rabbitmq-server start
    或 service rabbitmq-server start
# 添加开机启动RabbitMQ服务
chkconfig rabbitmq-server on 
# 开启web管理接口（可以更方便快速的对RabbitMQ进行操作）
rabbitmq-plugins enable rabbitmq_management
# 停止RabbitMQ服务
systemctl stop rabbitmq-server
    或 /sbin/service rabbitmq-server stop
    或 service rabbitmq-server stop

注：web管理接口应用的操作
rabbitmqctl stop_app　　停止web页面
rabbitmqctl start_app  启动web页面

实际操作说明：
rabbitmqctl list_users
    # 查看RabbitMQ里的所有用户
rabbitmqctl list_vhosts
    # 查看RabbitMQ里的所有vhosts
rabbitmqctl list_permissions
    # 查看RabbitMQ里所有用户的权限
rabbitmqctl list_user_permissions guest
    # 查看RabbitMQ里guest用户的权限

rabbitmqctl add_vhost test
    # 创建的一个虚拟主机项为 test 的名称
rabbitmqctl add_user admin 123
    # 创建一个用户为admin 密码为123
rabbitmqctl set_user_tags admin administrator
    # 设置admin的角色为超级管理员（administrator）
rabbitmqctl set_permissions -p test admin ".*" ".*" ".*"
    # 设置admin在test的vhost中，并设置全部文件的读写操作 
rabbitmqctl list_permissions -p test
    # 查看test中的vhost里的用户

systemctl stop rabbitmq-server
    # 停止RabbitMQ服务
yum list | grep rabbitmq
    # 查看RabbitMQ安装的相关列表
yum -y remove rabbitmq-server.noarch
    # 卸载RabbitMQ已安装的相关内容
yum list | grep erlang
    # 查看erlang安装的相关列表
yum -y remove erlang-*
yum remove erlang.x86_64
    # 卸载erlang已安装的相关内容
rm -rf /usr/lib64/erlang 
rm -rf /var/lib/rabbitmq
rm -rf /usr/local/erlang
rm -rf /usr/local/rabbitmq
    # 删除有关的所有文件

package cn.xw.helloWorld;

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

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2022-03-04 17:42
 */
public class Producer {
    //简单队列名称
    public static final String QUEUE_NAME = "helloWorldQueue";

    public static void main(String[] args) throws IOException, TimeoutException {
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置RabbitMQ服务的IP、账号、密码、Vhost虚拟主机(默认 "/" 则不需要设置)
        factory.setHost("192.168.31.51");
        factory.setUsername("admin");
        factory.setPassword("123");
        factory.setVirtualHost("test");
        //通过工厂对象获取一个连接
        Connection connection = factory.newConnection();
        //通过连接来获取一个信道
        Channel channel = connection.createChannel();
        //声明一个队列
        //参数一：队列名称
        //参数二：队列里的消息是否持久化，默认消息保存在内存中，默认false
        //参数三：该队列是否只供一个消费者进行消费的独占队列，则为 true（仅限于此连接），false（默认，可以多个消费者消费）
        //参数四：是否自动删除 最后一个消费者断开连接以后 该队列是否自动删除 true 自动删除，默认false
        //参数五：构建队列的其它属性，看下面扩展参数
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        //发送的消息
        byte[] msg = "这是一个简单消息".getBytes(StandardCharsets.UTF_8);
        //发送消息
        //参数一：将发送到RabbitMQ的哪个交换机上
        //参数二：路由的key是什么（直接交换机找到路由后，通过路由key来确定最终的队列）
        //参数三：其它参数
        //参数四：发送到队列的具体信息
        channel.basicPublish("", QUEUE_NAME, null, msg);
        System.out.println("消息发送完成！");
    }
}

package cn.xw.helloWorld;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2022-03-05 15:12
 */
public class Consumer {
    //简单队列名称
    public static final String QUEUE_NAME = "helloWorldQueue";

    public static void main(String[] args) throws IOException, TimeoutException {
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置RabbitMQ服务的IP、账号、密码、Vhost虚拟主机(默认 "/" 则不需要设置)
        factory.setHost("192.168.31.51");
        factory.setUsername("admin");
        factory.setPassword("123");
        factory.setVirtualHost("test");
        //通过工厂对象获取一个连接
        Connection connection = factory.newConnection();
        //通过连接来获取一个信道
        Channel channel = connection.createChannel();
        System.out.println("消费者开始监听队列消息....");
        //推送的消息如何进行消费的接口回调
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                System.out.println("获取队列信息：" + new String(message.getBody(), StandardCharsets.UTF_8));
            }
        };
        //取消消费的一个回调接口 如在消费的时候队列被删除掉了
        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
                System.out.println("监听的队列出现异常；可能队列被删除！");
            }
        };
        //消费者消费消息
        //参数一：消费哪个队列
        //参数二：消费成功之后是否要自动应答 true 代表自动应答 false 手动应答
        //参数三：接受队列消息的回调接口
        //参数四：取消消费的回调接口
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
    }
}

x-dead-letter-exchange：
    死信交换器
x-dead-letter-routing-key：
    死信消息的可选路由键
x-expires：
    队列在指定毫秒数后被删除
x-message-ttl：
    毫秒为单位的消息过期时间，队列级别
x-ha-policy：
    创建HA队列，此参数已失效
x-ha-nodes：
    HA队列的分布节点，此参数已失效
x-max-length：
    队列的消息条数限制。限制加入queue中消息的条数。先进先出原则，超过后，后面的消息会顶替前面的消息。
x-max-length-bytes：
    消息容量限制，该参数和x-max-length目的一样限制队列的容量，但是这个是靠队列大小（bytes）来达到限制。
x-max-priority：
    最大优先值为255的队列优先排序功能
x-overflow：
    设置队列溢出行为。这决定了当达到队列的最大长度时消息会发生什么。
    有效值是drop-head、reject-publish或reject-publish-dlx。
x-single-active-consumer：
    表示队列是否是单一活动消费者，true时，注册的消费组内只有一个消费者消费消息，
    其他被忽略，false时消息循环分发给所有消费者(默认false)
x-queue-mode：
    将队列设置为延迟模式，在磁盘上保留尽可能多的消息，以减少RAM的使用；
    如果未设置，队列将保留内存缓存以尽可能快地传递消息
x-queue-master-locator：
    在集群模式下设置镜像队列的主节点信息

public class ProducerA {
    //工作队列名称
    public static final String QUEUE_NAME = "workQueue";

    public static void main(String[] args) throws IOException {
        //调用自己的工具类获取信道
        Channel channel = ChannelUtil.getChannel();
        //创建队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        //循环发送消息
        for (int i = 0; i < 1000; i++) {
            byte[] msg = ("这是一个编号为：" + i + " 的待处理的消息").getBytes(StandardCharsets.UTF_8);
            channel.basicPublish("", QUEUE_NAME, null, msg);
        }
        System.out.println("消息发送完成！");
    }
}

public class ConsumerA {
    //工作队列名称
    public static final String QUEUE_NAME = "workQueue";

    public static void main(String[] args) throws IOException {
        //调用自己的工具类获取信道
        Channel channel = ChannelUtil.getChannel();
        //创建队列 以防启动消费者发现队列不存在报错
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        System.out.println("消费者A开始监听队列消息....");
        //消费者消费消息
        channel.basicConsume(QUEUE_NAME, true, (consumerTag, message) -> {
            System.out.println("A消费者获取队列信息并处理：" + new String(message.getBody(), StandardCharsets.UTF_8));
        }, consumerTag -> {
            System.out.println("监听的队列出现异常；可能队列被删除！");
        });
    }
}

public class ConsumerB {
    //工作队列名称
    public static final String QUEUE_NAME = "workQueue";

    public static void main(String[] args) throws IOException {
        //调用自己的工具类获取信道
        Channel channel = ChannelUtil.getChannel();
        //创建队列 以防启动消费者发现队列不存在报错
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        System.out.println("消费者B开始监听队列消息....");
        //消费者消费消息
        channel.basicConsume(QUEUE_NAME, true, (consumerTag, message) -> {
            System.out.println("B消费者获取队列信息并处理：" + new String(message.getBody(), StandardCharsets.UTF_8));
        }, consumerTag -> {
            System.out.println("监听的队列出现异常；可能队列被删除！");
        });
    }
}

注：basicConsume消息接收方法中的autoAck参数必须为false才可以显示为手动确认
手动应答分为三种情况:
②：手动拒绝 basicReject(long deliveryTag, boolean requeue)：
    拒绝deliveryTag对应的消息，第二个参数是否requeue，true则重新入队列，否则丢弃或者进入死信队列。
③：手动不确认 basicNack(long deliveryTag, boolean multiple, boolean requeue)
    不确认deliveryTag对应的消息，第二个参数是否应用于多消息，第三个参数是否requeue，与basic.reject区别就是同时支持多个消息，
    可以nack该消费者先前接收未ack的所有消息。nack后的消息也会被自己消费到。
③：手动恢复 basicRecover(boolean requeue)
    是否恢复消息到队列，true则重新入队列，并且尽可能的将之前recover的消息投递给其他消费者消费，而不是自己再次消费。
    false则消息会重新被投递给自己。
④：手动应答 basicAck(long deliveryTag, boolean multiple)
    如果消费者在处理消息的过程中，出了错，就没有什么办法重新处理这条消息，所以在平时都是处理消息成功后，再确认消息；
    当autoAck=false时，RabbitMQ会等待消费者手动发回ack信号后,才从内存（和磁盘，如果是持久化消息的话）中移除消息。
    它采用消息确认机制，消费者就有足够的时间处理消息(任务)，不用担心处理消息过程中消费者进程挂掉后消息丢失的问题，
    因为RabbitMQ会一直持有消息直到消费者手动调用channel.basicAck为止。对于RabbitMQ服务器端而言，如果服务器端一直没
    有收到消费者的ack信号，并且消费此消息的消费者已经断开连接，则服务器端会安排该消息重新进入队列，等待投递给下一个
    消费者（也可能还是原来的那个消费者）。这里我们启动了手动确认后，就必须调用channel.basicAck方法进行确认，
    否则的话RabbitMQ会一直进行等待，当我们这个消费者关闭后，RabbitMQ会将该条消息再发给对应的消费者进行消费，
    直到有消费者对该条消息进行消费并应答完成。
参数说明：
    deliveryTag：对应消息的ID；通过message.getEnvelope().getDeliveryTag()获取
    requeue：是否重新入列，true代表拒绝应答后会重新返回队列，false则直接删除或者进入死信队列
    multiple：是否批量应答，true代表批量应答
        假设有个队列依次排列为 1、2、3...10 (1最先出队，10最后出队)；
        当为true，发送1~5消息给消费者处理完都未确认，当到第6时执行应答方法，并且multiple为true，则代表1~6都被被批量应答
        当为false，发送1~5消息给消费者处理完都未确认，当到第6时执行应答方法，并且multiple为true，则代表只要6被应答

//生产者只管发任务消息，代码不变，消费者代码优化更改以下，多个消费者代码也和这一样
public class ConsumerB {
    //工作队列名称
    public static final String QUEUE_NAME = "workQueue";

    public static void main(String[] args) throws IOException {
        //调用自己的工具类获取信道
        Channel channel = ChannelUtil.getChannel();
        //创建队列 以防启动消费者发现队列不存在报错
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        System.out.println("消费者B开始监听队列消息....");
        //应答方式 true自动应答  false手动应答(若是手动应答必须设置false)
        boolean autoAck = false;
        //消费者消费消息requeue
        channel.basicConsume(QUEUE_NAME, autoAck, (consumerTag, message) -> {
            try {
                //这里我就一句打印语句，没有复杂逻辑，正常这里有复杂业务
                System.out.println("B消费者获取队列信息并处理：" + new String(message.getBody(), StandardCharsets.UTF_8));
                int i = 1/0;
                //手动确认应答 不批量应答
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            } catch (IOException e) {
                e.printStackTrace();
                //出现异常手动进行不应答;并且放入队列中（reject或者使用uack方式都可以，或者本次消息不处理了可以通过recover重新放到队列）
                channel.basicReject(message.getEnvelope().getDeliveryTag(), true);
            }
        }, consumerTag -> {
            System.out.println("监听的队列出现异常；可能队列被删除！");
        });
    }
}

与交换机的持久化相同，队列的持久化也是通过durable参数实现的(设置后队列也会有个D)，看一下方法的定义：
queueDeclare(String queue,boolean durable,boolean exclusive,boolean autoDelete,Map<String, Object> arguments)
boolean durable：
    参数跟交换机方法的参数一样，true表示做持久化，当RabbitMQ服务重启时，队列依然存在
boolean exclusive（补充）：
    排它队列。如果一个队列被声明为排他队列，那么这个队列只能被第一次声明它的连接所见，并在连接断开的时候自动删除。
    这里有三点需要说明：
        1：排它队列是基于连接可见的，同一连接的不同信道是可以同时访问同一连接创建的排它队列
        2：如果一个连接已经声明了一个排它队列，其它连接是不允许建立同名的排它队列的，这个与普通队列不同
        3：即使该队列是持久化的，一旦连接关闭或者客户端退出，该排它队列都会被自动删除的，这种队列适用于一
        个客户端发送读取消息的应用场景
boolean autoDelete（补充）：
    自动删除，如果该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列

basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
BasicProperties props设置消息持久化方式：
参数实现类：
public static class BasicProperties extends com.rabbitmq.client.impl.AMQBasicProperties {
        private String contentType;　　　　　　       //  消息的内容类型，如：text/plain
        private String contentEncoding;             //  消息内容编码
        private Map<String,Object> headers;         //  设置消息的header,类型为Map<String,Object>
        private Integer deliveryMode;               //  1（nopersistent）非持久化，2（persistent）持久化
        private Integer priority;                   //  消息的优先级
        private String correlationId;               //  关联ID
        private String replyTo;                     //  用于指定回复的队列的名称
        private String expiration;                  //  消息的失效时间
        private String messageId;                   //  消息ID
        private Date timestamp;                     //  消息的时间戳
        private String type;                        //  类型
        private String userId;                      //  用户ID
        private String appId;                       //  应用程序ID
        private String clusterId;                   //  集群ID
}
deliveryMode是设置消息持久化的参数，等于1不设置持久化，等于2设置持久化；
我们平时不会使用BasicProperties类而是使用MessageProperties，通过这个类来获取具体配置
设置 MessageProperties.PERSISTENT_TEXT_PLAIN
代表：
public static final BasicProperties PERSISTENT_TEXT_PLAIN =
    new BasicProperties("text/plain",null,null,2,0, null, null, null,null, null, null, null,null, null);

//也可以通过这种方式设置；发送消息的参数设置 expiration过期时间   deliveryMode 消息持久化方式
AMQP.BasicProperties properties = new AMQP.BasicProperties()
        .builder().expiration("10000").deliveryMode(2).build();

public class ConsumerA {
    //工作队列名称
    public static final String QUEUE_NAME = "workQueue";

    public static void main(String[] args) throws IOException {
        //调用自己的工具类获取信道
        Channel channel = ChannelUtil.getChannel();
        //创建队列 以防启动消费者发现队列不存在报错
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        //设置0代表轮询分发、1不公平分发、大于1代表预取值
        channel.basicQos(1);
        System.out.println("消费者A(处理资源很快)开始监听队列消息....");
        //应答方式 true自动应答  false手动应答(若是手动应答必须设置false)
        boolean autoAck = false;
        //消费者消费消息requeue
        channel.basicConsume(QUEUE_NAME, autoAck, (consumerTag, message) -> {
            try {
                //这里我就一句打印语句，没有复杂逻辑，正常这里有复杂业务
                System.out.println("A消费者获取队列信息并处理：" + new String(message.getBody(), StandardCharsets.UTF_8));
                Thread.sleep(3000);    //3秒才能处理完一个任务消息
                //手动确认应答 不批量应答
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            } catch (IOException | InterruptedException e) {
                e.printStackTrace();
                //出现异常手动进行不应答;并且放入队列中
                channel.basicReject(message.getEnvelope().getDeliveryTag(), true);
            }
        }, consumerTag -> {
            System.out.println("监听的队列出现异常；可能队列被删除！");
        });
    }
}
//消费者A(处理资源很快)开始监听队列消息....
//A消费者获取队列信息并处理：这是一个编号为：0 的待处理的消息
//A消费者获取队列信息并处理：这是一个编号为：2 的待处理的消息
//A消费者获取队列信息并处理：这是一个编号为：3 的待处理的消息
//A消费者获取队列信息并处理：这是一个编号为：4 的待处理的消息
//A消费者获取队列信息并处理：这是一个编号为：6 的待处理的消息
//A消费者获取队列信息并处理：这是一个编号为：7 的待处理的消息
//A消费者获取队列信息并处理：这是一个编号为：8 的待处理的消息

public class ConsumerB {
    //工作队列名称
    public static final String QUEUE_NAME = "workQueue";

    public static void main(String[] args) throws IOException {
        //调用自己的工具类获取信道
        Channel channel = ChannelUtil.getChannel();
        //创建队列 以防启动消费者发现队列不存在报错
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        //设置0代表轮询分发、1不公平分发、大于1代表预取值
        channel.basicQos(1);
        System.out.println("消费者B(处理资源很慢)开始监听队列消息....");
        //应答方式 true自动应答  false手动应答(若是手动应答必须设置false)
        boolean autoAck = false;
        //消费者消费消息requeue
        channel.basicConsume(QUEUE_NAME, autoAck, (consumerTag, message) -> {
            try {
                //这里我就一句打印语句，没有复杂逻辑，正常这里有复杂业务
                System.out.println("B消费者获取队列信息并处理：" + new String(message.getBody(), StandardCharsets.UTF_8));
                Thread.sleep(10000);    //10秒才能处理完一个任务消息
                //手动确认应答 不批量应答
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            } catch (IOException | InterruptedException e) {
                e.printStackTrace();
                //出现异常手动进行不应答;并且放入队列中
                channel.basicReject(message.getEnvelope().getDeliveryTag(), true);
            }
        }, consumerTag -> {
            System.out.println("监听的队列出现异常；可能队列被删除！");
        });
    }
}
//消费者B(处理资源很慢)开始监听队列消息....
//B消费者获取队列信息并处理：这是一个编号为：1 的待处理的消息
//B消费者获取队列信息并处理：这是一个编号为：5 的待处理的消息
//B消费者获取队列信息并处理：这是一个编号为：9 的待处理的消息

public class ProducerA {
    //单个发布确认
    public static final String SINGLE_RELEASE_CONFIRMATION = "singleReleaseConfirmation";
public static void main(String[] args) throws IOException, InterruptedException {
        long begin = System.currentTimeMillis();    //记录开始时间
        //获取信道
        Channel channel = ChannelUtil.getChannel();
        //创建一个信道
        channel.queueDeclare(SINGLE_RELEASE_CONFIRMATION, true, false, false, null);
        //开启发布确认功能
        channel.confirmSelect();
        //循环发送消息
        for (int i = 0; i < 1000; i++) {
            String str = "单个发布确认信息" + i;
            System.out.println("开始发送信息：" + i);
            //发布信息
            channel.basicPublish("", SINGLE_RELEASE_CONFIRMATION, MessageProperties.PERSISTENT_TEXT_PLAIN,
                    str.getBytes(StandardCharsets.UTF_8));
            //验证是否发送成功（等待确认）
            //channel.waitForConfirms(3000); 发送三秒后没得到回复将断定未发送过去
            boolean b = channel.waitForConfirms();
            if (b) {
                System.out.println("发送成功了：" + i);
            }
        }
        long end = System.currentTimeMillis();  //记录结尾时间
        System.out.println("单个发布确认用时：" + (end - begin)); //单个发布确认用时：2278
    }
}

public class ProducerA {
    //批量确认
    public static final String BATCH_CONFIRMATION = "batchConfirmation";public static void main(String[] args) throws IOException, InterruptedException {
        long begin = System.currentTimeMillis();    //记录开始时间
        //获取信道
        Channel channel = ChannelUtil.getChannel();
        //创建一个信道
        channel.queueDeclare(BATCH_CONFIRMATION, true, false, false, null);
        //开启发布确认功能
        channel.confirmSelect();
        //定义每次批量处理多少消息进行确认
        int batchNumber = 100;
        //循环发送消息
        for (int i = 0; i < 1000; i++) {
            String str = "批量发布确认信息" + i;
            System.out.println("开始发送信息：" + i);
            //发布信息
            channel.basicPublish("", BATCH_CONFIRMATION, MessageProperties.PERSISTENT_TEXT_PLAIN,
                    str.getBytes(StandardCharsets.UTF_8));
            //验证是否发送成功（等待确认）  用求余的方式来判断每轮100个
            //channel.waitForConfirms(3000); 发送三秒后没得到回复将断定未发送过去
            if ((i + 1) % batchNumber == 0) {
                if (channel.waitForConfirms()) {
                    System.out.println("批量发送成功了 范围为：" + (i - (batchNumber - 1)) + " ~ " + i);
                }
            }
        }
        long end = System.currentTimeMillis();  //记录结尾时间
        System.out.println("批量发布确认用时：" + (end - begin)); //批量发布确认用时：454
    }
}

public class ProducerA {
    //异步发布确认
    public static final String ASYNC_RELEASE_CONFIRMATION = "asyncReleaseConfirmation";

    public static void main(String[] args) throws IOException {
        long begin = System.currentTimeMillis();    //记录开始时间
        //获取信道
        Channel channel = ChannelUtil.getChannel();
        //创建一个信道
        channel.queueDeclare(ASYNC_RELEASE_CONFIRMATION, true, false, false, null);
        //开启发布确认功能
        channel.confirmSelect();
        //~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Start
        //线程安全有序的一个哈希表Map，适用于高并发的情况
        //1.轻松的将序号与消息进行关联 2.轻松批量删除条目 只要给到序列号 3.支持并发访问
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
        
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            //这个是回调成功的，回调成功后把集合中的数据删除，最终就代表失败的多少
            if (multiple) {
                ConcurrentNavigableMap<Long, String> longStringConcurrentNavigableMap =
                        outstandingConfirms.headMap(deliveryTag, true);
                longStringConcurrentNavigableMap.clear();
            } else {
                outstandingConfirms.remove(deliveryTag);
            }
            System.out.println("~~~~ 回调成功的数据：" + deliveryTag + "   是否批量确认：" + multiple);
        };
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            System.out.println("~~~~ 回调失败的数据：" + deliveryTag);
        };
        //添加监听器，监听返回(监听器一定要再发送消息之前就创建和监听) 参数1：回调成功 参数2：回调失败
        channel.addConfirmListener(ackCallback, nackCallback);
        //~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ End
        //循环发送消息 （因为是异步 这里正常发送不用管它）
        for (int i = 0; i < 10000; i++) {
            String str = "异步发布确认信息" + i;
            //记录要发送的数据添加到集合中
            outstandingConfirms.put(channel.getNextPublishSeqNo(),str);
            //发布信息
            channel.basicPublish("", ASYNC_RELEASE_CONFIRMATION, MessageProperties.PERSISTENT_TEXT_PLAIN,
                    str.getBytes(StandardCharsets.UTF_8));
        }
        long end = System.currentTimeMillis();  //记录结尾时间
        System.out.println("异步发布确认用时：" + (end - begin)); //异步发布确认用时：337
    }
}

我们没使用Exchange，但仍能够将消息发送到队列。之前能实现的原因是因为我们使用的是默认交换，我们通过空字符串("")进行标识
　　basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
　　当初使用：channel.basicPublish("", "当初写队列名称", null,"发送的消息");
第一个参数是交换机的名称：空字符串表示默认或无名称交换机
　　消息能路由发送到队列中其实是由routingKey(binding key)绑定key指定的，那时key都填写队列名称，所有直接被绑定到对应队列，
　　可以说使用的是直接交换机（direct）

创建交换机方法：exchangeDeclare(String exchange,BuiltinExchangeType type,boolean durable,
                    boolean autoDelete,boolean internal,Map<String, Object> arguments) 
exchange: 交换机名称
type: 交换机类型，direct、topic、 fanout、 headers
durable: 是否需要持久化
autoDelete: 当最后一个绑定到Exchange上的队列删除后，自动删除该Exchange
internal: 当前Exchange是否用于RabbitMQ内部使用，默认为False
arguments: 扩展参数，用于扩展AMQP协议定制化使用
注：推荐在编写生产者时创建交换机，在编写消费者时应该创建队列，并且队列绑定交换机，启动时先启动交换机

上图是一个队列绑定关系图，我们来看看他们之间数据接收情况是怎么样的：
     队列绑定交换机的Key               匹配规则
    quick.orange.rabbit         被队列 Q1 Q2 接收到
    lazy.orange.elephant        被队列 Q1 Q2 接收到
    quick.orange.fox            被队列 Q1 接收到
    lazy.brown.fox              被队列 Q2 接收到
    lazy.pink.rabbit            虽然满足两个绑定规则但两个规则都是在Q2队列，所有只有Q2接收一次
    quick.brown.fox             不匹配任何绑定不会被任何队列接收到会被丢弃
    quick.orange.male.rabbit    是四个单词不匹配任何绑定会被丢弃
    lazy.orange.male.rabbit     是四个单词但匹配 Q2

①：消息被拒绝（basic.reject/basic.nack)，且 requeue = false(代表不重新回到队列)
②：消息因TTL过期（就是任务消息上携带过期时间）
③：消息队列的消息数量已经超过最大队列长度，先入队的消息会被丢弃变为死信

public class TTLConsumer {
    //声明普通的交换机名称
    public static final String NORMAL_EXCHANGE = "NormalExchange";
    //声明死信交换机名称
    public static final String DLX_EXCHANGE = "DLXExchange";
    //声明普通队列名称
    public static final String Normal_Queue = "NormalQueue";
    //声明死信队列名称
    public static final String DLX_QUEUE = "DLXQueue";
    //声明路由绑定关系 Routing Key 普通交换机到普通队列
    public static final String NORMAL_KEY = "NormalKey";
    //声明路由绑定关系 Routing Key 死信交换机到死信队列
    public static final String DLX_KEY = "DLXKey";

    public static void main(String[] args) throws IOException {
        //调用自己的工具类获取信道
        Channel channel = ChannelUtil.getChannel();
        //声明普通exchange交换机 并设置为直接交换机；防止消费者先启动报错，找不到交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT,
                true, false, false, null);
        //声明死信exchange交换机 并设置为直接交换机；防止消费者先启动报错，找不到交换机
        channel.exchangeDeclare(DLX_EXCHANGE, BuiltinExchangeType.DIRECT,
                true, false, false, null);

        //声明死信队列
        channel.queueDeclare(DLX_QUEUE, true, false, false, null);
        //死信队列绑定死信交换机routingKey
        channel.queueBind(DLX_QUEUE, DLX_EXCHANGE, DLX_KEY);

        //参数设置
        Map<String, Object> arguments = new HashMap<>();
        //正常队列设置死信交换机 参数key是固定值
        arguments.put("x-dead-letter-exchange", DLX_EXCHANGE);
        //正常队列设置死信交换机到死信队列绑定Routing Key 参数key是固定值
        arguments.put("x-dead-letter-routing-key", DLX_KEY);
        //声明普通队列
        channel.queueDeclare(Normal_Queue, true, false, false, arguments);
        //普通队列绑定普通交换机routingKey
        channel.queueBind(Normal_Queue, NORMAL_EXCHANGE, NORMAL_KEY);
        System.out.println("初始化完成，等待接收消息");
        //接收队列消息
        channel.basicConsume(Normal_Queue, true, (consumerTag, message) -> {
            System.out.println("如同队列获取的任务并处理：" +
                    new String(message.getBody(), StandardCharsets.UTF_8));
        }, consumerTag -> {
            System.out.println("监听的队列出现异常；可能队列被删除！");
        });
    }
}

public class DLXConsumer {
    //声明死信交换机名称
    public static final String DLX_EXCHANGE = "DLXExchange";
    //声明死信队列名称
    public static final String DLX_QUEUE = "DLXQueue";
    //声明路由绑定关系 Routing Key 死信交换机到死信队列
    public static final String DLX_KEY = "DLXKey";

    public static void main(String[] args) throws IOException {
        //调用自己的工具类获取信道
        Channel channel = ChannelUtil.getChannel();
        //声明死信exchange交换机 并设置为直接交换机；防止消费者先启动报错，找不到交换机
        channel.exchangeDeclare(DLX_EXCHANGE, BuiltinExchangeType.DIRECT,
                true, false, false, null);
        //声明死信队列
        channel.queueDeclare(DLX_QUEUE, true, false, false, null);
        //死信队列绑定死信交换机routingKey
        channel.queueBind(DLX_QUEUE, DLX_EXCHANGE, DLX_KEY);
        System.out.println("初始化完成，等待接收消息");
        //接收队列消息
        channel.basicConsume(DLX_QUEUE, true, (consumerTag, message) -> {
            System.out.println("死信队列里获取的任务并处理：" +
                    new String(message.getBody(), StandardCharsets.UTF_8));
        }, consumerTag -> {
            System.out.println("监听的队列出现异常；可能队列被删除！");
        });
    }
}

public class DLXProducer {
    //声明普通的交换机名称
    public static final String NORMAL_EXCHANGE = "NormalExchange";
    //声明路由绑定关系 Routing Key 普通交换机到普通队列
    public static final String NORMAL_KEY = "NormalKey";

    public static void main(String[] args) throws IOException {
        //调用自己的工具类获取信道
        Channel channel = ChannelUtil.getChannel();
        //声明普通exchange交换机 并设置为直接交换机；防止消费者先启动报错，找不到交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT,
                true, false, false, null);
        //发送消息的参数设置 expiration过期时间10秒   deliveryMode 消息持久化方式
        AMQP.BasicProperties properties = new AMQP.BasicProperties()
                .builder().expiration("10000").deliveryMode(2).build();
        //循环发送消息
        for (int i = 0; i < 5; i++) {
            String str = "测试队列任务 " + i;
            //发布信息
            channel.basicPublish(NORMAL_EXCHANGE, NORMAL_KEY, properties,
                    str.getBytes(StandardCharsets.UTF_8));
        }
        System.out.println("消息发送完毕！");
    }
}

 　　　　//参数设置
        Map<String, Object> arguments = new HashMap<>();
        //正常队列设置死信交换机 参数key是固定值
        arguments.put("x-dead-letter-exchange", DLX_EXCHANGE);
        //正常队列设置死信交换机到死信队列绑定Routing Key 参数key是固定值
        arguments.put("x-dead-letter-routing-key", DLX_KEY);
        //设置正常队列的长度限制 为3 
        arguments.put("x-max-length",3);

//接收队列消息
channel.basicConsume(Normal_Queue, false, (consumerTag, message) -> {
    //获取的任务消息
    String msg = new String(message.getBody(), StandardCharsets.UTF_8);
    //手动不确认，拒收，并丢去队列
    if ("测试队列任务 3".equals(msg)) {
        //出现异常手动进行不应答;并且不放入队列中
        channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
    } else {
        System.out.println("如同队列获取的任务并处理：" + msg);
    }
}, consumerTag -> {
    System.out.println("监听的队列出现异常；可能队列被删除！");
});

那么什么时候需要用延时队列呢？考虑一下以下场景：
    ①：订单在十分钟之内未支付则自动取消。
    ②：新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。
    ③：账单在一周内未支付，则自动结算。
    ④：用户注册成功后，如果三天内没有登陆则进行短信提醒。
    ⑤：用户发起退款，如果三天内没有得到处理则通知相关运营人员。
    ⑥：预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议。

设置这个TTL值有两种方式（队列设置、消息设置）：
第一种是在创建队列的时候设置队列的 "x-message-ttl" 属性，如下：
    Map<String, Object> arguments = new HashMap<>();
    //设置消息延迟10秒；投递到该队列的消息超过10秒直接丢弃
    arguments.put("x-message-ttl",10000);
    //创建队列，并指定参数
    channel.queueDeclare(Normal_Queue, true, false, false, arguments);
第二种方式针对每条消息设置TTL：
    AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
    //这条消息的过期时间也被设置成了10s , 超过10秒未处理则执行到此消息后被丢弃
    builder.expiration("10000");
    AMQP.BasicProperties properties = builder.build();
    channel.basicPublish(exchangeName, routingKey, mandatory, properties, "msg body".getBytes());

区别的：第一种在队列上设置TTL属性，那么一旦消息过期，就会被队列丢弃；而第二种方式，消息即使过期，也不一定会被马上丢弃，
因为消息是否过期是在即将投递到消费者之前判定的，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间。
另外，还需要注意的一点是，如果不设置TTL，表示消息永远不会过期，如果将TTL设置为0，则表示除非此时可以直接投递该消息到消费者，
否则该消息将会被丢弃。