RabbitMQ 超详细入门篇

[root@iZj6ciuzx7luldnazt4iswZ ~]# cd /
[root@iZj6ciuzx7luldnazt4iswZ /]# ls
bin   dev  home        lib    lost+found  mnt  patch  root  sbin  sys  usr  www
boot  etc  install.sh  lib64  media       opt  proc   run   srv   tmp  var
[root@iZj6ciuzx7luldnazt4iswZ /]# cd usr
[root@iZj6ciuzx7luldnazt4iswZ usr]# ls
bin  etc  games  include  lib  lib64  libexec  local  sbin  share  src  tmp
[root@iZj6ciuzx7luldnazt4iswZ usr]# mkdir wsm
[root@iZj6ciuzx7luldnazt4iswZ usr]# ls
bin  etc  games  include  lib  lib64  libexec  local  sbin  share  src  tmp  wsm
[root@iZj6ciuzx7luldnazt4iswZ usr]# cd wsm
[root@iZj6ciuzx7luldnazt4iswZ wsm]# ls
erlang-21.3-1.el7.x86_64.rpm  rabbitmq-server-3.8.8-1.el7.noarch.rpm					#上传的两个文件

# 解压安装 erlang
rpm -ivh erlang-21.3.8.21-1.el7.x86_64.rpm
# 云下载一个 初始化一些配置, 过程比较慢请耐心等待~, 在这之后才可以进行 安装 RabbitMQ
yum install socat -y
# 解压安装 rabbitmq
rpm -ivh rabbitmq-server-3.8.8-1.el7.noarch.rpm

# 启动服务
systemctl start rabbitmq-server
# 查看服务状态
systemctl status rabbitmq-server
# 开机自启动
systemctl enable rabbitmq-server
# 停止服务
systemctl stop rabbitmq-server
# 重启服务
systemctl restart rabbitmq-server

# 安装web控制面板
rabbitmq-plugins enable rabbitmq_management

# 安装完毕以后，重启服务即可
systemctl restart rabbitmq-server

# 访问 http://服务器ip:15672 ，用默认账号密码(guest)登录，出现权限问题
# 默认情况只能在 localhost 本机下访问，所以需要添加一个远程登录的用户
# 创建账号和密码: admin 123456
rabbitmqctl add_user admin 123456
# 设置用户角色,用户级别: administrator monitoring policymaker managment
rabbitmqctl set_user_tags admin administrator
# 为用户添加资源权限
# set_permissions [-p <vhostpath>] <user> <conf> <write> <read>  # 添加配置、写、读权限
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"

##### 扩展一些命令：#####
关闭应用的命令为：  rabbitmqctl stop_app
清除的命令为：		 rabbitmqctl reset
重新启动命令为：	rabbitmqctl start_app

<dependencies>
    <!-- rabbitMQ 依赖 -->
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.8.0</version>
    </dependency>
</dependencies>

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

/** 消息生产者 **/
public class Producer {
    // 定义队列名称
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws IOException, TimeoutException {
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("47.243.109.199");
        factory.setUsername("admin");
        factory.setPassword("123456");
        //channel 实现了自动 close 接口 自动关闭 不需要显示关闭
        //创建连接对象
        Connection connection = factory.newConnection();
        //根据连接对象,获取信道
        Channel channel = connection.createChannel();

        /**设置消息队列的属性！
         *  queue       :队列名称
         *  durable     :是否持久化 如果持久化,mq重启后队列数据还在! (队列是在虚拟路径上的...)
         *  exclusive   :队列是否独占此连接,队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
         *  autoDelete  :队列不再使用时是否自动删除此队列,如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
         *  arguments   :队列参数 null,可以设置一个队列的扩展参数,需要时候使用！比如：可设置存活时间
         * */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        /**发送消息,参数:
         * exchange     :指定的交换机,不指定就会有默认的....
         * routingKey   :路由key,交换机根据路由key来将消息转发到指定的队列,如果使用默认交换机routingKey设置为队列的名称
         * props        :消息包含的属性: 后面介绍,可以是一个一个对象... 消息持久化配置...
         * body         :发送的消息,AMQP以字节方式传输...
         * */
        channel.basicPublish("", QUEUE_NAME, null, "Hello Word你好世界".getBytes());
        System.out.println("消息发送完毕");
    }
}

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

/** 消息消费者 **/
public class Consumer {
    // 定义队列名称
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("47.243.109.199");
        factory.setUsername("admin");
        factory.setPassword("123456");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        System.out.println("等待接收消息.........");
        //收到消息后用来处理消息的回调对象
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody());
            System.out.println(message);
        };

        //取消消费的一个回调接口 如在消费的时候队列被删除掉了
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("消息消费被中断");
        };
       	/**
         * 消费者消费消息 - 接受消息
         * queue            消费哪个队列
         * autoAck          消费成功之后是否要自动应答 true 代表自动应答 false 手动应答,要通过编程实现回复验证,这就是Unacked 为返回ack的数据
         * deliverCallback  消费方法，当消费者接收到消息要执行的方法, 参数是一个函数式接口可以使用 lambda表达式~
         * cancelCallback   消息被取消时的回调
         */
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
    }
}

