RabbitMQ系列—Java操作之事务模式和Confirm模式

迷南。 2022-05-08 11:14 304阅读 0赞

在之前介绍到了RabbitMQ的消息持久化和消费者端手动确认，解决了消费者异常导致的数据丢失问题，那么我们如何确定生产者生产的消息已经被发送到rabbitmq服务器了呢？通俗点说，如果消息经过交换器进入队列就可以完成消息的持久化，但如果消息在没有到达broker之前出现意外，那就造成消息丢失，有没有办法可以解决这个问题？有两种方式：

*  通过AMQP协议，AMQP协议实现了事务机制。
 *  通过Confirm模式

### AMQP的事务机制 ###

事务的实现主要是对通道（Channel）的设置，主要的方法有三个：

*  channel.txSelect()：声明启动事务模式
 *  channel.txComment()：提交事务
 *  channel.txRollback()：回滚事务

**生产者Sender**

public class Sender {
    	private static final String QUEUE = "test_tx_queue";
    
    	public static void main(String[] args) {
    		Connection con = null;
    		Channel channel = null;
    		try {
    			// 获取连接
    			con = ConnectionUtils.getConnection();
    			// 从连接中创建通道
    			channel = con.createChannel();
    			// 声明一个队列
    			channel.queueDeclare(QUEUE, false, false, false, null);
    			// 消息内容
    			String msg = "tx queue hello!";
    			// 开启事务
    			channel.txSelect();
    			// 发送消息
    			channel.basicPublish("", QUEUE, null, msg.getBytes());
    			// 模拟异常
    			int num = 1/0;
    			// 提交事务
    			channel.txCommit();
    			System.out.println("send success");
    		} catch (Exception e) {
    			// 事务回滚
    			try {
    				channel.txRollback();
    			} catch (IOException e1) {
    				e1.printStackTrace();
    			}
    			e.printStackTrace();
    		} finally {
    			// 关闭连接
    			ConnectionUtils.close(channel, con);
    		}
    
    	}
    
    }

**消费者Recver**

public class Recver {
    	private static final String QUEUE = "test_tx_queue";
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		// 获取连接
    		Connection con = ConnectionUtils.getConnection();
    		// 从连接中创建通道
    		Channel channel = con.createChannel();
    		// 声明队列
    		channel.queueDeclare(QUEUE, false, false, false, null);
    		// 创建消费者
    		Consumer consumer = new DefaultConsumer(channel) {
    			// 获取消息
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				String msg = new String(body, "utf-8");
    				System.out.println("接收到消息——" + msg);
    			}
    
    		};
    		// 监听队列
    		channel.basicConsume(QUEUE, true, consumer);
    	}
    }

运行消费者和生产者，生产者报错而且消费者也没有收到消息，说明消息已经被回滚了。

**消费者模式使用事务**

我们知道，消费者可以使用消息自动或手动发送来确认消费消息，那如果我们在消费者模式中使用事务（当然如果使用了手动确认消息，完全用不到事务的），会发生什么呢？

结果分为两种情况：

*  autoAck=false手动应对的时候是支持事务的，也就是说即使你已经手动确认了消息已经收到了，但在确认消息会等事务的返回解决之后，在做决定是确认消息还是重新放回队列，如果你手动确认现在之后，又回滚了事务，那么已事务回滚为主，此条消息会重新放回队列。
 *  autoAck=true如果自定确认为true的情况是不支持事务的，也就是说你即使在收到消息之后在回滚事务也是于事无补的，队列已经把消息移除了。

这种事务模式有个缺陷：性能差，降低了rabbitmq的消息吞吐量，使用了事务模式比非事务模式性能差很多，那么有没有既能保证消息的可靠性又能兼顾性能的解决方案呢？那就是下面的Confirm模式。

### Confirm模式 ###

Confirm发送方确认模式使用和事务类似，也是通过设置Channel进行发送方确认的。

**实现原理**

