第二章 RabbitMQ安装使用

谁借莪１个温暖的怀抱￠ 2024-04-20 09:02 65阅读 0赞

## **一、RabbitMQ安装和运行** ##

### **1、在Linux中安装RabbitMQ演示** ###

\#\#在Linux中安装RabbitMQ演示，以VirtualBox中的虚拟机为操作系统载体

\#\#操作系统版本为CentOS7，纯净未安装RabbitMQ

\#以root用户安装

su - root

\#\#安装

wget https://packages.erlang-solutions.com/erlang-solutions-1.0-1.noarch.rpm

rpm -Uvh erlang-solutions-1.0-1.noarch.rpm

yum install epel-release

yum install erlang

wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.6/rabbitmq-server-3.6.6-1.el7.noarch.rpm

yum install rabbitmq-server-3.6.6-1.el7.noarch.rpm

\#\#管理

service rabbitmq-server start

service rabbitmq-server status

\#安装管理控制台

rabbitmq-plugins enable rabbitmq\_management

\#重启RabbitMQ

service rabbitmq-server stop

service rabbitmq-server start

\#可在虚拟机浏览器中检查能否访问管理控制台

\#开启相对应的端口

firewall-cmd --permanent --add-port=15672/tcp

firewall-cmd --permanent --add-port=5672/tcp

\#查看已有虚拟主机并增加名为enjoyedu的虚拟主机

rabbitmqctl list\_vhosts

rabbitmqctl add\_vhost chj\_vhost

rabbitmqctl list\_vhosts

\#增加名为mark的用户并配置administrator角色,增加相应的权限

rabbitmqctl add\_user chj 123456

rabbitmqctl set\_permissions -p chj\_vhost chj '.\*' '.\*' '.\*'

rabbitmqctl set\_user\_tags  chj administrator

\#如需在本地机器中访问，需要在虚拟机中配置端口转发

\#在本地浏览器中检查能否访问管理控制台并运行代码测试

\#注意：系统缺省的guest用户是不能进行远程登录的，除非另行配置相关参数

备注：

在 Linux(以 CentOS7 为例)下安装 RabbitMQ

1、wget https://packages.erlang-solutions.com/erlang-solutions-1.0-1.noarch.rpm

2、rpm -Uvh erlang-solutions-1.0-1.noarch.rpm

3、yum install epel-release4、yum install erlang

5 wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.6/rabbitmq-server-3.6.6-1.el7.noarch.rpm

6、yum install rabbitmq-server-3.6.6-1.el7.noarch.rpm

注意：常用命令

rabbitmqctl add\_user <username> <password>

rabbitmqctl delete\_user <username>

修改密码：

rabbitmqctl change\_password <username> <newpassword>

rabbitmqctl clear\_password <username>

添加交换器：

rabbitmqctl add\_vhost chj-vhost

rabbitmqctl delete\_vhost chj-vhost

查看：

rabbitmqctl list\_vhosts list\_vhosts

设置用户权限：

rabbitmqctl set\_permissions \[-p <vhostpath>\] <user> <conf> <write> <read>

例如：rabbitmqctl set\_permissions -p chj\_vhost chj '.\*' '.\*' '.\*'

rabbitmqctl clear\_permissions \[-p <vhostpath>\] <username>

查看权限：

rabbitmqctl list\_permissions -p chj\_vhost

rabbitmqctl list\_user\_permissions <username>

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70][]

### **2、在Linux中卸载RabbitMQ** ###

/sbin/service rabbitmq-server stop

yum list | grep rabbitmq

yum -y remove rabbitmq-server.noarch

yum list | grep erlang

yum -y remove erlang-\*

yum remove erlang.x86\_64

rm -rf /usr/lib64/erlang

rm -rf /var/lib/rabbitmq

yum remove epel-release

rm -rf /var/cache/yum/x86\_64/6/epel/

### **3、RabbitMQ 常用端口**  ###

client 端通信端口： 5672

管理端口 ： 15672

server 间内部通信端口： 25672

### **4、可能的问题**  ###

如端口出现不能访问，考虑是否防火墙问题，可以使用形如以下命令开启或直接关闭防火墙：

firewall-cmd --add-port=15672/tcp --permanent

运行 rabbitmqctl status出现Error: unable to connect to node rabbit@controller: nodedown之类问题考虑如下几种解决办法：

1、重启服务

service rabbitmq-server stop

service rabbitmq-server start