import com.rabbitmq.client.Channel; //导入MQ的包~
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/** RabbitMQ 连接配置类: **/
public class MQChannelUtil {
    //得到一个连接的 channel
    public static Channel getChannel() throws Exception {
        //创建一个连接工厂, 设置连接: IP 端口 用户 密码
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("42.192.149.71");
//      factory.setPort("设置对应的端口,默认就是: 5672");
        factory.setUsername("admin");
        factory.setPassword("123456");
        //创建连接对象 信道对象
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        return channel;
    }
}

import com.rabbitmq.client.Channel;
import com.wsm.Util.MQChannelUtil;
import java.util.Scanner;

/** 消息生产者 **/
public class Producer {
    // 定义队列名称
    private final static String QUEUE_NAME = "Word";

    public static void main(String[] args) throws Exception {
        // 工具类创建一个信道
        Channel channel = MQChannelUtil.getChannel();
        // Java控制台测试法消息:
        Scanner scanner = new Scanner(System.in);
        //创建交换机
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        // 循环多次发布消息:
        while (scanner.hasNext()){
            String message = scanner.next();
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("消息发送完毕");
        }
    }
}

import com.rabbitmq.client.*;
import com.wsm.Util.MQChannelUtil;

/** 消息消费者 **/
public class Consumer1 {
    // 定义队列名称
    private final static String QUEUE_NAME = "Word";

    public static void main(String[] args) throws Exception {
        // 工具类创建一个信道
        Channel channel = MQChannelUtil.getChannel();

        //收到消息后用来处理消息的回调对象
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody());
            System.out.println(message);
        };

        //取消消费的一个回调接口 如在消费的时候队列被删除掉了
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("消息消费被中断");
        };

        /** 消费者消费消息 - 接受消息: 注意参数两个回调函数~ */
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
    }
}

import com.rabbitmq.client.Channel;
import com.wsm.Util.MQChannelUtil;
import java.util.Scanner;

/** 消息生产者 **/
public class Producer {
    // 定义队列名称
    private final static String QUEUE_NAME = "ack_test";

    public static void main(String[] args) throws Exception {
        // 工具类创建一个信道
        Channel channel = MQChannelUtil.getChannel();
        // Java控制台测试法消息:
        Scanner scanner = new Scanner(System.in);
        // 创建队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 循环多次发布消息:
        while (scanner.hasNext()){
            String message = scanner.next();
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("消息发送完毕");
        }
    }
}

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.wsm.Util.MQChannelUtil;

/** 消息消费者 **/
public class Consumer2 {
    // 定义队列名称
    private final static String QUEUE_NAME = "ack_test";

    public static void main(String[] args) throws Exception {
        // 工具类创建一个信道
        Channel channel = MQChannelUtil.getChannel();
        //收到消息后用来处理消息的回调对象
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            // 为了方便查看处理效果,我们将消费者 线程休眠一段时间 模拟处理数据;
            try {
                Thread.sleep(10000); // 毫秒 *1000;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            String message = new String(delivery.getBody());
            System.out.println(message);
            /** 接收处理完毕消息之后给MQ 回复ack
             *  参数1 消息的唯一标识Tag
             *  参数2 是否支持批量回复,一般建议false 保证数据安全！
             * **/
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
        };
        //取消消费的一个回调接口 如在消费的时候队列被删除掉了
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("消息消费被中断");
        };
        /** 消费者消费消息 - 接受消息: 注意参数两个回调函数~ */
        /** 参数二 设置 false 手动应答 **/
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, cancelCallback);
    }
}

// 让队列持久化
boolean durable = true;	// false 不持久化   true 持久化 
// 声明队列
channel.queueDeclare("队列名", durable, false, false, null);	// 设置true 之后,每次MQ重启的时候,该队列都会自动重新在虚拟路径上自动加载... 

// 发送者发送消息的时候，带上 MessageProperties.PERSISTENT_TEXT_PLAIN 告诉,MQ消息要进行持久化;
channel.basicPublish("交换机", "队列名", MessageProperties.PERSISTENT_TEXT_PLAIN, "要发送的消息,字节传输");

/** 单个发布确认 **/
public static void singleConfirm() throws Exception {
    Channel channel = MQChannelUtil.getChannel();
    // 队列声明
    channel.queueDeclare("singleConfirm", false, false, false, null);
    // 开启发布确认
    channel.confirmSelect();

    // 为了方便计算各个 发布确认策略 耗时: 开始-结束放一个系统时间获取毫秒数;
    long begin = System.currentTimeMillis();
    
    // 发送1000 个消息....
    for (int i = 0; i < MESSAGE_COUNT; i++) {
        String message = i + "";
        channel.basicPublish("", "singleConfirm", null, message.getBytes());

        // 服务端返回 false 或超时时间内未返回，生产者可以消息重发
        boolean flag = channel.waitForConfirms();
        if (flag) {
            System.out.println("消息发送成功");
        }
        // else{ /** 消息重发处理... **/  }
    }
    
    // 1000个消息结束...
    long end = System.currentTimeMillis();
    System.out.println("发布" + MESSAGE_COUNT + "个单独确认消息,耗时" + (end - begin) + "ms");
}

