<!--rabbitmq 依賴客戶端-->
<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>amqp-client</artifactId>
  <version>5.8.0</version>
</dependency>

public class Producer {
    //建立隊(duì)列
    private static final String QUEUE_NAME="hello";
    public static void main(String[] args)  {
        //創(chuàng)建連接工場(chǎng)
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setUsername("guest");
        factory.setUsername("guest");

        try {
            //建立連接和信道
            //channel 實(shí)現(xiàn)了自動(dòng) close 接口 自動(dòng)關(guān)閉 不需要顯示關(guān)閉
            Connection connection=factory.newConnection();
            Channel channel=connection.createChannel();

            /**
             * 生成一個(gè)隊(duì)列,并將信道和隊(duì)列連接
             * 1.隊(duì)列名稱
             * 2.隊(duì)列里面的消息是否持久化 默認(rèn)消息存儲(chǔ)在內(nèi)存中
             * 3.該隊(duì)列是否只供一個(gè)消費(fèi)者進(jìn)行消費(fèi) 是否進(jìn)行共享 true 可以多個(gè)消費(fèi)者消費(fèi)
             * 4.是否自動(dòng)刪除 最后一個(gè)消費(fèi)者端開連接以后 該隊(duì)列是否自動(dòng)刪除 true 自動(dòng)刪除
             * 5.其他參數(shù)
             */
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            String message="hello world";
            /**
             * 發(fā)送一個(gè)消息
             * 1.發(fā)送到那個(gè)交換機(jī)
             * 2.路由的 key 是哪個(gè)
             * 3.其他的參數(shù)信息
             * 4.發(fā)送消息的消息體
             */
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            System.out.println("消息發(fā)送成功");
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

public class Consumer {
    //定義隊(duì)列名
    private static final String QUEUE_NAME="hello";

    public static void main(String[] args) {
        //建立連接和信道
        try {
            ConnectionFactory factory=new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setUsername("guest");
            factory.setPassword("guest");
            Connection connection=factory.newConnection();
            Channel channel=connection.createChannel();
            System.out.println("等待接收消息");


            /**
             *1.同一個(gè)會(huì)話， consumerTag 是固定的 可以做此會(huì)話的名字， deliveryTag 每次接收消息+1，可以做此消息處理通道的名字。
             *2.包含消息的字節(jié)形式的類
             */
            DeliverCallback deliverCallback=(consumerTag,delivery)->{
                String message=new String(delivery.getBody());
                System.out.println(message);
            };
            CancelCallback cancelCallback=(consumerTag)->{
                System.out.println("消息消費(fèi)被取消");
            };
            /* 消費(fèi)者消費(fèi)消息
             * 1.消費(fèi)哪個(gè)隊(duì)列
             * 2.消費(fèi)成功之后是否要自動(dòng)應(yīng)答 true 代表自動(dòng)應(yīng)答 false 手動(dòng)應(yīng)答
             * 3.消費(fèi)者未成功消費(fèi)的回調(diào)
             * 4.消費(fèi)者取消消費(fèi)的回調(diào)
             */
            channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);



        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

public class Producer {
    public static void main(String[] args) throws IOException, InterruptedException {
        Channel channel=RabbitMQChannelUtil.getChannel();
        if(channel == null){
            System.out.println("失敗");
            return;
        }
        channel.queueDeclare(RabbitMQChannelUtil.QUEUE_NAME,false,false,false,null);
        int i=0;
        while (true){
            String message="消息"+i;
            i++;
            /**
             * 發(fā)送一個(gè)消息
             * 1.發(fā)送到那個(gè)交換機(jī)
             * 2.路由的 key 是哪個(gè)
             * 3.其他的參數(shù)信息
             * 4.發(fā)送消息的消息體
             */
            channel.basicPublish("",RabbitMQChannelUtil.QUEUE_NAME,null,message.getBytes());
            System.out.println(message);
            Thread.sleep(500);
        }
    }
}

public class Consumer {
    public static void main(String[] args) {
        Channel channel=RabbitMQChannelUtil.getChannel();
        if(channel == null){
            System.out.println("消費(fèi)失敗");
            return;
        }
        DeliverCallback deliverCallback=(consumerTag, delivery)->{
            String message=new String(delivery.getBody());
            System.out.println(Thread.currentThread().getName()+"消費(fèi)了"+message);
        };
        CancelCallback cancelCallback=(consumerTag)->{
            System.out.println("消息消費(fèi)被取消");
        };
        Thread[] threads=new Thread[5];
        for (int i = 0; i <threads.length ; i++) {
            threads[i]=new Thread(()->{
                try {
                    System.out.println(Thread.currentThread().getName()+"啟動(dòng)等待消費(fèi)");
                    channel.basicConsume(RabbitMQChannelUtil.QUEUE_NAME,true,deliverCallback,cancelCallback);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        }
        for (int i = 0; i <threads.length ; i++) {
            threads[i].start();
        }
    }
}

/* 消費(fèi)者消費(fèi)消息
 * 1.消費(fèi)哪個(gè)隊(duì)列
 * 2.消費(fèi)成功之后是否要自動(dòng)應(yīng)答 true 代表自動(dòng)應(yīng)答 false 手動(dòng)應(yīng)答
 * 3.當(dāng)一個(gè)消息發(fā)送過(guò)來(lái)后的回調(diào)接口
 * 4.消費(fèi)者取消消費(fèi)的回調(diào)
 */
boolean ack=false;
channel.basicConsume(QUEUE_NAME,ack,deliverCallback,cancelCallback);

/**
 * 消息發(fā)送過(guò)來(lái)后的回調(diào)接口
 *1.同一個(gè)會(huì)話， consumerTag 是固定的 可以做此會(huì)話的名字， deliveryTag 每次接收消息+1，可以做此消息處理通道的名字。
 *2.消息類
 */
DeliverCallback deliverCallback=(consumerTag,delivery)->{
    String message=new String(delivery.getBody());
    System.out.println(message);
    /**
     * 參數(shù)說(shuō)明
     * 1.消息的標(biāo)記tag
     * 2.是否批量應(yīng)答
     */
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
};

channel.queueDeclare(RabbitMQChannelUtil.QUEUE_NAME,true,false,false,null);

channel.basicPublish("",RabbitMQChannelUtil.QUEUE_NAME,null,message.getBytes());

channel.basicPublish("",RabbitMQChannelUtil.QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());

channel.basicQos(1);

channel.confirmSelect();

public static void singleConfirm(){
    try {
        Channel channel=RabbitMQChannelUtil.getChannel();
        if(channel == null){
            System.out.println("信道建立失敗");
            return;
        }
        //開啟發(fā)布確認(rèn)
        channel.confirmSelect();
        long begin=System.currentTimeMillis();
        for (int i = 0; i <MESSAGE_COUNT ; i++) {
            String message=i+"";
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            //可以加時(shí)間參數(shù)，當(dāng)消息發(fā)送失敗或超過(guò)參數(shù)時(shí)間沒成功，則返回false
            boolean flag=channel.waitForConfirms();
            //如果失敗可以重發(fā)
            if(flag){
                System.out.println(message+"發(fā)送成功");
            }else {
                //這里可以實(shí)現(xiàn)重發(fā)
                System.out.println(message+"發(fā)送失敗");
            }
        }
        long end=System.currentTimeMillis();
        System.out.println("發(fā)送"+MESSAGE_COUNT+"條消息，耗時(shí)"+(end-begin)+"ms");
    }catch (Exception e){
        e.printStackTrace();
    }
}

public static void batchConfirm(){
    try {
        Channel channel=RabbitMQChannelUtil.getChannel();
        if(channel == null){
            System.out.println("建立連接失敗");
            return;
        }
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //當(dāng)100條消息發(fā)布成功時(shí)，再確認(rèn)
        int ackMessageCount=100;
        //未確認(rèn)的消息個(gè)數(shù)
        int needAckMessageCount=0;
        //開啟發(fā)布確認(rèn)
        channel.confirmSelect();
        long begin=System.currentTimeMillis();
        for (int i = 0; i <MESSAGE_COUNT ; i++) {
            String message=i+"";
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            needAckMessageCount++;
            if(needAckMessageCount == ackMessageCount){
                //確認(rèn)
                channel.waitForConfirms();
                needAckMessageCount=0;
            }
        }
        //判斷可能還有消息未發(fā)送，再發(fā)送依次
        if(needAckMessageCount > 0){
            channel.waitForConfirms();
        }
        long end= System.currentTimeMillis();
        System.out.println("發(fā)送"+MESSAGE_COUNT+"條消息，耗時(shí)"+(end-begin)+"ms");
    }catch (Exception e){
        e.printStackTrace();
    }
}

public static void asyncConfirm() throws Exception {
    try (Channel channel = RabbitMQChannelUtil.getChannel()) {
        if(channel == null){
            return;
        }
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //開啟發(fā)布確認(rèn)
        channel.confirmSelect();
        /**
         * 線程安全有序的一個(gè)哈希表，適用于高并發(fā)的情況
         * 1.輕松的將序號(hào)與消息進(jìn)行關(guān)聯(lián)
         * 2.輕松批量刪除條目 只要給到序列號(hào)
         * 3.支持并發(fā)訪問
         */
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new
                ConcurrentSkipListMap<>();
        /**
         * 確認(rèn)收到消息的一個(gè)回調(diào)
         * 1.消息序列號(hào)
         * 2.true 可以確認(rèn)小于等于當(dāng)前序列號(hào)的消息
         * false 確認(rèn)當(dāng)前序列號(hào)消息
         */
        ConfirmCallback ackCallback = (sequenceNumber, multiple) -> {
            if (multiple) {
                //返回的是小于等于當(dāng)前序列號(hào)的未確認(rèn)消息 是一個(gè) map
                ConcurrentNavigableMap<Long, String> confirmed =
                        outstandingConfirms.headMap(sequenceNumber, true);
                //清除該部分未確認(rèn)消息
                confirmed.clear();
            }else{
                //只清除當(dāng)前序列號(hào)的消息
                outstandingConfirms.remove(sequenceNumber);
            }
        };
        ConfirmCallback nackCallback = (sequenceNumber, multiple) -> {
            String message = outstandingConfirms.get(sequenceNumber);
            System.out.println("發(fā)布的消息"+message+"未被確認(rèn)，序列號(hào)"+sequenceNumber);
        };
        /**
         * 添加一個(gè)異步確認(rèn)的監(jiān)聽器
         * 1.確認(rèn)收到消息的回調(diào)
         * 2.未收到消息的回調(diào)
         */
        channel.addConfirmListener(ackCallback, null);
        long begin = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            /**
             * channel.getNextPublishSeqNo()獲取下一個(gè)消息的序列號(hào)
             * 通過(guò)序列號(hào)與消息體進(jìn)行一個(gè)關(guān)聯(lián)
             * 全部都是未確認(rèn)的消息體
             */
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        }
        long end = System.currentTimeMillis();
        System.out.println("發(fā)布" + MESSAGE_COUNT + "個(gè)異步確認(rèn)消息,耗時(shí)" + (end - begin) +
                "ms");
    } 
}

channel.basicPublish("", QUEUE_NAME, null, message.getBytes());

String queueName = channel.queueDeclare().getQueue();

String queueName = channel.queueDeclare().getQueue();

//聲明交換機(jī)名稱及類型
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
//把該臨時(shí)隊(duì)列綁定我們的 exchange 其中 routingkey(也稱之為 binding key)為空字符串
 channel.queueBind(queueName, EXCHANGE_NAME, "");

public class Producer {
    private static final String NORMAL_EXCHANGE="normal_exchange";
    public static void main(String[] args) {
        try {
            Channel channel= RabbitMQChannelUtil.getChannel();
            if(channel == null){
                return;
            }
            //聲明交換機(jī)類型
            channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
            //設(shè)置消息TTL時(shí)間
            AMQP.BasicProperties basicProperties=new AMQP.BasicProperties().builder().expiration("1000").build();
            //用作演示消息隊(duì)列的限制個(gè)數(shù)
            for (int i = 0; i <10 ; i++) {
                String message="info"+i;
                channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",basicProperties,message.getBytes());
                System.out.println("生產(chǎn)者發(fā)送消息");
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

public class NormalConsumer {
    //普通交換機(jī)名稱
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    //死信交換機(jī)名稱
    private static final String DEAD_EXCHANGE = "dead_exchange";
    public static void main(String[] argv) throws Exception {
        Channel channel = RabbitMQChannelUtil.getChannel();
        if(channel == null){
            return;
        }
        //聲明死信和普通交換機(jī) 類型為 direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        //聲明死信隊(duì)列
        String deadQueue = "dead-queue";
        channel.queueDeclare(deadQueue, false, false, false, null);
        //死信隊(duì)列綁定死信交換機(jī)與 routingkey
        channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
        //正常隊(duì)列綁定死信隊(duì)列信息
        Map<String, Object> params = new HashMap<>();
        //正常隊(duì)列設(shè)置死信交換機(jī) 參數(shù) key 是固定值
        params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        //正常隊(duì)列設(shè)置死信 routing-key 參數(shù) 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("NormalConsumer 接收到消息"+message);
        };
        channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> {
        });
    }
}

public class DeadConsumer {
 private static final String DEAD_EXCHANGE = "dead_exchange";
 public static void main(String[] argv) throws Exception {
        Channel channel = RabbitMQChannelUtil.getChannel();
        if (channel == null) {
         return;
        }
        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("等待接收死信隊(duì)列消息.....");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
         String message = new String(delivery.getBody(), "UTF-8");
         System.out.println("DeadConsumer 接收死信隊(duì)列的消息" + message);
        };
        channel.basicConsume(deadQueue, true, deliverCallback, consumerTag -> {
        });
 }