RabbitMQ工作模式中的发布确认模式示例详解_Java

发布确认模式

概述

发布确认模式用于确保消息已经被正确地发送到rabbitmq服务器，并被成功接收和持久化。通过使用发布确认，生产者可以获得对消息的可靠性保证，避免消息丢失。这一机制基于通道（channel）级别，通过两个阶段的确认来保证消息的可靠性。

消息丢失问题

作为消息中间件, 都会⾯临消息丢失的问题.
消息丢失⼤概分为三种情况:

1. ⽣产者问题. 因为应⽤程序故障, ⽹络抖动等各种原因, ⽣产者没有成功向broker发送消息.
2. 消息中间件⾃⾝问题. ⽣产者成功发送给了broker, 但是broker没有把消息保存好, 导致消息丢失.
3. 消费者问题. broker 发送消息到消费者, 消费者在消费消息时, 因为没有处理好, 导致broker将消费失败的消息从队列中删除了。

rabbitmq也对上述问题给出了相应的解决⽅案. 问题2可以通过持久化机制. 问题3可以采⽤消息应答机制.
针对问题1, 可以采⽤发布确认(publisher confirms)机制实现.

发布确认的三种模式

rabbitmq的发布确认模式主要有三种形式：单条确认、批量确认和异步确认。

单条确认（single publisher confirm）

特点：在发布一条消息后，等待服务器确认该消息是否成功接收。
优点：实现简单，每条消息的确认状态清晰。
缺点：性能开销较大，特别是在高并发的场景下，因为每条消息都需要等待服务器的确认。

批量确认（batch publisher confirm）

特点：允许在一次性确认多个消息是否成功被服务器接收。
优点：在大量消息的场景中可以提高效率，因为可以减少确认消息的数量。
缺点：当一批消息中有一条消息发送失败时，整个批量确认失败。此时需要重新发送整批消息，但不知道是哪条消息发送失败，增加了调试和处理的难度。

异步确认（asynchronous confirm）

特点：通过回调函数处理消息的确认和未确认事件，更加灵活。
优点：在异步场景中能够更好地处理消息的状态，提高了系统的并发性能和响应速度。
缺点：实现相对复杂，需要处理回调函数的逻辑和状态管理。

实现步骤

1.设置通道为发布确认模式：在生产者发送消息之前，需要将通道设置为发布确认模式。这可以通过调用channel.confirmselect()方法来实现。
2.发送消息并等待确认：生产者发送消息时，每条消息都会分配一个唯一的、递增的整数id（deliverytag）。生产者可以通过调用channel.waitforconfirms()方法来等待所有已发送消息的确认，或者通过其他方式处理确认回调。
3.处理确认回调：为了处理确认回调，需要创建一个confirmcallback接口的实现。在实现的handleack()方法中，可以处理成功接收到确认的消息的逻辑；在handlenack()方法中，可以处理未成功接收到确认的消息的逻辑。

应用场景

发布确认模式适用于对数据安全性要求较高的场景，如金融交易、订单处理等。在这些场景中，消息的丢失或重复都可能导致严重的业务问题。通过使用发布确认模式，可以确保消息被正确地发送到rabbitmq服务器，并被成功接收和持久化，从而提高了系统的可靠性和稳定性。

代码案例

引入依赖

<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency>
    <groupid>com.rabbitmq</groupid>
    <artifactid>amqp-client</artifactid>
    <version>5.21.0</version>
</dependency>

常量类

public class constants {
    public static final string host = "47.98.109.138";
    public static final int port = 5672;
    public static final string user_name = "study";
    public static final string password = "study";
    public static final string virtual_host = "aaa";
    //publisher confirms
    public static final string publisher_confirms_queue1 = "publisher.confirms.queue1";
    public static final string publisher_confirms_queue2 = "publisher.confirms.queue2";
    public static final string publisher_confirms_queue3 = "publisher.confirms.queue3";
}

单条确认

import com.rabbitmq.client.channel;
import com.rabbitmq.client.confirmlistener;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;
import rabbitmq.constant.constants;
import java.io.ioexception;
import java.util.collections;
import java.util.sortedset;
import java.util.treeset;
public class publisherconfirms {
    private static final integer message_count = 100;
    static connection createconnection() throws exception {
        connectionfactory connectionfactory = new connectionfactory();
        connectionfactory.sethost(constants.host);
        connectionfactory.setport(constants.port); //需要提前开放端口号
        connectionfactory.setusername(constants.user_name);//账号
        connectionfactory.setpassword(constants.password);  //密码
        connectionfactory.setvirtualhost(constants.virtual_host); //虚拟主机
        return connectionfactory.newconnection();
    }
    public static void main(string[] args) throws exception {
        //strategy #1: publishing messages individually
        //单独确认
        publishingmessagesindividually();
    }
    /**
     * 单独确认
     */
    private static void publishingmessagesindividually() throws exception {
        try(connection connection = createconnection()) {
            //1. 开启信道
            channel channel = connection.createchannel();
            //2. 设置信道为confirm模式
            channel.confirmselect();
            //3. 声明队列
            channel.queuedeclare(constants.publisher_confirms_queue1, true, false, false, null);
            //4. 发送消息, 并等待确认
            long start = system.currenttimemillis();
            for (int i = 0; i < message_count; i++) {
                string msg = "hello publisher confirms"+i;
                channel.basicpublish("",constants.publisher_confirms_queue1, null, msg.getbytes());
                //等待确认
                channel.waitforconfirmsordie(5000);
            }
            long end = system.currenttimemillis();
            system.out.printf("单独确认策略, 消息条数: %d, 耗时: %d ms \n",message_count, end-start);
        }
    }
}