/** 批量发布确认 **/
public static void batchConfirm() throws Exception {
    Channel channel = MQChannelUtil.getChannel();
    // 队列声明
    channel.queueDeclare("batchConfirm", false, false, false, null);
    // 开启发布确认
    channel.confirmSelect();

    // 为了方便计算各个 发布确认策略 耗时: 开始-结束放一个系统时间获取毫秒数;
    long begin = System.currentTimeMillis();

    // 批量确认消息大小,当发送的消息数到 100 执行 waitForConfirms(); 询问MQ 当前所有的消息有没有抵达~
    int batchSize = 100;
    // 未确认消息个数, 每次发送消息 ++ 用于判断是否改批量验证消息发送;
    int outstandingMessageCount = 0;
    for (int i = 0; i < MESSAGE_COUNT; i++) {
        // 发送的消息,并发送~
        String message = i + "";
        channel.basicPublish("", "batchConfirm", null, message.getBytes());
        // 每次发送一个消息进行计算当前是第几个,为批量验证的消息;
        outstandingMessageCount++;
        // 判断当前的 100 个消息有没有都发送的MQ上！
        if (outstandingMessageCount == batchSize) {
            channel.waitForConfirms();
            outstandingMessageCount = 0;
        }
    }

    //为了确保还有剩余没有确认消息 再次确认
    if (outstandingMessageCount > 0) {
        channel.waitForConfirms();
    }

    long end = System.currentTimeMillis();
    System.out.println("发布" + MESSAGE_COUNT + "个批量确认消息,耗时" + (end - begin) + "ms");
}

/** 异步发布确认 **/
public static void syncConfirm() throws Exception{
    Channel channel = MQChannelUtil.getChannel();
    // 队列声明
    channel.queueDeclare("syncConfirm", false, false, false, null);
    /** 开启发布确认 **/
    channel.confirmSelect();
    // 为了方便计算各个 发布确认策略 耗时: 开始-结束放一个系统时间获取毫秒数;
    long begin = System.currentTimeMillis();

    /** 步骤一: 创建一个线程安全的一个哈希表,用于记录每一个消息发送,这样MQ异步返回时候可以知道具体是那一个消息发送成功|失败 **/
    /**
         * 线程安全有序的一个哈希表，适用于高并发的情况
         * 1.轻松的将序号与消息进行关联: k,v 存储结构, k消息标识 v发送的消息体,每次发送消息前先存在集合中;
         * 2.轻松批量删除条目只要给到序列号: 对于发现成功的消息,直接从集合中移除...
         * 3.支持并发访问,Concurrent接口 是线程安全的; */
    ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();

    /** 步骤三: 编写回调监听器,因为: 消息发送出错要立刻进行监听所以,所以创建在发送消息之前; **/
    /** ack 确认收到消息的一个回调 1.消息序列号 2.true 批量确认接受小于等于当前序列号的数据 false 确认当前序列号消息 */
    ConfirmCallback ackCallback = (sequenceNumber, multiple) -> {
        if (multiple) {
            System.out.println("消息成功接收:"+sequenceNumber);
            // ConcurrentNavigableMap方法()返回的是小于|等于 K 的集合, true:小于等于 false:返回小于该序列号的数据集合;
            ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(sequenceNumber, true);
            // 清除该部分确认消息 confirmed 里保存的都是,MQ 已经接收的消息;
            // 遍历 confirmed K, 根据 K 删除 outstandingConfirms 的值...
            // outstandingConfirms 里面保存的都是,MQ 还未确认的消息...
        }else{
            //只清除当前序列号的消息
            outstandingConfirms.remove(sequenceNumber);
        }
    };
    // nack 消息失败执行{} 可以写,消息失败需要执行的代码...
    ConfirmCallback nackCallback = (sequenceNumber, multiple) -> {
        // 这里就输出一下为被确认的消息...
        String message = outstandingConfirms.get(sequenceNumber);
        System.out.println("发布的消息"+message+"未被确认，序列号"+sequenceNumber);
    };
    // 发生者 等待MQ回调消息确认的 监听器, 本次程序值监听 ack成功的消息;
    channel.addConfirmListener(ackCallback, null);

    /** 步骤二: 发送者一直往MQ发送消息 **/
    for (int i = 0; i < MESSAGE_COUNT; i++) {
        String message = "消息" + i;
        // channel.getNextPublishSeqNo() 获取下一个消息的序列号,通过序列号与消息体进行一个关联,全部都是未确认的消息体
        outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
        // 发送消息;
        channel.basicPublish("", "syncConfirm", null, message.getBytes());
    }

    long end = System.currentTimeMillis();
    System.out.println("发布" + MESSAGE_COUNT + "个批量确认消息,耗时" + (end - begin) + "ms");
}

/** 消息生产者 **/
public class Producer {
    // 交换机名
    public static final String EXCHANGE_NAME = "wsm";