2、检查/var/lib/rabbitmq中是否存在.erlang.cookie，没有则新建一个，里面随便输入一段字符串

3、重新安装服务

4、百度或者Google一下

## **二、原生 Java 客户端进行消息通信** ##

### **1、Direct交换器**  ###

参见代码 native 模块包 com.chj.exchange.direct 中：

DirectProducer：direct 类型交换器的生产者

NormalConsumer：普通的消费者

MulitBindConsumer：队列绑定到交换器上时，是允许绑定多个路由键的，也就是多重绑定

MulitChannelConsumer：一个连接下允许有多个信道

MulitConsumerOneQueue：一个队列多个消费者，则会表现出消息在消费者之间的轮询发送。

**1.1、生产者和消费者一般用法**

DirectProducer：direct 类型交换器的生产者。

NormalConsumer：普通的消费者。

使用 DirectProducer作为生产者，NormalConsumer作为消费者，消费者绑定一个队列。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 1][]

**1.2、队列和交换器的多重绑定**

DirectProducer：direct 类型交换器的生产者

final Channel channel = connection.createChannel();
    // 在信道中设置交换器
    channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
    //申明队列（放在消费者中去做）--申明路由键\消息体
    String[] routeKeys ={"hankin01","hankin02","hankin03"};
    for (int i=0;i<6;i++) {
        String routeKey = routeKeys[i % 3];
        String msg = "Hello,RabbitMQ" + (i + 1);
        channel.basicPublish(EXCHANGE_NAME,routeKey,null,msg.getBytes());
        System.out.println("Send message:"+routeKey+":"+msg);
    }

NormalConsumer：普通的消费者

final Channel channel = connection.createChannel();
        // 在信道中设置交换器
        channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        // 申明队列（放在消费者中去做）
        String queueName = "queue-hankin01";
        channel.queueDeclare(queueName, false, false, false, null);
        //绑定：将队列(queuq-king)与交换器通过 路由键 绑定(king)
        String routeKey ="hankin01";
        channel.queueBind(queueName,DirectProducer.EXCHANGE_NAME,routeKey);
        System.out.println("waiting for message ......");
        // 申明一个消费者
        final DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String message = new String(body,"UTF-8");
                System.out.println("Received["+envelope.getRoutingKey()+"]"+message);
            }
        };
        // 消息者正是开始在指定队列上消费。(queue-hankin01)
        // TODO 这里第二个参数是自动确认参数，如果是true则是自动确认
        channel.basicConsume(queueName,true,consumer);
    }

MultiBindConsumer：队列绑定到交换器上时，是允许绑定多个路由键的，也就是多重绑定

final Channel channel = connection.createChannel();
        // 在信道中设置交换器
        channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        // 声明一个随机队列
        String queueName = channel.queueDeclare().getQueue();
        // TODO：队列绑定到交换器上时，是允许绑定多个路由键的，也就是多重绑定
        String[] routeKeys ={"hankin01","hankin02","hankin03"};
        for(String routeKey : routeKeys){
            channel.queueBind(queueName, DirectProducer.EXCHANGE_NAME, routeKey);
        }
        System.out.println(" [*] Waiting for messages:");
        // 创建队列消费者
        final DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String message = new String(body,"UTF-8");
                System.out.println("Received["+envelope.getRoutingKey()+"]"+message);
            }
        };
        // TODO 消息者正是开始在指定队列上消费。这里第二个参数是自动确认参数，如果是true则是自动确认
        channel.basicConsume(queueName,true,consumer);
    }

对比：单个绑定的消费者只能收到指定的消息，多重绑定的的消费者可以收到所有的消息。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 2][]

**1.3、一个连接多个信道**

DirectProducer：direct 类型交换器的生产者

MultiChannelConsumer：一个连接下允许有多个信道