将Channel设置为Confirm模式后，此Channel发送的每条消息都会有标识这条消息的ID（从1开始），当r消息投放到匹配的队列后，broker会返回一个确认信息（包含消息的唯一ID）给生产者通知生产者已经成功发送到队列。如果消息和队列是可持久化的，在队列将消息写人到磁盘后再返回给生产者确认信息。broker回传给生产者的确认消息中deliver-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域，表示这个序列号之前的所有消息都已经得到了处理。

**三种编程方式**

*  串行confirm模式：peoducer每发送一条消息后，调用waitForConfirms()方法，等待broker端confirm。
 *  批量confirm模式：producer每发送一批消息后，调用waitForConfirms()方法，等待broker端confirm。
 *  异步confirm模式：提供一个回调方法，broker confirm了一条或者多条消息后producer端会回调这个方法。

Confirm模式最大的优点就是它是异步的。

**串行confirm模式**

生产者SingleSender

public class SingleSender {
    	private static final String QUEUE = "test_confirm_queue";
    
    	public static void main(String[] args) {
    		Connection con = null;
    		Channel channel = null;
    		try {
    			// 获取连接
    			con = ConnectionUtils.getConnection();
    			// 从连接中创建通道
    			channel = con.createChannel();
    			// 声明一个队列
    			channel.queueDeclare(QUEUE, false, false, false, null);
    			// 消息内容
    			String msg = "confirm queue hello!";
    			// 将Channel设置为Confirm模式
    			channel.confirmSelect();
    			// 发送消息
    			channel.basicPublish("", QUEUE, null, msg.getBytes());
    			// 消息确认
    			if(channel.waitForConfirms()){
    				System.out.println("send success");
    			}else{
    				System.out.println("send fail");
    			}
    			
    		} catch (IOException e) {
    			e.printStackTrace();
    		} catch (TimeoutException e) {
    			e.printStackTrace();
    		}catch (InterruptedException e) {
    			e.printStackTrace();
    		} finally {
    			// 关闭连接
    			ConnectionUtils.close(channel, con);
    		}
    
    	}
    }

消费者Recver

public class Recver{
    	private static final String QUEUE = "test_confirm_queue";
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		// 获取连接
    		Connection con = ConnectionUtils.getConnection();
    		// 从连接中创建通道
    		Channel channel = con.createChannel();
    		// 声明队列
    		channel.queueDeclare(QUEUE, false, false, false, null);
    		// 创建消费者
    		Consumer consumer = new DefaultConsumer(channel) {
    			// 获取消息
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				String msg = new String(body, "utf-8");
    				System.out.println("接收到消息——" + msg);
    			}
    
    		};
    		// 监听队列
    		channel.basicConsume(QUEUE, true, consumer);
    	}
    }

普通模式需要一条一条确认，性能慢，可以选择批量模式。

**批量confirm模式**

生产者BatchSender

public class BatchSender {
    	private static final String QUEUE = "test_confirm_queue";
    
    	public static void main(String[] args) {
    		Connection con = null;
    		Channel channel = null;
    		try {
    			// 获取连接
    			con = ConnectionUtils.getConnection();
    			// 从连接中创建通道
    			channel = con.createChannel();
    			// 声明一个队列
    			channel.queueDeclare(QUEUE, false, false, false, null);
    			// 将Channel设置为Confirm模式
    			channel.confirmSelect();
    			for (int i = 0; i < 20; i++) {
    				// 消息内容
    				String msg = "confirm queue hello!";
    				// 发送消息
    				channel.basicPublish("", QUEUE, null, msg.getBytes());
    			}
    			// 消息确认
    			if (channel.waitForConfirms()) {
    				System.out.println("send success");
    			} else {
    				System.out.println("send fail");
    			}
    
    		} catch (IOException e) {
    			e.printStackTrace();
    		} catch (TimeoutException e) {
    			e.printStackTrace();
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		} finally {
    			// 关闭连接
    			ConnectionUtils.close(channel, con);
    		}
    
    	}
    }