    public static void main(String[] args) throws Exception {
        // 创建连接对象,声明交换机 发送消息
        Channel channel = MQChannelUtil.getChannel();
        /**
         * 声明一个 exchange
         * 1.exchange 的名称
         * 2.exchange 的类型, 可以是String直接写,也可以是 枚举类型;
         */
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);

        Scanner sc = new Scanner(System.in);
        System.out.println("请输入信息");
        while (sc.hasNext()) {
            String message = sc.nextLine();
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
            System.out.println("生产者发出消息" + message);
        }
    }
}

/** 消息消费者 **/
public class Consumer1 {
    // 定义交换机名称
    public static final String EXCHANGE_NAME = "wsm";

    public static void main(String[] args) throws Exception {
        Channel channel = MQChannelUtil.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

        /** 生成一个临时的队列 队列的名称是随机的 当消费者断开和该队列的连接时 队列自动删除  */
        String queueName = channel.queueDeclare().getQueue();
        // 绑定: 把该临时队列绑定我们的 exchange 其中 routingkey(也称之为 binding key)为空字符串: Fanout模式 routingkey 没作用!
        channel.queueBind(queueName, EXCHANGE_NAME, "");

        // 发送回调
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("wsm 发布的最新消息:"+message);
        };
        // 消费者监听消息
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
    }
}

/** 消息生产者 **/
public class Producer {
    // 交换机名
    public static final String EXCHANGE_NAME = "DIRECT";

    public static void main(String[] args) throws Exception {
        // 创建连接对象,声明交换机 发送消息
        Channel channel = MQChannelUtil.getChannel();
        /**
         * 声明一个 exchange
         * 1.exchange 的名称
         * 2.exchange 的类型, 可以是String直接写,也可以是 枚举类型; */
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        /** 发送消息 **/
        channel.basicPublish(EXCHANGE_NAME, "Conkey1", null, "Conkey1 发送的消息".getBytes("UTF-8"));
        channel.basicPublish(EXCHANGE_NAME, "Conkey2", null, "Conkey2 发送的消息".getBytes("UTF-8"));
    }
}

/** 消息消费者 **/
public class Consumer1 {
    // 定义交换机名称
    public static final String EXCHANGE_NAME = "DIRECT";

    public static void main(String[] args) throws Exception {
        Channel channel = MQChannelUtil.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

        /** 生成一个临时的队列 队列的名称是随机的 当消费者断开和该队列的连接时 队列自动删除  */
        String queueName = channel.queueDeclare().getQueue();
        // 绑定: 把该临时队列绑定我们的 exchange
        // 参数二 设置该队列和交换和绑定的 routingkey
        channel.queueBind(queueName, EXCHANGE_NAME, "Conkey1");

        // 发送回调
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("最新的消息是:"+message);
        };
        // 消费者监听消息
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
    }
}

// 可以设置多个key
channel.queueBind(queueName, EXCHANGE_NAME, "Conkey2");

//交换机类型
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
/** 发送消息 **/
channel.basicPublish(EXCHANGE_NAME, "Conkey.one", null, "Conkey.one 发送的消息".getBytes("UTF-8"));
channel.basicPublish(EXCHANGE_NAME, "Conkey.two.123", null, "Conkey.two.123 发送的消息".getBytes("UTF-8"));

// 参数二 设置该队列和交换和绑定的routingkey , Topic模式可以用过 通配符进行动态匹配: * 表示一个任意的单词;
channel.queueBind(queueName, EXCHANGE_NAME, "Conkey.*");

// 参数二 设置该队列和交换和绑定的routingkey , Topic模式可以用过 通配符进行动态匹配: # 表示一个|多个任意的单词;
channel.queueBind(queueName, EXCHANGE_NAME, "Conkey.#");

/** 消息生产者 **/
public class Producer {
    // 交换机名
    public static final String EXCHANGE_NAME = "HEADERS";

    public static void main(String[] args) throws Exception {
        // 创建连接对象,声明交换机 发送消息
        Channel channel = MQChannelUtil.getChannel();
        /**
         * 声明一个 exchange
         * 1.exchange 的名称
         * 2.exchange 的类型, 可以是String直接写,也可以是 枚举类型; */
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.HEADERS);

        //发送的消息的消息头head map类型, 需要与队列的map 规范匹配才可以发送成功！ 不然发送失败(交换机不知道往那个队列上发送;
        HashMap<String ,Object> param = new HashMap<String, Object>();
        param.put("id","1");
        param.put("name","wsm");
        //设置Map 匹配参数！
        AMQP.BasicProperties.Builder builder=new AMQP.BasicProperties.Builder();
        builder.headers(param);

        /** 发送消息 **/
        channel.basicPublish(EXCHANGE_NAME, "", builder.build(), "header的内容lalala~~".getBytes("UTF-8"));
    }
}

/** 消息消费者 **/
public class Consumer1 {
    // 定义交换机名称
    public static final String EXCHANGE_NAME = "HEADERS";

    public static void main(String[] args) throws Exception {
        Channel channel = MQChannelUtil.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.HEADERS);
        /** 生成一个临时的队列 队列的名称是随机的 当消费者断开和该队列的连接时 队列自动删除  */
        String queueName = channel.queueDeclare().getQueue();