public class MultiChannelConsumer {
        private static class ConsumerWorker implements Runnable {
            final Connection connection;
            public ConsumerWorker(Connection connection) {
                this.connection = connection;
            }
            public void run() {
                try {
                    final Channel channel = connection.createChannel();
                    // 在信道中设置交换器
                    channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
                    // 声明一个随机队列
                    String queueName = channel.queueDeclare().getQueue();
                    // TODO：队列绑定到交换器上时，是允许绑定多个路由键的，也就是多重绑定
                    String[] routeKeys ={"hankin01","hankin02","hankin03"};
                    for(String routeKey : routeKeys){
                        channel.queueBind(queueName, DirectProducer.EXCHANGE_NAME, routeKey);
                    }
                    System.out.println(" [*] Waiting for messages:");
                    // 创建队列消费者
                    final DefaultConsumer consumer = new DefaultConsumer(channel) {
                        @Override
                        public void handleDelivery(String consumerTag, Envelope envelope,
                                                   AMQP.BasicProperties properties,byte[] body) throws IOException {
                            String message = new String(body,"UTF-8");
                            System.out.println("Received["+envelope.getRoutingKey()+"]"+message+", channel=="+channel.getChannelNumber());
                        }
                    };
                    // TODO 消息者正是开始在指定队列上消费。这里第二个参数是自动确认参数，如果是true则是自动确认
                    channel.basicConsume(queueName,true,consumer);
                }catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
        public static void main(String[] argv) throws IOException,InterruptedException, TimeoutException {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("10.200.169.5");
            Connection connection = connectionFactory.newConnection();
            //一个连接多个信道
            for(int i=0;i<2;i++){// 将连接作为参数，传递给每个线程
                Thread worker = new Thread(new ConsumerWorker(connection));
                worker.start();
            }
        }
    }

一个连接，我们可以使用多线程的方式模拟多个信道进行通讯。这样可以做到多路复用。

**1.4、一个队列多个消费者**

MultiConsumerOneQueue：一个队列多个消费者，则会表现出消息在消费者之间的轮询发送。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 3][]

/**
     * 一个队列多个消费者，则会表现出消息在消费者之间的轮询发送。
     */
    public class MultiConsumerOneQueue {
        private static class ConsumerWorker implements Runnable {
            final Connection connection;
            final String queueName;
            public ConsumerWorker(Connection connection,String queueName) {
                this.connection = connection;
                this.queueName = queueName;
            }
            public void run() {
                try {
                    final Channel channel = connection.createChannel();
                    channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
                    // TODO：声明一个队列,rabbitmq，如果队列已存在，不会重复创建
                    channel.queueDeclare(queueName, false,false, false,null);
                    // TODO：队列绑定到交换器上时，是允许绑定多个路由键的，也就是多重绑定
                    String[] routeKeys ={"hankin01","hankin02","hankin03"};
                    for(String routeKey : routeKeys){
                        channel.queueBind(queueName, DirectProducer.EXCHANGE_NAME, routeKey);
                    }
                    System.out.println(" [*] Waiting for messages:");
                    // 创建队列消费者
                    final DefaultConsumer consumer = new DefaultConsumer(channel) {
                        @Override
                        public void handleDelivery(String consumerTag, Envelope envelope,
                                                   AMQP.BasicProperties properties,byte[] body) throws IOException {
                            String message = new String(body,"UTF-8");
                            System.out.println("Received["+envelope.getRoutingKey()+"]"+message;
                        }
                    };
                    // TODO 消息者正是开始在指定队列上消费。这里第二个参数是自动确认参数，如果是true则是自动确认
                    channel.basicConsume(queueName,true,consumer);
                }catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
        public static void main(String[] argv) throws IOException,InterruptedException, TimeoutException {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("10.200.169.5");
            Connection connection = connectionFactory.newConnection();
            // 3个线程，线程之间共享队列,一个队列多个消费者
            String queueName = "focusAll";
            for(int i=0;i<3;i++){/*将队列名作为参数，传递给每个线程*/
                Thread worker =new Thread(new ConsumerWorker(connection,queueName));
                worker.start();
            }
        }
    }

### **2、Fanout交换器** ###

消息广播到绑定的队列

参见代码native模块包com.chj.exchange.fanout中：

通过测试表明，不管我们如何调整生产者和消费者的路由键，都对消息的接受没有影响。

**示例代码：**

**fanout生产者：**

// 创建一个连接
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            // TODO 指定转发 FANOUT 交换器
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
    //        String queueName = "producer_create";
    //        channel.queueDeclare(queueName,false,false,false,null);
    //        channel.queueBind(queueName,EXCHANGE_NAME,"test");
            // 路由键设置
            String[] routekeys = {"chj01","chj02","chj03"};
            for(int i=0;i<3;i++){
                String routekey = routekeys[i%3];//每一次发送一条消息
                String message = "Hello World_"+(i+1);
                //参数1：exchange name 参数2：routing key
                channel.basicPublish(EXCHANGE_NAME, routekey,null, message.getBytes());
                System.out.println(" [x] Sent '" + routekey +"':'" + message + "'");
            }
            // 关闭频道和连接
            channel.close();
            connection.close();

**fanout消费者--绑定多个路由键：**

final Channel channel = connection.createChannel();
        // TODO 指定转发 FANOUT 交换器
        channel.exchangeDeclare(FanoutProducer.EXCHANGE_NAME,BuiltinExchangeType.FANOUT);
        // 声明一个随机队列
        String queueName = channel.queueDeclare().getQueue();
        // 队列绑定到交换器上时，是允许绑定多个路由键的，也就是多重绑定
        String[] routekeys = {"chj01","chj02","chj03"};
        for(String routekey : routekeys){
            channel.queueBind(queueName,FanoutProducer.EXCHANGE_NAME,routekey);
        }
        System.out.println(" ["+queueName+"] Waiting for messages:");
        // 创建队列消费者
        final Consumer consumerA = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,Envelope envelope,
                                       AMQP.BasicProperties properties,byte[] body)throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("Received "  + envelope.getRoutingKey() + "':'" + message + "'");
            }
        };
        channel.basicConsume(queueName, true, consumerA);// 消息确认
    }