通过循环批量发送20条消息，但只在控制台输出了一行“发send success”，该方法会等到最后一条消息得到ack或者得到nack才会结束，也就是说在waitForConfirms处会造成当前程序的阻塞，这点我们看出broker端默认情况下是进行批量回复的，并不是针对每条消息都发送一条ack消息。

缺陷

一批数据中有一条消息发送失败会都回滚。

**异步模式**

普通模式和批量模式都是串行的、同步执行的，如果消息发送出去没有返回确认消息会一直等待，而异步模式执行效率高，不需要等待消息执行完，只需要监听消息即可。

生产者AsyncSender

public class AsyncSender {
    	private static final String QUEUE = "test_confirm_queue";
    
    	public static void main(String[] args) {
    		Connection con = null;
    		Channel channel = null;
    		try {
    			// 获取连接
    			con = ConnectionUtils.getConnection();
    			// 从连接中创建通道
    			channel = con.createChannel();
    			// 声明一个队列
    			channel.queueDeclare(QUEUE, false, false, false, null);
    			// 将Channel设置为Confirm模式
    			channel.confirmSelect();
    			// 异步监听确认和未确认的消息
    			channel.addConfirmListener(new ConfirmListener() {
    				/**
    				 * 处理返回确认成功
    				 * 
    				 * @param deliveryTag
    				 *            如果是多条，这个就是最后一条消息的tag
    				 * @param multiple
    				 *            是否多条 true是false否
    				 * @throws IOException
    				 */
    
    				public void handleAck(long deliveryTag, boolean multiple) throws IOException {
    					System.out.println("ack:deliveryTag:" + deliveryTag + "，multiple:" + multiple);
    				}
    
    				/**
    				 * 处理返回确认失败
    				 * 
    				 * @param deliveryTag
    				 * @param multiple
    				 * @throws IOException
    				 */
    				public void handleNack(long deliveryTag, boolean multiple) throws IOException {
    					System.out.println("nack:deliveryTag:" + deliveryTag + "，multiple:" + multiple);
    				}
    			});
    			for (int i = 0; i < 50; i++) {
    				// 消息内容
    				String msg = "confirm queue hello!" + i;
    				// long tag = channel.getNextPublishSeqNo();
    				// 发送消息
    				channel.basicPublish("", QUEUE, null, msg.getBytes());
    				// System.out.println("消息tag" + tag);
    			}
    
    			System.out.println("执行结束");
    		} catch (Exception e) {
    			e.printStackTrace();
    		} finally {
    			// 关闭连接
    			ConnectionUtils.close(channel, con);
    		}
    
    	}
    
    }

运行后控制台打印

ack:deliveryTag:2，multiple:true
    ack:deliveryTag:13，multiple:true
    ack:deliveryTag:14，multiple:false
    ack:deliveryTag:15，multiple:false
    ack:deliveryTag:16，multiple:false
    ack:deliveryTag:18，multiple:true
    ack:deliveryTag:20，multiple:true
    ack:deliveryTag:22，multiple:true
    ack:deliveryTag:23，multiple:false
    ack:deliveryTag:24，multiple:false
    ack:deliveryTag:25，multiple:false
    ack:deliveryTag:26，multiple:false
    ack:deliveryTag:27，multiple:false
    ack:deliveryTag:28，multiple:false
    执行结束
    ack:deliveryTag:29，multiple:false
    ack:deliveryTag:34，multiple:true

可以看到，发送50条消息，收到的ack个数不一样多次运行程序会发现每次发送回来的ack消息中的deliveryTag域的值并不是一样的，说明broker端批量回传给发送者的ack消息并不是以固定的批量大小回传的。

### 性能比较 ###

事务模式性能是最差的，普通confirm模式性能比事务模式稍微好点，但是和批量confirm模式还有异步confirm模式相比，还是小巫见大巫。批量confirm模式的问题在于confirm之后返回false之后进行重发这样会使性能降低，异步confirm模式(async)编程模型较为复杂，至于采用哪种方式具体看实际情况。  
注意：AMQP的事务模式和Confirm模式不能一起使用。