        //设置队列上的 map 参数,用于匹配请求时候的参数！
        //特殊参数 x-match 值 all 或 any
        //all  在发布消息时携带的map 必须和绑定在队列上的所有map 完全匹配
        //any  只要在发布消息时携带的有一对键值map 满足队列定义的多个参数map的其中一个就能匹配上
        //注意: 这里是键值对的完全匹配，只匹配到键了，值却不一样是不行的；
        HashMap<String ,Object> param = new HashMap<String, Object>();
        param.put("x-match","all");
        param.put("id","1");            // 可以尝试改变,map 信息,生产者消息还能发送到队列上面~
        param.put("name","wsm");

        // 绑定: 把该临时队列绑定我们的 exchange
        // 队列绑定时需要指定参数,注意虽然不需要路由键但仍旧不能写成null，需要写成空字符串"", 参数四: map参数,规范！
        channel.queueBind(queueName, EXCHANGE_NAME, "",param);

        // 发送回调
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("最新的消息是:"+message);
            System.out.println("Map传入参数数据:"+delivery.getProperties().getHeaders());
        };
        // 消费者监听消息
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
    }
}

//消费者：队列规则all 所有匹配即可
        Map<String ,Object> param = new HashMap<String, Object>();
        param.put("x-match","all");
        param.put("id","1");
        param.put("name","wsm");
		
		//生产者：传入head map;  
        Map<String ,Object> param = new HashMap<String, Object>();
        param.put("id","2");
        param.put("name","wsm");
       	//不匹配
		
		//生产者 
		Map<String ,Object> param = new HashMap<String, Object>();
        param.put("x-match","all");
        param.put("id","1");
        param.put("name","wsm");
		//匹配
		
//消费者：队列规则any 一个匹配即可
        Map<String ,Object> param = new HashMap<String, Object>();
        param.put("x-match","any");
        param.put("id","1");
        param.put("name","wsm");     
        //生产者：传入head map;  
		Map<String ,Object> param = new HashMap<String, Object>();
        param.put("id","1");		//匹配
        
        //生产者：传入head map;  
       	Map<String ,Object> param = new HashMap<String, Object>();
        param.put("id","2"); 		//不匹配 key /value 都要匹配才可以
        ....

/** 消息生产者 **/
public class Producer {
    // 普通交换机名
    public static final String EXCHANGE_NAME = "normal_exchange";

    public static void main(String[] args) throws Exception {
        Channel channel = MQChannelUtil.getChannel();

        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        //设置消息的 TTL 时间 10s秒
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
        //该信息是用作演示队列个数限制
        for (int i = 1; i < 11; i++) {
            String message = "info" + i;
            channel.basicPublish(EXCHANGE_NAME, "zhangsan", properties, message.getBytes());
            System.out.println("生产者发送消息:" + message);
        }
    }
}

/** 消息消费者 **/
public class Consumer1 {
    // 普通交换机名称
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    // 死信交换机名称
    private static final String DEAD_EXCHANGE = "dead_exchange";

    public static void main(String[] args) throws Exception {
        Channel channel = MQChannelUtil.getChannel();
        // 声明死信和普通交换机 类型为 direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        // 声明死信队列
        String deadQueue = "dead-queue";
        channel.queueDeclare(deadQueue, false, false, false, null);
        //死信队列绑定：队列、交换机、路由键（routingKey）
        channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");

        // 正常队列绑定死信队列信息
        HashMap<String, Object> params = new HashMap<>();
        // 正常队列设置死信交换机 参数 key 是固定值
        params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        // 正常队列设置死信 routing-key 参数 key 是固定值
        params.put("x-dead-letter-routing-key", "lisi");

        // 正常队列
        String normalQueue = "normal-queue";
        channel.queueDeclare(normalQueue, false, false, false, params);
        channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");

        System.out.println("等待接收消息........... ");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Consumer01 接收到消息" + message);
        };
        channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> {
        });
    }
}

/** 消息消费者 **/
public class Consumer2 {
    //死信交换机名称
    private static final String DEAD_EXCHANGE = "dead_exchange";

    public static void main(String[] args) throws Exception {
        Channel channel = MQChannelUtil.getChannel();

        //声明交换机
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        //声明队列
        String deadQueue = "dead-queue";
        channel.queueDeclare(deadQueue, false, false, false, null);
        channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");

        System.out.println("等待接收死信消息........... ");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Consumer02 接收到消息" + message);
        };
        channel.basicConsume(deadQueue, true, deliverCallback, consumerTag -> {
        });
    }
}

//设置正常队列的长度限制，例如发10个，4个则为死信 注意：此时需要把原先队列删除 因为参数改变了
params.put("x-max-length",6);

<dependencies>
    <!-- SpringBoot依赖 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <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>
    <!-- 阿里巴巴fastjson -->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.47</version>
    </dependency>
    <!-- lombok依赖 -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
    <!--RabbitMQ 测试依赖-->
    <dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-rabbit-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

