RabbitMQ——使用事务控制消息的发送和接收

Posted 张起灵-小哥

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RabbitMQ——使用事务控制消息的发送和接收相关的知识，希望对你有一定的参考价值。

文章目录：

1.写在前面

2.AMQP中的事务案例

2.1 编写消息发送类

2.2 编写消息接收类

2.3 测试结果

3.发送者/接收者确认模式中的事务案例

3.1 编写消息接收类

3.2 channel.waitForConfirms()普通发送方确认模式

3.2.1 编写消息发送类

3.2.2 测试结果

3.3 channel.waitForConfirmsOrDie()批量确认模式

3.3.1 编写消息发送类 

3.3.2 测试结果

3.4 channel.addConfirmListener()异步监听发送方确认模式

3.4.1 编写消息发送类

3.4.2 测试结果

1.写在前面

事务消息与数据库的事务类似，只是MQ中的消息是要保证消息是否会全部发送成功，防止丢失消息的一种策略。

RabbitMQ有两种方式来解决这个问题：

  1. 通过AMQP提供的事务机制实现；
  2. 使用发送者确认模式实现；

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

  1. channel.txSelect()声明启动事务模式；
  2. channel.txCommint()提交事务；
  3. channel.txRollback()回滚事务

2.AMQP中的事务案例

2.1 编写消息发送类

package com.szh.rabbitmq.transaction;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 *
 */