**fanout消费者--绑定一个不存在的路由键：**

final Channel channel = connection.createChannel();
        // TODO 指定转发 FANOUT 交换器
        channel.exchangeDeclare(FanoutProducer.EXCHANGE_NAME,BuiltinExchangeType.FANOUT);
        // 声明一个随机队列
        String queueName = channel.queueDeclare().getQueue();
        // TODO 设置一个不存在的路由键
        String routekey="xxx";
        channel.queueBind(queueName, FanoutProducer.EXCHANGE_NAME, routekey);
        System.out.println(" ["+queueName+"] Waiting for messages:");
        // 创建队列消费者
        final Consumer consumerA = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,
                                       byte[] body)throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("Received "  + envelope.getRoutingKey() + "':'" + message + "'");
            }
        };
        channel.basicConsume(queueName, true, consumerA);
    }

### **3、Topic交换器** ###

参见代码native模块包com.chj.exchange.topic中：通过使用“\*”和“\#”，使来自不同源头的消息到达同一个队列，”.”将路由键分为了几个标识符，“\*”匹配 1 个，“\#”匹配一个或多个。

例子如下：

假设有交换器：topic\_course，

讲课老师有：chj01,chj02,chj03，

技术专题有：kafka,jvm,redis，

课程章节有：A、B、C，

路由键的规则为 讲课老师+“.”+技术专题+“.”+课程章节，如：chj01.kafka.A。

**3.1、要关注所有的课程，怎么做？**

声明一个队列并绑定到交换器上：

*//****TODO  \#  绑定所有队列***  
channel.queueBind(queueName,TopicProducer.***EXCHANGE\_NAME***, **"\#"**);

**3.2、关注chj01的的所有课程，怎么办？**

声明一个队列并绑定到交换器上：

*//* ***TODO 关注chj01的所有课程***  
channel.queueBind(queueName,TopicProducer.***EXCHANGE\_NAME***, **"chj01.\#"**);

注意：

如果这里改为 chj01.\* 的话，则不会出现任何信息，因为 “\*”匹配 1 个（使用.分割的标识的个数）。

**3.3、关注chj01的所有的A章节，怎么办？**

声明一个队列并绑定到交换器上，或者声明一个队列并绑定到交换器上：

*//* ***TODO 关注chj01的所有的A章节***  
channel.queueBind(queueName,TopicProducer.***EXCHANGE\_NAME***, **"chj01.\*.A"**);  
channel.queueBind(queueName,TopicProducer.***EXCHANGE\_NAME***, **"chj01.\#.A"**);

**3.4、关注kafka所有的课程，怎么办？**

声明一个队列并绑定到交换器上：

*//* ***TODO 关注所有的 kafka 课程***  
channel.queueBind(queueName,TopicProducer.***EXCHANGE\_NAME***,**"\#.kafka.\#"**);

**3.5、关注所有的B章节，怎么办？**

声明一个队列并绑定到交换器上：

*//* ***TODO 关注所有的 B 章节***  
channel.queueBind(queueName,TopicProducer.***EXCHANGE\_NAME***, **"\#.B"**);

**3.6、关注chj01的kafka的A章节，怎么办？**

声明一个队列并绑定到交换器上：