# SpringBoot 配置RabbitMQ ip 端口 用户 密码;
spring.rabbitmq.host=47.243.109.199
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456

Map<String, Object> args = new HashMap<>(3);
//声明当前队列绑定的死信交换机
args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
//声明当前队列的死信路由 key
args.put("x-dead-letter-routing-key", "YD");
//声明队列的 TTL
args.put("x-message-ttl", 10000);
//队列绑定交换机
QueueBuilder.durable(QUEUE_A).withArguments(args).build();

// Spring方式
// 编辑参数
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
// 发送消息
channel.basicPublish("交换机", "队列", properties, "消息".getBytes());

// SpringBoot方式
// 通过 rabbitTemplate 发送消息...
rabbitTemplate.convertAndSend("交换机", "队列", "消息", correlationData -> {
    correlationData.getMessageProperties().setExpiration("10000");
    return correlationData;
});

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class TtlQueueConfig {
    // 普通交换机 普通队列
    public static final String X_EXCHANGE = "X";
    public static final String QUEUE_A = "QA";
    public static final String QUEUE_B = "QB";
    // 死信交换机
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    // 死信队列
    public static final String DEAD_LETTER_QUEUE = "QD";

    // 声明 xExchange
    @Bean("xExchange")
    public DirectExchange xExchange() {
        return new DirectExchange(X_EXCHANGE);
    }

    // 声明 死信队列交换机
    @Bean("yExchange")
    public DirectExchange yExchange() {
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    }

    //声明队列 A ttl 为 10s 并绑定到对应的死信交换机
    @Bean("queueA")
    public Queue queueA() {     // 导包: org.springframework.amqp.core
        Map<String, Object> args = new HashMap<>(3);
        //声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //声明当前队列的死信路由 key
        args.put("x-dead-letter-routing-key", "YD");
        //声明队列的 TTL
        args.put("x-message-ttl", 10000);
        return QueueBuilder.durable(QUEUE_A).withArguments(args).build();
    }

    // 声明队列 A 绑定 X 交换机
    @Bean
    public Binding queueaBindingX(@Qualifier("queueA") Queue queueA,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    }

    //声明队列 B ttl 为 40s 并绑定到对应的死信交换机
    @Bean("queueB")
    public Queue queueB() {
        Map<String, Object> args = new HashMap<>(3);
        //声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //声明当前队列的死信路由 key
        args.put("x-dead-letter-routing-key", "YD");
        //声明队列的 TTL
        args.put("x-message-ttl", 40000);
        return QueueBuilder.durable(QUEUE_B).withArguments(args).build();
    }

    //声明队列 B 绑定 X 交换机
    @Bean
    public Binding queuebBindingX(@Qualifier("queueB") Queue queue1B,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queue1B).to(xExchange).with("XB");
    }


    //声明死信队列 QD
    @Bean("queueD")
    public Queue queueD() {
        return new Queue(DEAD_LETTER_QUEUE);
    }

    //声明死信队列 QD 绑定关系
    @Bean
    public Binding deadLetterBindingQAD(@Qualifier("queueD") Queue queueD,
                                        @Qualifier("yExchange") DirectExchange yExchange) {
        return BindingBuilder.bind(queueD).to(yExchange).with("YD");
    }
}

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.Date;

@Slf4j
@RequestMapping("ttl")
@RestController
public class SendMsgController {
    // RabbitTemplate 对rabbitmq 的服务接口API 进行了封装;
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("sendMsg/{message}")
    public void sendMsg(@PathVariable String message) {
        log.info("当前时间：{},发送一条信息给两个 TTL 队列:{}", new Date(), message);
        rabbitTemplate.convertAndSend("X", "XA", "消息来自 ttl 为 10S 的队列: " + message);
        rabbitTemplate.convertAndSend("X", "XB", "消息来自 ttl 为 40S 的队列: " + message);
    }
}

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.Date;

/**
 * 消费者 - 死信队列
 * @author wsm
 */
@Slf4j
@Component
public class DeadLetterQueueConsumer {
    //@RabbitListener 负责监听具体那个队列...
    @RabbitListener(queues = "QD")
    public void receiveD(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.info("当前时间：{},收到死信队列信息{}", new Date().toString(), msg);
    }
}