public class Send {
    public static void main(String[] args) {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.40.130");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("root");

        Connection connection=null;
        Channel channel=null;

        try {
            connection=factory.newConnection();
            channel=connection.createChannel();

            channel.queueDeclare("transactionQueue",true,false,false,null);
            channel.exchangeDeclare("directTransactionExchange","direct",true);
            channel.queueBind("transactionQueue","directTransactionExchange","transactionRoutingKey");

            String message="RabbitMQ中的事务消息绑定";

            //开启一个事务，启动事务以后，所有写入到消息队列中的消息
            //必须显示的调用 txCommit() 提交事务 或 txRollback() 回滚事务，否则消息不会真正的写入消息队列中
            channel.txSelect();
            channel.basicPublish("directTransactionExchange","transactionRoutingKey",null,message.getBytes(StandardCharsets.UTF_8));
            //提交事务，提交之后会将内存中的消息写入队列，并释放内存
            channel.txCommit();

            System.out.println("消息发送成功：" + message);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } finally {
            if (channel != null) {
                try {
                    //回滚事务，放弃当前事务中所有还未提交的消息，释放内存
                    channel.txRollback();
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null ) {
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

2.2 编写消息接收类

package com.szh.rabbitmq.transaction;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 *
 */
public class Receive {
    public static void main(String[] args) {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.40.130");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("root");

        Connection connection=null;
        Channel channel=null;

        try {
            connection=factory.newConnection();
            channel=connection.createChannel();

            channel.queueDeclare("transactionQueue",true,false,false,null);
            channel.exchangeDeclare("directTransactionExchange","direct",true);
            channel.queueBind("transactionQueue","directTransactionExchange","transactionRoutingKey");
            /**
             * 开启事务，即使不进行事务的提交，那么也可以获取队列中的消息，并且将消息从队列中取出
             * 注意：这种情况只是暂时的！！！
             *
             * 如果开启了事务，并且消息还是手动确认的，那么必须在最后提交事务
             * 否则即使调用了确认方法，消息也不会从队列中移除
             */
            channel.txSelect();
            channel.basicConsume("transactionQueue",true,"",new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message=new String(body);
                    System.out.println("消息接收成功：" + message);
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

2.3 测试结果

这里因为交换机类型是direct，它是一对一的模式，所以这里我们先启动消息发送者、再启动消息接收者。

3.发送者/接收者确认模式中的事务案例

这种发送者确认模式也是RabbitMQ中实现事务的一种方法，它其中主要有三种实现方式：

  1. channel.waitForConfirms()普通发送方确认模式。
  2. channel.waitForConfirmsOrDie()批量确认模式。
  3. channel.addConfirmListener()异步监听发送方确认模式。

为了保证消息从队列可靠地到达消费者，RabbitMQ提供消息确认机制(message acknowledgment)。消费者在声明队列时，可以指定noAck参数，当noAck=false时，RabbitMQ会等待消费者显式发回ack信号后才从内存(和磁盘，如果是持久化消息的话)中移去消息。否则，RabbitMQ会在队列中消息被消费后立即删除它。

在Consumer中Confirm模式中分为手动确认和自动确认。

手动确认主要并使用以下方法：

basicAck(): 用于肯定确认，multiple参数用于多个消息确认。

basicRecover()：是路由不成功的消息可以使用recovery重新发送到队列中。

basicReject()：是接收端告诉服务器这个消息我拒绝接收,不处理,可以设置是否放回到队列中还是丢掉，而且只能一次拒绝一个消息,官网中有明确说明不能批量拒绝消息，为解决批量拒绝消息才有了basicNack。

basicNack()：可以一次拒绝N条消息，客户端可以设置basicNack方法的multiple参数为true。

3.1 编写消息接收类

这三种发送者确认模式对应的消息接收类都是一样的，所以我在这里先给出消息接收类的源码。

package com.szh.rabbitmq.confirm;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 *
 */
public class Receive {
    public static void main(String[] args) {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.40.130");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("root");

        Connection connection=null;
        Channel channel=null;

        try {
            connection=factory.newConnection();
            channel=connection.createChannel();

            channel.queueDeclare("confirmQueue",true,false,false,null);
            channel.exchangeDeclare("confirmExchange","direct",true);
            channel.queueBind("confirmQueue","confirmExchange","confirmRoutingKey");
            /**
             * 参数2为消息确认机制，true表示自动消息确认，确认以后消息会从队列中被移除，当读取完消息之后会自动确认
             *                  false表示手动消息确认
             */
            channel.basicConsume("confirmQueue",false,"",new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message=new String(body);
                    System.out.println("消息接收成功：" + message);
                    //获取消息的编号，我们需要根据消息的编号来确认消息
                    long tag=envelope.getDeliveryTag();
                    //获取当前内部类中的通道
                    Channel c=this.getChannel();
                    //手动确认消息，确认以后表示当前消息已经成功处理了，需要从队列中移除
                    //参数1为消息的编号，参数2为是否确认多个消息。true表示确认小于等于当前编号的所有消息；false表示确认当前消息
                    c.basicAck(tag,true);
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

3.2 channel.waitForConfirms()普通发送方确认模式

3.2.1 编写消息发送类

package com.szh.rabbitmq.confirm.waitForConfirms;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 *
 */
public class Send {
    public static void main(String[] args) {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.40.130");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("root");

        Connection connection=null;
        Channel channel=null;

        try {
            connection=factory.newConnection();
            channel=connection.createChannel();

            channel.queueDeclare("confirmQueue",true,false,false,null);
            channel.exchangeDeclare("confirmExchange","direct",true);
            channel.queueBind("confirmQueue","confirmExchange","confirmRoutingKey");

            String message="RabbitMQ中的普通发送者确认模式";

            //启动发送者确认模式
            channel.confirmSelect();
            channel.basicPublish("confirmExchange","confirmRoutingKey",null,message.getBytes(StandardCharsets.UTF_8));
            //消息发送成功返回true；否则返回false
            //该方法有一个重载方法：boolean waitForConfirms(long var1)，用于指定一个毫秒来指定超时时间
            //如果超时，则会抛出InterruptedException异常
            boolean flag=channel.waitForConfirms();

            System.out.println("消息发送成功：" + message + " ---> " + flag);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            if (channel != null) {
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null ) {
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

3.2.2 测试结果

3.3 channel.waitForConfirmsOrDie()批量确认模式

3.3.1 编写消息发送类 

package com.szh.rabbitmq.confirm.waitForConfirmsOrDie;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 *
 */
public class Send {
    public static void main(String[] args) {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.40.130");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("root");

        Connection connection=null;
        Channel channel=null;

        try {
            connection=factory.newConnection();
            channel=connection.createChannel();

            channel.queueDeclare("confirmQueue",true,false,false,null);
            channel.exchangeDeclare("confirmExchange","direct",true);
            channel.queueBind("confirmQueue","confirmExchange","confirmRoutingKey");

            String message="RabbitMQ中的普通发送者确认模式";

            //启动发送者确认模式
            channel.confirmSelect();
            channel.basicPublish("confirmExchange","confirmRoutingKey",null,message.getBytes(StandardCharsets.UTF_8));
            /**
             * waitForConfirmsOrDie() 批量消息确认它会同时向服务中发送的所有消息是否已经全部发送成功
             * 这个方法没有任何的返回值，如果其中有一条消息没有发送成功或者服务器出现问题，则被认定为消息发送失败，所以此时需要进行所有消息的补发
             * 该方法有一个重载方法，可以用来指定一个毫秒级的超时时间
             */
            channel.waitForConfirmsOrDie();

            System.out.println("消息发送成功：" + message);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            if (channel != null) {
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null ) {
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

3.3.2 测试结果

3.4 channel.addConfirmListener()异步监听发送方确认模式

3.4.1 编写消息发送类

package com.szh.rabbitmq.confirm.addConfirmListener;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 *
 */
public class Send {
    public static void main(String[] args) {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.40.130");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("root");

        Connection connection=null;
        Channel channel=null;

        try {
            connection=factory.newConnection();
            channel=connection.createChannel();

            channel.queueDeclare("confirmQueue",true,false,false,null);
            channel.exchangeDeclare("confirmExchange","direct",true);
            channel.queueBind("confirmQueue","confirmExchange","confirmRoutingKey");

            String message="RabbitMQ中的普通发送者确认模式";

            //启动发送者确认模式
            channel.confirmSelect();
            channel.addConfirmListener(new ConfirmListener() {
                /**
                 * 消息确认以后的回调方法
                 * @param l ：被确认消息的编号，从1开始自动递增，用于标记当前是第几个消息
                 * @param b ：当前消息是否同时确认了多个
                 * @throws IOException
                 * 如果参数b为true，则表示本次确认同时确认了多条消息，确认的消息数量为 (上一次消息确认的编号,本次消息确认的编号]
                 * 如果参数b为false，则表示只确认了当前编号的消息
                 */
                @Override
                public void handleAck(long l, boolean b) throws IOException {
                    System.out.println("消息被确认了 ---> 消息编号：" + l + " ---> 是否确认了多条：" + b);
                }

                /**
                 * 消息没有确认的回调方法
                 * @param l ：没有被确认消息的编号，从1开始自动递增，用于标记当前是第几个消息
                 * @param b ：当前消息是否同时没有确认了多个
                 * @throws IOException
                 * 如果参数b为true，则表示小于当前编号的所有消息可能都没有发送成功，需要对这些消息进行补发
                 * 如果参数b为false，则表示当前编号的消息没有发送成功，需要对当前这条消息进行补发
                 */
                @Override
                public void handleNack(long l, boolean b) throws IOException {
                    System.out.println("消息没有被确认了 ---> 消息编号：" + l + " ---> 是否没有确认了多条：" + b);
                }
            });
            for (int i = 0; i < 1000; i++) {
                channel.basicPublish("confirmExchange","confirmRoutingKey",null,message.getBytes(StandardCharsets.UTF_8));
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } finally {
            if (channel != null) {
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null ) {
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

3.4.2 测试结果

以上是关于RabbitMQ——使用事务控制消息的发送和接收的主要内容，如果未能解决你的问题，请参考以下文章

RabbitMQ - RabbitMQ交换机；事务；发送接收消息

RabbitMQ

RabbitMQ事务和Confirm发送方消息确认

RabbitMQ事务和Confirm发送方消息确认——深入解读

rabbitmq能控制消息队列发送数量吗

消息队列的常见问题