*//* ***TODO 关注chj01的kafka的A章节***  
channel.queueBind(queueName,TopicProducer.***EXCHANGE\_NAME***, **"chj01.kafka.A"**);

## **三、消息发布时的权衡**  ##

在RabbitMQ在设计的时候，特意让生产者和消费者“脱钩”，也就是消息的发布和消息的消费之间是解耦的。在RabbitMQ中，有不同的投递机制（生产者），但是每一种机制都对性能有一定的影响。一般来讲速度快的可靠性低，可靠性好的性能差，具体怎么使用需要根据你的应用程序来定，所以说没有最好的方式，只有最合适的方式。只有把你的项目和技术相结合，才能找到适合你的平衡。

![20190920161753285.png][]

### **1、无保障** ###

在演示各种交换器中使用的就是无保障的方式，通过basicPublish发布你的消息并使用正确的交换器和路由信息，你的消息会被接收并发送到合适的队列中。但是如果有网络问题，或者消息不可路由，或者RabbitMQ自身有问题的话，这种方式就有风险。所以无保证的消息发送一般情况下不推荐。

### **2、失败确认** ###

在发送消息时设置mandatory标志，告诉RabbitMQ，如果消息不可路由，应该将消息返回给发送者，并通知失败。可以这样认为，开启mandatory是开启故障检测模式。

注意：它只会让RabbitMQ向你通知失败，而不会通知成功。如果消息正确路由到队列，则发布者不会受到任何通知。带来的问题是无法确保发布消息一定是成功的，因为通知失败的消息可能会丢失。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 4][]

channel.addConfirmListener则用来监听RabbitMQ发回的信息。

如何使用，参见代码native模块包com.chj.producer\_balance.mandatory中。

**监听器的小甜点**

在信道关闭和连接关闭时，还有两个监听器可以使用：

*//连接关闭时执行*  
connection.addShutdownListener(**new** ShutdownListener() \{   
    **public void** shutdownCompleted(ShutdownSignalException cause) \{   
        System.***out***.println(**"connection was shutdown:"**\+cause.getMessage()+ System.*currentTimeMillis*());  
    \}  
\});  
*//信道关闭时执行*  
channel.addShutdownListener(**new** ShutdownListener() \{   
    **public void** shutdownCompleted(ShutdownSignalException cause) \{   
        System.***out***.println(**"channel was shutdown:"**\+cause.getMessage()+ System.*currentTimeMillis*());  
    \}  
\});

### **3、MQ事务** ###

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

1）channel.txSelect()声明启动事务模式；

2）channel.txComment()提交事务；

3）channel.txRollback()回滚事务；

在发送消息之前，需要声明channel为事务模式，提交或者回滚事务即可；开启事务后，客户端和 RabbitMQ 之间的通讯交互流程：

*  客户端发送给服务器Tx.Select(开启事务模式)
 *  服务器端返回Tx.Select-Ok（开启事务模式ok）
 *  推送消息
 *  客户端发送给事务提交Tx.Commit
 *  服务器端返回Tx.Commit-Ok

以上就完成了事务的交互流程，如果其中任意一个环节出现问题，就会抛出 IoException 移除，这样用户就可以拦截异常进行事务回滚，或决定要不要重复消息。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 5][]

那么，既然已经有事务了，为何还要使用发送方确认模式呢，原因是因为事务的性能是非常差的。根据相关资料，事务会降低 2~10 倍的性能。

如何使用，参见代码native模块包com.chj.producer\_balance.transaction中。

**生产者——事务模式：**

public class ProducerTransaction {
        public final static String EXCHANGE_NAME = "producer_transaction";
        public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
            // 创建连接连接到RabbitMQ
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");// 设置MabbitMQ所在主机ip或者主机名
            Connection connection = factory.newConnection();// 创建一个连接
            Channel channel = connection.createChannel(); // 创建一个信道
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);// 指定转发
            String[] routekeys={"king","mark","james"};
            //加入事务
            channel.txSelect();
            try {
                for(int i=0;i<3;i++){
                    String routekey = routekeys[i%3];
                    String message = "Hello World_"+(i+1) +("_"+System.currentTimeMillis());
                    channel.basicPublish(EXCHANGE_NAME, routekey, true,null, message.getBytes());
                    System.out.println("----------------------------------");
                    System.out.println(" Sent Message: [" + routekey +"]:'" + message + "'");
                    Thread.sleep(200);
                }
                //事务提交
                channel.txCommit();
            } catch (IOException e) {
                e.printStackTrace();
                //事务回滚
                channel.txRollback();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            channel.close();
            connection.close();
        }
    }

