channel.txSelect(): 將當(dāng)前信道設(shè)置成事務(wù)模式
channel.txCommit(): 用于提交事務(wù)
channel.txRollback(): 用于回滾事務(wù)

package confirm;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import util.ConnectionUtil;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class PublisherConfirmsProducer {
    public static void main(String[] args) throws Exception{
        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();
        // 向RabbitMQ服務(wù)器發(fā)送AMQP命令，將當(dāng)前通道標(biāo)記為發(fā)送方確認(rèn)通道
        final AMQP.Confirm.SelectOk selectOk = channel.confirmSelect();
        channel.queueDeclare("queue.pc", true, false, false, null);
        channel.exchangeDeclare("ex.pc", "direct", true, false, null);
        channel.queueBind("queue.pc", "ex.pc", "key.pc");
        try {
        // 發(fā)送消息
        for (int i = 1 ; i < 10000 ; i++){
            channel.basicPublish("ex.pc", "key.pc", null, "hello world".getBytes());
        }
            // 同步的方式等待RabbitMQ的確認(rèn)消息
            channel.waitForConfirmsOrDie(5000);
            System.out.println("發(fā)送的消息已經(jīng)得到確認(rèn)");
        } catch (IOException ex) {
            System.out.println("消息被拒收");
        } catch (IllegalStateException ex) {
            System.out.println("發(fā)送消息的通道不是PublisherConfirms通道");
        } catch (TimeoutException ex) {
            System.out.println("等待消息確認(rèn)超時(shí)");
        }
        channel.close();
        connection.close();
    }
}

package confirm;
/**
 * 創(chuàng)建者: 魏紅
 * 創(chuàng)建時(shí)間: 2023-02-28
 * 描述:
 */
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import util.ConnectionUtil;
public class PublisherConfirmsProducer2 {
    public static void main(String[] args) throws Exception {
        //獲取連接
        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();
        // 向RabbitMQ服務(wù)器發(fā)送AMQP命令，將當(dāng)前通道標(biāo)記為發(fā)送方確認(rèn)通道
        final AMQP.Confirm.SelectOk selectOk = channel.confirmSelect();
        channel.queueDeclare("queue.pc", true, false, false, null);
        channel.exchangeDeclare("ex.pc", "direct", true, false, null);
        channel.queueBind("queue.pc", "ex.pc", "key.pc");
        String message = "hello-";
        // 批處理的大小
        int batchSize = 10;
        // 用于對(duì)需要等待確認(rèn)消息的計(jì)數(shù)
        int outstrandingConfirms = 0;
        for (int i = 0; i < 10000; i++) {
            channel.basicPublish("ex.pc", "key.pc", null, (message + i).getBytes());
            outstrandingConfirms++;
            if (outstrandingConfirms == batchSize) {
                // 此時(shí)已經(jīng)有一個(gè)批次的消息需要同步等待broker的確認(rèn)消息
                // 同步等待
                channel.waitForConfirmsOrDie(5000);
                System.out.println("消息已經(jīng)被確認(rèn)了");
                outstrandingConfirms = 0;
            }
        }
        if (outstrandingConfirms > 0) {
            channel.waitForConfirmsOrDie(5000);
            System.out.println("剩余消息已經(jīng)被確認(rèn)了");
        }
        channel.close();
        connection.close();
    }
}

package confirm;
/**
 * 創(chuàng)建者: 魏紅
 * 創(chuàng)建時(shí)間: 2023-02-28
 * 描述:
 */
import com.rabbitmq.client.*;
import util.ConnectionUtil;
import javax.management.loading.MLet;
import java.io.IOException;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
public class PublisherConfirmsProducer3 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();
        // 向RabbitMQ服務(wù)器發(fā)送AMQP命令，將當(dāng)前通道標(biāo)記為發(fā)送方確認(rèn)通道
        final AMQP.Confirm.SelectOk selectOk = channel.confirmSelect();
        channel.queueDeclare("queue.pc", true, false, false, null);
        channel.exchangeDeclare("ex.pc", "direct", true, false, null);
        channel.queueBind("queue.pc", "ex.pc", "key.pc");