2022-01-23 19:12:01.380  INFO 1508 --- [nio-8080-exec-3] c.example.controller.SendMsgController   : 当前时间：Sun Jan 23 19:12:01 CST 2022,发送一条信息给两个 TTL 队列:嘻嘻嘻
2022-01-23 19:12:11.684  INFO 1508 --- [ntContainer#0-1] c.e.consumer.DeadLetterQueueConsumer     : 当前时间：Sun Jan 23 19:12:11 CST 2022,收到死信队列信息消息来自 ttl 为 10S 的队列: 拉拉拉
2022-01-23 19:12:41.566  INFO 1508 --- [ntContainer#0-1] c.e.consumer.DeadLetterQueueConsumer     : 当前时间：Sun Jan 23 19:12:41 CST 2022,收到死信队列信息消息来自 ttl 为 40S 的队列: 拉拉拉

@Configuration
public class MsgTtlQueueConfig {
    // 死信交换机
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    // 普通的队列
    public static final String QUEUE_C = "QC";

    //声明队列 C 死信交换机
    @Bean("queueC")
    public Queue queueB() {
        HashMap<String, Object> args = new HashMap<>(3);
        //声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //声明当前队列的死信路由 key
        args.put("x-dead-letter-routing-key", "YD");
        //没有声明 TTL 属性
        return QueueBuilder.durable(QUEUE_C).withArguments(args).build();
    }

    //声明队列 B 绑定 X 交换机
    @Bean
    public Binding queuecBindingX(@Qualifier("queueC") Queue queueC,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueC).to(xExchange).with("XC");
    }
}

/**
* 延时队列优化
* @param message 消息
* @param ttlTime 延时的毫秒
*/
@GetMapping("sendExpirationMsg/{message}/{ttlTime}")
public void sendMsg(@PathVariable String message, @PathVariable String ttlTime) {
    rabbitTemplate.convertAndSend("X", "XC", message, correlationData -> {
        // 发送消息的时候,设置消息的延迟时间...
        correlationData.getMessageProperties().setExpiration(ttlTime);
        return correlationData;
    });
    log.info("当前时间：{},发送一条时长{}毫秒 TTL 信息给队列 C:{}", new Date(), ttlTime, message);
}

# 工具引入,插件安装包;
[root@iZj6ciuzx7luldnazt4iswZ wsm]# ls
erlang-21.3-1.el7.x86_64.rpm  rabbitmq_delayed_message_exchange-3.8.0.ez  rabbitmq-server-3.8.8-1.el7.noarch.rpm

# 将插件移动到 RabbitMQ的plugins 包下: /usr/lib 是linux 默认安装服务路径...
[root@iZj6ciuzx7luldnazt4iswZ wsm]# cp rabbitmq_delayed_message_exchange-3.8.0.ez /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins
# rabbitmq 开始安装启动插件补丁...
[root@iZj6ciuzx7luldnazt4iswZ wsm]# rabbitmq-plugins enable rabbitmq_delayed_message_exchange
Enabling plugins on node rabbit@iZj6ciuzx7luldnazt4iswZ:
rabbitmq_delayed_message_exchange
The following plugins have been configured:
  rabbitmq_delayed_message_exchange
  rabbitmq_management
  rabbitmq_management_agent
  rabbitmq_web_dispatch
Applying plugin configuration to rabbit@iZj6ciuzx7luldnazt4iswZ...
The following plugins have been enabled:
  rabbitmq_delayed_message_exchange
started 1 plugins.
# 重启服务:
[root@iZj6ciuzx7luldnazt4iswZ wsm]# systemctl restart rabbitmq-server

@Configuration
public class DelayedQueueConfig {
    // 队列
    public static final String DELAYED_QUEUE_NAME = "delayed.queue";
    // 交换机
    public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
    // key
    public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";

    // 声明队列
    @Bean
    public Queue delayedQueue() {
        return new Queue(DELAYED_QUEUE_NAME);
    }

    // 自定义交换机 我们在这里定义的是一个延迟交换机
    @Bean
    public CustomExchange delayedExchange() {
        HashMap<String, Object> args = new HashMap<>();
        //自定义交换机的类型
        args.put("x-delayed-type", "direct");
        return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false, args);
    }

    // 绑定 队列和交换机;
    @Bean
    public Binding bindingDelayedQueue(@Qualifier("delayedQueue") Queue queue,
                                       @Qualifier("delayedExchange") CustomExchange delayedExchange) {
        return BindingBuilder.bind(queue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
    }
}

// 交换机 和 routingkey
public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";
// 请求接口:
@GetMapping("sendDelayMsg/{message}/{delayTime}")
public void sendMsg(@PathVariable String message, @PathVariable Integer delayTime) {
    rabbitTemplate.convertAndSend(DELAYED_EXCHANGE_NAME, DELAYED_ROUTING_KEY, message,
                                  correlationData -> {
                                      correlationData.getMessageProperties().setDelay(delayTime);
                                      return correlationData;
                                  });
    log.info(" 当 前 时 间 ： {}, 发 送 一 条 延 迟 {} 毫秒的信息给队列 delayed.queue:{}", new Date(), delayTime, message);
}

/**
 * 消费者 - 基于插件的延时队列
 *
 * @author wsm
 */
@Slf4j
@ComponentScan
public class DelayQueueConsumer {

    public static final String DELAYED_QUEUE_NAME = "delayed.queue";

    @RabbitListener(queues = DELAYED_QUEUE_NAME)
    public void receiveDelayedQueue(Message message) {
        String msg = new String(message.getBody());
        log.info("当前时间：{},收到延时队列的消息：{}", new Date().toString(), msg);
    }
}

# 开启RabbitMQ 发布确认模式：
spring.rabbitmq.publisher-confirm-type=correlated
# NONE 值是禁用发布确认模式，是默认值
# CORRELATED 值是发布消息成功到交换器后会触发回调方法
# SIMPLE 值经测试有两种效果
	# 其一效果和 CORRELATED 值一样会触发回调方法
	# 其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果
	# 根据返回结果来判定下一步的逻辑，注意的点是 waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel 则接下来无法发送消息到 broker

/** SpringBoot 消息确认模式 **/
@Configuration
public class ConfirmConfig {
    public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";
    public static final String CONFIRM_QUEUE_NAME = "confirm.queue";

    //声明业务 Exchange
    @Bean("confirmExchange")
    public DirectExchange confirmExchange() {
        return new DirectExchange(CONFIRM_EXCHANGE_NAME);
    }

    // 声明确认队列
    @Bean("confirmQueue")
    public Queue confirmQueue() {
        return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
    }

    // 声明确认队列绑定关系
    @Bean
    public Binding queueBinding(@Qualifier("confirmQueue") Queue queue,
                                @Qualifier("confirmExchange") DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("key1");
    }
}

@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback {
    /**
     * 交换机不管是否收到消息的一个回调方法
     *
     * @param correlationData 消息相关数据
     * @param ack             交换机是否收到消息, true(ack) false(nack)
     * @param cause           为收到消息的原因: 异常信息
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (ack) {
            log.info("交换机已经收到 id 为:{}的消息", id);
        } else {
            log.info("交换机还未收到 id 为:{}消息，原因:{}", id, cause);
        }
    }
}

@RestController
@RequestMapping("/confirm")
@Slf4j
public class ProducerController {
    @Autowired  // rabbitmq 模板对象;
    private RabbitTemplate rabbitTemplate;
    @Autowired  // 发布确认消息,消息回调方法类;
    private MyCallBack myCallBack;
    //依赖注入 rabbitTemplate 之后再设置它的回调对象
    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(myCallBack);
    }

    public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";

    
    /**
     * 消息回调和退回
     * @param message
     */
    @GetMapping("sendMessage/{message}")
    public void sendMessage(@PathVariable String message) {
        //指定消息 id 为 1
        CorrelationData correlationData1 = new CorrelationData("1");
        String routingKey = "key1";
        rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME, routingKey, message + routingKey, correlationData1);
        log.info(routingKey + "发送消息内容:{}", message + routingKey);

        CorrelationData correlationData2 = new CorrelationData("2");
        routingKey = "key2";
        rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME, routingKey, message + routingKey, correlationData2);
        log.info(routingKey + "发送消息内容:{}", message + routingKey);
    }
}