**消费者——事务模式：**

public class ConsumerProducerTransaction {
        public static void main(String[] argv) throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            // 打开连接和创建频道，与发送端一样
            Connection connection = factory.newConnection();
            final Channel channel = connection.createChannel();
            channel.exchangeDeclare(ProducerTransaction.EXCHANGE_NAME,BuiltinExchangeType.DIRECT);
            String queueName = "producer_confirm";
            channel.queueDeclare(queueName,false,false,false,null);
            //只关注king的日志
            String routekey="king";
            channel.queueBind(queueName, ProducerTransaction.EXCHANGE_NAME,routekey);
            System.out.println(" [*] Waiting for messages......");
            // 创建队列消费者
            final Consumer consumerB = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag,Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");
                    System.out.println( "Received ["+ envelope.getRoutingKey()+ "] "+message);
                }
            };
            channel.basicConsume(queueName, true, consumerB);
        }
    }

### **4、发送方确认模式**  ###

基于事务的性能问题，RabbitMQ团队为我们拿出了更好的方案，即采用发送方确认模式，该模式比事务更轻量，性能影响几乎可以忽略不计。

原理：生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都将会被指派一个唯一的 ID(从1开始)，由这个id在生产者和RabbitMQ之间进行消息的确认。

**不可路由的消息**，当交换器发现消息不能路由到任何队列，会进行确认操作，表示收到了消息。如果发送方设置了mandatory模式，则会先调用addReturnListener监听器。

**可路由的消息**，要等到消息被投递到所有匹配的队列之后，broker会发送一个确认给生产者(包含消息的唯一ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker回传给生产者的确认消息中delivery-tag域包含了确认消息的序列号。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 6][]

confirm模式最大的好处在于他可以是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息决定下一步的处理。

**Confirm** **的三种实现方式：**

方式一：channel.waitForConfirms()普通发送方确认模式；消息到达交换器，就会返回true。

// TODO 启用发送者确认模式
    channel.confirmSelect();
    // 所有日志严重性级别
    for(int i=0;i<2;i++) {
        String message = "Hello World_" + (i + 1); // 发送的消息
        //参数1：exchange name  参数2：routing key
        channel.basicPublish(EXCHANGE_NAME,ROUTE_KEY,true,null,message.getBytes());
        System.out.println(" Sent Message: [" + ROUTE_KEY +"]:'"+ message + "'");
        // TODO 确认是否成功
        if(channel.waitForConfirms()){
            System.out.println("send success");
        }else{
            System.out.println("send failure");
        }
    }

方式二：channel.waitForConfirmsOrDie()批量确认模式；使用同步方式等所有的消息发送之后才会执行后面代码，只要有一个消息未到达交换器就会抛出IOException异常。

// TODO 启用发送者确认模式
    channel.confirmSelect();
    for(int i=0;i<2;i++) {
        String message = "Hello World_" + (i + 1); // 发送的消息
        //参数1：exchange name  参数2：routing key
        channel.basicPublish(EXCHANGE_NAME,ROUTE_KEY,true,null,message.getBytes());
        System.out.println(" Sent Message: [" + ROUTE_KEY +"]:'"+ message + "'");
    }
    // TODO 启用发送者确认模式（批量确认）
    channel.waitForConfirmsOrDie();
    // 关闭频道和连接
    channel.close();
    connection.close();

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

// 指定转发
    channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
    // TODO 启用发送者确认模式
    channel.confirmSelect();
    // TODO 添加发送者确认监听器
    channel.addConfirmListener(new ConfirmListener(){
    public void handleAck(long deliveryTag, boolean multiple) throws IOException {
        System.out.println("send_handleAck:"+deliveryTag+",multiple:"+multiple);
    }
    public void handleNack(long deliveryTag, boolean multiple) throws IOException {
         System.out.println("send_handleNack:"+deliveryTag+",multiple:"+multiple);
       }
    });
    // TODO 添加失败者通知
    channel.addReturnListener(new ReturnListener() {
    public void handleReturn(int replyCode, String replyText, String exchange, String routingKey,AMQP.BasicProperties properties, byte[] body) throws IOException {
        String message = new String(body);
             System.out.println("RabbitMq路由失败:  "+routingKey+"."+message);
        }
    });
    String[] routekeys={"hankin01","hankin02"};
    for(int i=0;i<6;i++) {
        String routekey = routekeys[i%2];
       // 发送的消息
       String message = "Hello World_"+(i+1)+("_"+System.currentTimeMillis());
       channel.basicPublish(EXCHANGE_NAME,routekey,true,MessageProperties.PERSISTENT_BASIC,message.getBytes());
       System.out.println(" Sent Message: [" + routekey +"]:'"+ message + "'");
     }