运行代码

我们可以看到，以发送消息条数为100条为例，单条确认模式是非常耗时的。

批量确认

import com.rabbitmq.client.channel;
import com.rabbitmq.client.confirmlistener;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;
import rabbitmq.constant.constants;
import java.io.ioexception;
import java.util.collections;
import java.util.sortedset;
import java.util.treeset;
public class publisherconfirms {
    private static final integer message_count = 10000;
    static connection createconnection() throws exception {
        connectionfactory connectionfactory = new connectionfactory();
        connectionfactory.sethost(constants.host);
        connectionfactory.setport(constants.port); //需要提前开放端口号
        connectionfactory.setusername(constants.user_name);//账号
        connectionfactory.setpassword(constants.password);  //密码
        connectionfactory.setvirtualhost(constants.virtual_host); //虚拟主机
        return connectionfactory.newconnection();
    }
    public static void main(string[] args) throws exception {
        //strategy #2: publishing messages in batches
        //批量确认
        publishingmessagesinbatches();
    }
    /**
     * 批量确认
     * @throws exception
     */
    private static void publishingmessagesinbatches() throws exception{
        try(connection connection = createconnection()) {
            //1. 开启信道
            channel channel = connection.createchannel();
            //2. 设置信道为confirm模式
            channel.confirmselect();
            //3. 声明队列
            channel.queuedeclare(constants.publisher_confirms_queue2, true, false, false, null);
            //4. 发送消息, 并进行确认
            long start = system.currenttimemillis();
            int batchsize = 100;
            int outstandingmessagecount = 0;
            for (int i = 0; i < message_count; i++) {
                string msg = "hello publisher confirms"+i;
                channel.basicpublish("",constants.publisher_confirms_queue2, null, msg.getbytes());
                outstandingmessagecount++;
                if (outstandingmessagecount==batchsize){
                    channel.waitforconfirmsordie(5000);
                    outstandingmessagecount = 0;
                }
            }
            if (outstandingmessagecount>0){
                channel.waitforconfirmsordie(5000);
            }
            long end = system.currenttimemillis();
            system.out.printf("批量确认策略, 消息条数: %d, 耗时: %d ms \n",message_count, end-start);
        }
    }
}

运行代码

我们可以看到，以发送消息条数为10000条为例，单条确认模式是比较快的。

异步确认

import com.rabbitmq.client.channel;
import com.rabbitmq.client.confirmlistener;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;
import rabbitmq.constant.constants;
import java.io.ioexception;
import java.util.collections;
import java.util.sortedset;
import java.util.treeset;
public class publisherconfirms {
    private static final integer message_count = 10000;
    static connection createconnection() throws exception {
        connectionfactory connectionfactory = new connectionfactory();
        connectionfactory.sethost(constants.host);
        connectionfactory.setport(constants.port); //需要提前开放端口号
        connectionfactory.setusername(constants.user_name);//账号
        connectionfactory.setpassword(constants.password);  //密码
        connectionfactory.setvirtualhost(constants.virtual_host); //虚拟主机
        return connectionfactory.newconnection();
    }
    public static void main(string[] args) throws exception {
        //strategy #3: handling publisher confirms asynchronously
        //异步确认
        handlingpublisherconfirmsasynchronously();
    }
    /**
     * 异步确认
     */
    private static void handlingpublisherconfirmsasynchronously() throws exception{
        try (connection connection = createconnection()){
            //1. 开启信道
            channel channel = connection.createchannel();
            //2. 设置信道为confirm模式
            channel.confirmselect();
            //3. 声明队列
            channel.queuedeclare(constants.publisher_confirms_queue3, true, false, false, null);
            //4. 监听confirm
            //集合中存储的是未确认的消息id
            long start = system.currenttimemillis();
            sortedset<long> confirmseqno = collections.synchronizedsortedset(new treeset<>());
            channel.addconfirmlistener(new confirmlistener() {
                @override
                public void handleack(long deliverytag, boolean multiple) throws ioexception {
                    if (multiple){
                        confirmseqno.headset(deliverytag+1).clear();
                    }else {
                        confirmseqno.remove(deliverytag);
                    }
                }
                @override
                public void handlenack(long deliverytag, boolean multiple) throws ioexception {
                    if (multiple){
                        confirmseqno.headset(deliverytag+1).clear();
                    }else {
                        confirmseqno.remove(deliverytag);
                    }
                    //业务需要根据实际场景进行处理, 比如重发, 此处代码省略
                }
            });
            //5. 发送消息
            for (int i = 0; i < message_count; i++) {
                string msg = "hello publisher confirms"+i;
                long seqno = channel.getnextpublishseqno();
                channel.basicpublish("",constants.publisher_confirms_queue3, null, msg.getbytes());
                confirmseqno.add(seqno);
            }
            while (!confirmseqno.isempty()){
                thread.sleep(10);
            }
            long end = system.currenttimemillis();
            system.out.printf("异步确认策略, 消息条数: %d, 耗时: %d ms \n",message_count, end-start);
        }
    }
}

运行代码