@Component
@Slf4j
public class ConfirmConsumer {
    public static final String CONFIRM_QUEUE_NAME = "confirm.queue";

    @RabbitListener(queues = CONFIRM_QUEUE_NAME)
    public void receiveMsg(Message message) {
        String msg = new String(message.getBody());
        log.info("消费方法接受到队列 confirm.queue 消息:{}", msg);
    }
}

#消息退回
spring.rabbitmq.publisher-returns=true

//当消息无法路由的时候的回调方法
@Override
public void returnedMessage(ReturnedMessage returned) {

    log.error("消息：{}，被交换机 {} 退回，原因：{}，路由key：{},code:{}",
              new String(returned.getMessage().getBody()), returned.getExchange(),
              returned.getReplyText(), returned.getRoutingKey(),
              returned.getReplyCode());

}

@Override
public void returnedMessage(Message message, int replyCode, String replyText, String
exchange, String routingKey) {
	log.info("消息:{}被服务器退回，退回原因:{}, 交换机是:{}, 路由 key:{}",new String(message.getBody()),replyText, exchange, routingKey);
}

//依赖注入 rabbitTemplate 之后再设置它的回调对象
@PostConstruct
public void init() {
    rabbitTemplate.setConfirmCallback(myCallBack);
    /**
    * true：交换机无法将消息进行路由时，会将该消息返回给生产者
    * false：如果发现消息无法进行路由，则直接丢弃
    */
    rabbitTemplate.setMandatory(true);
    //设置回退消息交给谁处理
    rabbitTemplate.setReturnsCallback(myCallBack);
}

Map<String, Object> params = new HashMap();
params.put("x-max-priority", 10);
channel.queueDeclare("hello", true, false, false, params);

AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().priority(10).build();

public class PriorityProducer {
    private static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();

        //给消息赋予一个 priority 属性
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().priority(10).build();

        for (int i = 1; i < 11; i++) {
            String message = "info" + i;
            if (i == 5) {
                channel.basicPublish("", QUEUE_NAME, properties, message.getBytes());
            } else {
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            }
            System.out.println("发送消息完成:" + message);
        }
    }
}

public class PriorityConsumer {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();

        //设置队列的最大优先级 最大可以设置到 255 官网推荐 1-10 如果设置太高比较吃内存和 CPU
        Map<String, Object> params = new HashMap();
        params.put("x-max-priority", 10);
        channel.queueDeclare(QUEUE_NAME, true, false, false, params);

        //推送的消息如何进行消费的接口回调
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody());
            System.out.println(message);
        };
        //取消消费的一个回调接口 如在消费的时候队列被删除掉了
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("消息消费被中断");
        };

        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
    }

}

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-queue-mode", "lazy");
channel.queueDeclare("myqueue", false, false, false, args);