如何使用，参见代码native模块包 com.chj.producer\_balance.producerconfirm中。

### **5、备用交换器**  ###

在第一次声明交换器时被指定，用来提供一种预先存在的交换器，如果主交换器无法路由消息，那么消息将被路由到这个新的备用交换器。

如果发布消息时同时设置了mandatory会发生什么？如果主交换器无法路由消息，RabbitMQ并不会通知发布者，因为，向备用交换器发送消息，表示消息已经被路由了。

注意，新的备用交换器就是普通的交换器，没有任何特殊的地方。

使用备用交换器，向往常一样，声明Queue和备用交换器，把Queue绑定到备用交换器上。然后在声明主交换器时，通过交换器的参数，alternate-exchange将备用交换器设置给主交换器。 建议备用交换器设置为faout类型，Queue绑定时的路由键设置为“\#”。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 7][]

如何使用，参见代码native模块包com.chj.producer\_balance.backupexchange中。

生产者--绑定了一个备用交换器：

public class BackupExProducer {
        public final static String EXCHANGE_NAME = "main-exchange";
        public final static String BAK_EXCHANGE_NAME = "backup-exchange";
        public static void main(String[] argv)throws IOException, TimeoutException {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("10.200.169.5");
            // 打开连接和创建频道，与发送端一样
            Connection connection = connectionFactory.newConnection();
            final Channel channel = connection.createChannel();
            //TODO 声明备用交换器
            Map<String,Object> argsMap = new HashMap<String,Object>();
            argsMap.put("alternate-exchange",BAK_EXCHANGE_NAME);
            // 主交换器
            channel.exchangeDeclare(EXCHANGE_NAME,"direct", false,false, argsMap);
            // 备用交换器
            channel.exchangeDeclare(BAK_EXCHANGE_NAME, BuiltinExchangeType.FANOUT,true,false,null);
            // 定义路由键
            String[] routekeys = {"hankin01","hankin02","hankin03"};
            for(int i=0;i<3;i++) {
                String routekey = routekeys[i % 3];
                String msg = "Hello, RabbitMQ" + (i + 1);
                //发布消息，需要参数：交换器，路由键，其中以日志消息级别为路由键
                channel.basicPublish(EXCHANGE_NAME,routekey,null,msg.getBytes());
                System.out.println("Send "+routekey+":"+msg);
            }
            channel.close();
            connection.close();
        }
    }

消费者——绑定备用交换器队列的消费者：

public class BackupExConsumer {
        public static void main(String[] argv)throws IOException, TimeoutException {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("10.200.169.5");
            // 打开连接和创建频道，与发送端一样
            Connection connection = connectionFactory.newConnection();
            final Channel channel = connection.createChannel();
            channel.exchangeDeclare(BackupExProducer.BAK_EXCHANGE_NAME, BuiltinExchangeType.FANOUT, true, false, null);
            // 声明一个队列
            String queueName = "fetchother";
            channel.queueDeclare(queueName,false,false,false,null);
            channel.exchangeBind(queueName,BackupExProducer.BAK_EXCHANGE_NAME,"#");
            System.out.println(" [*] Waiting for messages......");
            Consumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag,Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");
                    System.out.println("Received["+envelope.getRoutingKey() +"]"+message);
                }
            };
            // 消费者正式开始在指定队列上消费消息
            channel.basicConsume(queueName,true,consumer);
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/60df241e53bb48aaae4ddefac6a78158.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/15e0ece495324b6a8ce0f2388e944005.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/b784976c324b4d68bda4657cdd3fb294.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/3dcd9c90164643589291928df44d704f.png
[20190920161753285.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/305ec6c42c6749e99ffd3635c90d9c5d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/27376267e6b34190aea6c3dae7fed374.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 5]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/d8e393a747f7487a993d293ad65129c7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 6]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/30982e861cee4043ab4d695b044e10b8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 7]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/b111ee04fec94fc0967674d37f25ac30.png