//        ConfirmCallback clearOutstandingConfirms = new ConfirmCallback() {
//            @Override
//            public void handle(long deliveryTag, boolean multiple) throws IOException {
//                if (multiple) {
//                    System.out.println("編號(hào)小于等于 " + deliveryTag + " 的消息都已經(jīng)被確認(rèn)了");
//                } else {
//                    System.out.println("編號(hào)為：" + deliveryTag + " 的消息被確認(rèn)");
//                }
//            }
//        };
        ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
        ConfirmCallback clearOutstandingConfirms = (deliveryTag, multiple) -> {
            if (multiple) {
                System.out.println("編號(hào)小于等于 " + deliveryTag + " 的消息都已經(jīng)被確認(rèn)了");
                final ConcurrentNavigableMap<Long, String> headMap
                        = outstandingConfirms.headMap(deliveryTag, true);
                // 清空outstandingConfirms中已經(jīng)被確認(rèn)的消息信息
                headMap.clear();
            } else {
                // 移除已經(jīng)被確認(rèn)的消息
                outstandingConfirms.remove(deliveryTag);
                System.out.println("編號(hào)為：" + deliveryTag + " 的消息被確認(rèn)");
            }
        };
        ConfirmCallback confirmCallback = (deliveryTag, multiple) -> {
            if (multiple) {
                // 將沒有確認(rèn)的消息記錄到一個(gè)集合中
                // 此處省略實(shí)現(xiàn)
                System.out.println("消息編號(hào)小于等于：" + deliveryTag + " 的消息 不確認(rèn)");
            } else {
                System.out.println("編號(hào)為：" + deliveryTag + " 的消息不確認(rèn)");
            }
        };
        // 設(shè)置channel的監(jiān)聽器，處理確認(rèn)的消息和不確認(rèn)的消息
        channel.addConfirmListener(clearOutstandingConfirms, confirmCallback);
        String message = "hello-";
        for (int i = 0; i < 500000; i++) {
            // 獲取下一條即將發(fā)送的消息的消息ID
            final long nextPublishSeqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("ex.pc", "key.pc", null, (message + i).getBytes());
            System.out.println("編號(hào)為：" + nextPublishSeqNo + " 的消息已經(jīng)發(fā)送成功，尚未確認(rèn)");
            outstandingConfirms.put(nextPublishSeqNo, (message + i));
        }
        // 等待消息被確認(rèn)
        Thread.sleep(10000);
        channel.close();
        connection.close();
    }
}

package workmode;
import com.rabbitmq.client.*;
import util.ConnectionUtil;
import java.io.IOException;
/**
* NONE模式，則只要收到消息后就立即確認(rèn)（消息出列，標(biāo)記已消費(fèi)），有丟失數(shù)據(jù)的風(fēng)險(xiǎn)
* AUTO模式，看情況確認(rèn)，如果此時(shí)消費(fèi)者拋出異常則消息會(huì)返回到隊(duì)列中
* MANUAL模式，需要顯式的調(diào)用當(dāng)前channel的basicAck方法
 */
public class Recer2 {
    public static void main(String[] args) throws  Exception {
        // 1.獲得連接
        Connection connection = ConnectionUtil.getConnection();
        // 2.獲得通道（信道）
        final Channel channel = connection.createChannel();
        channel.queueDeclare("work_queue",false,false,false,null);
        // 3.從信道中獲得消息
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override //交付處理（收件人信息，包裹上的快遞標(biāo)簽，協(xié)議的配置，消息）
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String s = new String(body);
//                System.out.println("【顧客2】吃掉 " + s+" ! 總共吃【"+i+++"】串！");
                System.out.println("【消費(fèi)者2】得到 " + s);
                // 模擬網(wǎng)絡(luò)延遲
                try{
                    Thread.sleep(400);
                }catch (Exception e){
                }
                // 手動(dòng)確認(rèn)（收件人信息，是否同時(shí)確認(rèn)多個(gè)消息）
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        // 4.監(jiān)聽隊(duì)列 false:手動(dòng)消息確認(rèn)
        channel.basicConsume("work_queue", false,consumer);
    }
}