[大数据]RabbitMQ重要知识

RabbitMQ

一、RabbitMQ简介

消息中间件

? 消息（Message）是指在应用间传送的数据

? 消息队列中间件（Message Queue Middleware,简称MQ）是指利用高效可靠的消息传递机制进行与平台无关的数据交流，并基于数据通信来进行分布式系统的集成

? 消息队列中间件又称为消息中间件，它一般由两种消息传递模式：点对点模式（P2P）和发布/订阅模式（Pub/Sub），消息中间件提供基于存储和转发的应用程序之间的异步数据发送

RabbitMQ

? RabbitMQ是采用Erlang语言实现的AMQP（Advanced Message Queuing Protocol，高级消息队列协议）的消息中间件，具有可靠性，灵活性，支持多种协议的特点

二、RabbitMQ安装

? RabbitMQ是由Erlang语言编写的，因此RabbitMQ的运行需要Erlang语言的支持

安装Erlang

wget https://erlang.org/download/otp_src_24.0.tar.gz
tar -zxvf opt_src_24.0.tar.gz
./configrue --prefix=/opt/erlang
make && make install

ERLANG_HOME=/opt/erlang
export PATH=$PATH:/opt/erlang/sbin
export ERLANG_HOME

source /etc/profile

安装RabbitMQ

wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.9.2/rabbitmq-server-generic-unix-3.9.2.tar.xz
xz -d rabbitmq-server-generic-unix-3.9.2.tar.xz
tar	-xvf rabbitmq-server-generic-unix-3.9.2.tar	//xz解压后的文件不支持gzip，不能加-z参数
mv rabbitmq_server-3.9.3 rabbitmq

export RABBITMQ_HOME=/opt/rabbitmq
export PATH=$PATH:RABBITMQ_HOME/sbin

source /etc/profile

运行测试

rabbitmq-server -detached ：-detached 表示以后台形式运行rabbitmq

rabbitmqctl status ：查看RabbitMQ的运行状态

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-2VcesVst-1630109160145)(images/rabbitmq-status.png)]

• 默认情况下RabbitMQ有一个账号，用户名密码都是"guest" ，但这个账号只能通过本地（localhost）来访问RabbitMQ，远程网络访问受到限制，我们可以为RabbitMQ新增一个账号，并设置其远程访问权限
  • 新增账号
    • 格式：rabbitmqctl add_user username password
    • 案例：rabbitmqctl add_user root root
  • 设置权限
    • 格式：rabbitmqctl set_permissions -p / root ".*" ".*" ".*" ：/表示默认虚拟主机vhost ，root为账号
  • 设置为管理员
    • 格式：rabbitmqctl set_user_tags root administrator
• 启动web管理界面web-management
  • 格式：rabbitmq-plugins enable rabbitmq_management
  • 查看plugins列表：[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-rfawCznV-1630109160147)(images/pluginslist.png)]
  • 通过15672 端口访问web-ui界面[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-oi4QxCke-1630109160149)(images/guestweb.png)]
  • [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-rYpdRLjL-1630109160151)(images/rootweb.png)]
  • 可以看到guest用户只能在loclhost下访问，为root设置权限后即可通过远程网络访问

三、RabbitMQ应用

Java语言

• 在使用RabbitMQ之前需要导入RabbitMQ依赖

  • <!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.13.0</version>
    </dependency>
    
    

  • public class ProduceDemo {
        //定义交换机名称
        private static final String EXCHANGE_NAME = "exchange_demo";
        //定义路由键，用于路由到指定的队列
        private static final String ROUTING_KEY = "routingkey_demo";
        //定义队列名称
        private static final String QUEUE_NAME = "queue_name";
        //Rabbitmq服务器地址
        private static final String IP_ADDRESS = "192.168.189.118";
        //Rabbitmq服务器默认端口号为5672
        private static final int PORT = 5672;
    
        public static void main(String[] args) throws IOException, TimeoutException {
            //创建一个用于连接RabbitMQ的连接工厂，该连接工厂生成的Connection都使用同一账号连接同一IP：PORT的RabbitMQ服务器
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost(IP_ADDRESS);
            connectionFactory.setPort(PORT);
            connectionFactory.setUsername("root");
            connectionFactory.setPassword("root");
            //使用连接工厂创建一个连接
            Connection connection = connectionFactory.newConnection();
            //连接Connection可以定义一个或多个信道Channel
            Channel channel = connection.createChannel();
            //一个信道中要声明要管理的交换机exchange和队列queue
            channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
          	channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            //将交换机与队列通过routing_key进行绑定，到消息到达交换机时会根据消息的路由键routing_key路由到指定的队列中
            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);
            String message = "你好，世界";
            //发布消息到指定的交换机，并给消息设置routing_key用于给交换机路由到指定队列，并指定消息的格式，消息内容转换为byte后再发布
            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
            channel.close();
            connection.close();
    
        }
    }
    

  • public class ConsumerDemo {
        //定义要从哪个队列中获取消息
        private static final String QUEUE_NAME = "queue_demo";
        private static final String IP_ADDRESS = "192.168.189.118";
        private static final int PORT = 5672;
    
        public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
            //可以将RabbitMq服务器的地址存放到Address数组中，这样可以灵活来配置不同服务器地址的RabbitMq
            Address[] addresses = new Address[]{ new Address(IP_ADDRESS, PORT)};
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setUsername("root");
            connectionFactory.setPassword("root");
            Connection connection = connectionFactory.newConnection(addresses);
            final Channel channel = connection.createChannel();
            //客户端最多可以接收未被确认消费的消息数为64条
            channel.basicQos(64);
            //定义一个消费者，并指定其消费类型和在哪个channel进行消费
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("recv message:" + new String(body));
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            channel.basicConsume(QUEUE_NAME, defaultConsumer);
            TimeUnit.SECONDS.sleep(2);
            channel.close();
            connection.close();
        }
    }
    

RabbitMQ相关概念

? RabbitMQ整体上是一个生产者和消费者模型，主要负责接收、存储和转发消息

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-cmeLSiMG-1630109160153)(images/rabbitMq模型.png)]

• 生产者
  • Producer：生产者，投递消息到消息队列中的一方，消息一般包含2个部分：消息体和标签（Label），消息的标签用于将消息存储到指定队列中，给特定的消费者消费
• Broker：消息中间件的服务节点，用于将生产者发送过来的消息提交给指定的交换机再存储到指定的消息队列
  • Exchange：交换器，用于将消息路由到指定的消息队列（可以是一个或多个），当路由不到时会返回给生产者或直接丢弃
  • Queue：队列，是RabbitMQ的内部对象，用于存放消息
• 消费者
  • Consumer：消费者，获取指定队列中的消费来消费

队列

? RabbitMQ中的消息队列是一个内部对象，用于存储消息，存储的消息中不包含消息的标签，只存储消息的消息体，因而消息消费者得到消息时不会知道该消息由谁产生的

? RabbitMQ中的消息都只能存储在队列中，多个消费者可以订阅同一个队列时，消费者之间存在竞争关系，即队列中的消息被某个消费者消费后，其他的消费者就获取不到被消费的消息了

交换器

? 生产者将消息发送给交换器Exchange，交换器再将消息路由到一个或多个队列中，如果路由不到会返回给生产者或者直接丢弃

? RabbitMQ中的交换器有四种类型，不同类型的交换器有不同的路由策略

? BindingKey ：绑定键，用于交换机与消息队列间的绑定关系，当消息到达交换器时，根据消息的RoutingKey 和BindingKey 来决定消息流向哪个队列中，BindingKey 只有在特定的交换器类型下才生效，如fanout类型的交换器会无视BindingKey ，把消息路由到与该交换器绑定的所有队列中

? RoutingKey ：路由键，生产者将消息发送给交换器时，一般会指定一个RoutingKey 用来指定这个消息的路由规则，当RoutingKey 和BindingKey 匹配时，该消息会流向与该BindingKey 绑定的队列中

? 一般会把BindingKey 和RoutingKey 视为同一样东西，只是在不同的交换器类型下，它们的匹配精度不同而已

交换器类型

? RabbitMQ常用的交换器类型有fanout 、direct 、topic 、headers 四种，AMQP协议中还有两种System 和自定义类型

• fanout
  • fanout类型的交换器会把消息路由到所有与该交换器绑定的队列中
• direct
  • direct 类型的交换器会把消息路由到RoutingKey 和BindingKey 完全匹配的队列中
• topic
  • topic 类型的交换器会把消息路由到RoutingKey 和BindingKey 模糊匹配的队列中
    • BindingKey 中可以存在两种特殊的符号* 和# ，用于做模糊匹配，其中* 表示匹配一个单词，#表示匹配零个或多个 单词
• headers
  • headers 类型的交换器不依赖于路由键的匹配规则来路由消息，而是根据消息的headers 属性进行匹配

RabbitMQ运转流程

• 生产者

  1. 生产者连接到RabbitMQ：建立一个连接Connection，用该Connection开启一个信道

  Connection connection = connectionFactory.newConnection();
  Channel channel = connection.createChannel();
  

  2. 生产者声明交换器和队列：在信道中声明交换器和队列，并将队列与交换器进行绑定

  channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
  channel.queueDeclare(QUEUE_NAME, true, false, false, null);
  channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);
  

  3. 发送消息的RabbitMQ：包含路由键、交换器信息

  channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
  

  4. 关闭信道、关闭连接

  channel.close();
  connection.close();
  

• 消费者

  1. 消费者连接到RabbitMQ

  Connection connection = connectionFactory.newConnection();
  Channel channel = connection.createChannel();
  

  2. 消费者向RabbitMQ请求消费队列中的消息，接收消息后确认消费到消息

   DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
              @Override
              public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                  System.out.println("recv message:" + new String(body));
                  try {
                      TimeUnit.SECONDS.sleep(1);
                  } catch (InterruptedException e) {
                      e.printStackTrace();
                  }
                  channel.basicAck(envelope.getDeliveryTag(),false);
              }
          };
          channel.basicConsume(QUEUE_NAME, defaultConsumer);
  

  3. 关闭信道、关闭连接

  channel.close();
  connection.close();
  

连接和信道

? 无论是生产则还是消费者都需要与RabbitMQ建立连接，这个连接是一个TCP连接Connection，然后再使用这个TCP连接创建一个AMQP信道，每个信道都会被指派一个唯一的ID，信道Channel是建立在连接Connection上的虚拟连接，RabbitMQ处理每条AMQP指令都是通过信道完成的

? 信道还可以复用TCP连接，也就是说一条TCP连接可以创建多个信道，多个信道共享一条TCP连接，这样能够减少TCP创建连接的资源，提高系统的性能

AMQP协议

? AMQP，高级消息队列协议，而RabbitMQ就是AMQP协议通过Erlang语言的一种实现

? AMQP协议本身包括三层：

• Module Layer：位于协议最高层，主要定义一些供客户端使用的接口
• Session Layer：主要负责将客户端的命令发送给服务器，再将服务器的响应返回给客户端，为客户端和服务器的通信提供同步机制和错误处理
• Transport Layer：位于协议最底层，主要负责传输二进制数据流，提供帧的处理，信道复用

四、RabbitMQ相关方法

exchangeDeclare方法

Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, boolean internal, Map<String,Object> arguments) throws IOException;

• 返回值Exchange.DeclareOk用来标识成功声明了一个交换器
• 参数
  • exchange：交换器的名称
  • type：交换器的类型（fanout，direct，topic，headers）
  • durable：是否持久化
  • autoDelete：是否自动删除，当所有与该交换器绑定的队列和交换器都与此解绑后才会删除
  • internal：是否内置的，如果为true，则客户端无法直接发送消息到该交换器上，只能通过交换器路由到该交换器上的方式
  • arguments：结构化参数

queueDeclare方法

Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive,boolean autoDelete,Map<String,Object> arguments) throws IOException;

• 参数
  • queue：队列的名称
  • durable：是否可持久化
  • exclusive：是否排他性，如果设置为true，则这个队列仅对首次声明它的连接可见，并在连接断开后自动删除。需要注意的是，排他队列是基于连接的，同一连接内的不同信道可以访问该队列
  • autoDelete：是否自动删除
  • arguments：结构化参数

queueDelete方法和queuePurge方法

? queueDelete 方法删除队列

? queuePurge 方法删除队列中的消息，而不删除队列本身

queueBind方法

Queue.BindOk queueBind(String queue, String exchange, String routingKey, Map<String,Object> arguments) throws IOException;

• 参数
  • queue：队列名称
  • exchange：交换器名称
  • routingKey：用来绑定队列和交换器的路由键
  • arguments：结构化参数
• Queue.UnbindOk queueUbind(String queue, String exchange, String routingKey)throws IOException;
  • 将队列与交换器解绑

exchangeBind方法

Exchange.BindOk exchangeBind(String destination, String source, String routingKey, Map<String, Object> arguments) throws IOException;

• exchangeBind 方法将消息从source交换机转发到destination 交换机

basicPublish方法

void basicPublish(String exchange, String routingKey, boolean mandatory,boolean immediate, BasicProperties props, byte[] body)throws IOExcepion;

• 参数
  • exchange：交换器名称，指明消费会发送到哪个交换器上
  • routingKey：路由键，指明交换器路由到哪些队列上
  • mandatory：在消息无法路由到指定队列时，当mandatory 为true时，RabbitMQ会调用Basic.Return 将消息返回给生产者，当mandatory 为false时，会将消息直接丢弃
  • immediate：immediate=ture ,当交换器将消息路由到指定队列时，但该队列上并没有任何消费者，那么这条消息并不会存入队列，而是会调用Basic.Return 返回至生产者
  • props：消息的基本属性集合
  • byte[] body：消息内容

消费消息

? RabbitMQ的消费模式分两种：推模式（Push）和拉模式（Pull）模式，推模式使用Basic.Consume ，而拉模式使用Basic.Get

? Basic.Consume 将信道置为接收模式，RabbitMQ会不断地推送消息给消费者，当然消息数量会受到Basic.Qos 的限制，直到消费者取消订阅，如果想从队列获取单条消息而不是持续订阅，可以使用Basic.Get ，但不能将Basic.Get 放在一个循环中来持续Get 消息，这样会严重影响性能

• 推模式：接收消息一般通过实现Consumer 接口或继承DefaultConsumer 类来实现，同一个Channel中的消费者需要通过唯一的消费者标签来区分彼此

• String basicConsume(String queue, boolean autoAck, String consumerTag,boolean noLocal, boolean exclusive, Map<String, Object> arguments, Consumer callback) throws IOException;
  

  • 参数
    • queue：队列名称
    • autoAck：是否自动确认，建议为false，采用自动确认的方式
    • consumerTag：消费者标签，用来区分每个消费者
    • noLocal：为true时表示不能将同一个Connection中生产者发送的消息传送给这个Connection中的消费者
    • exclusive：是否排他
    • callback：消费者的回调函数，用来处理RabbitMQ推送过来的消息

• 拉模式：通过channel.basicGet 可以自动获取队列中的一条消息

• GetResponse basicGet(String queue, boolean autoAck) throws IOException;
  

消息确认和拒绝

? 为了保证消息从队列可靠地送达 到消费者，RabbitMQ提供消息确认机制，消费者订阅队列时，可以指定autoAck 参数为true 来自动确认消息，当设置为自动确认消息时，RabbitMQ会把发送出去的消息置为确认，然后删除消息，不管消费者是否真正消费到被删除的消息，当消费者没有确认消息时，RabbitMQ会等待消费者回复确认信号后才从内存或磁盘中删除消息

• 消费者通过autoAck 参数或显示自动确认Basic.Ack 来确认消息

• 拒绝消息

  • //只能拒绝一条消息
    void basicReject(long deliveryTag, boolean requeue) throws IOException;
    

  • requeue=true ：当消费者拒绝消息时，RabbitMQ会把被拒绝的消息重新存入队列

  • requeue=false： 当消费者拒绝消息时，RabbitMQ会把被拒绝的消息直接删除

  • //能够拒绝多条消息
    void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException;
    

备份交换器

? 备份交换器（Alternate Exchange，AE），如果生产者发送消息时没有设置mandatory 参数，在消息未能被发送到指定的队列的情况下会被直接丢弃，如果设置了mandatory 参数，则在客户端需要添加监听器ReturnListener 来处理返回的消息，从而增加了代码的逻辑复杂度

? 可以通过备份交换器来路由哪些未能路由到指定队列的消息，通过为交换器设置一个备份交换器，当交换器所绑定的队列中没有和消息的RoutingKey 相匹配的队列时，该消息会被转发到备份交换器，再由备份交换器路由到与备份交换器绑定的队列上

channel.exchangeDeclare("Exchange1", "direct", true, true, null);
channel.queueDeclare("queue1",true,false,false,null);
channel.queueBind("queue1","Exchange1","");
Map<String,Object> args = new HashMap<String,Object>();
//将Exchange1声明为备份交换器
args.put("alternate-exchange", "Exchange1");
//将args参数传入，表明Exchange2的备份交换器为Exchange1
channel.exchangeDeclare("Exchange2","direct",true,true,args);

过期时间

? TTL，Time to Live，RabbitMQ可以为消息和队列设置过期时间

• 队列的过期时间

  • 通过channel.queueDeclare 方法中传入x-expires 参数可以控制队列的过期时间，这里的TTL是表示该队列没有被使用的空闲时间到达过期时间

  • Map<String,Object> args = new Map<String,Object>();
    args.put("x-expires", 60000);
    channel.queueDeclare(queueName,durable,exclusive,autoDelete,args);
    

  • ?

• 队列消息的过期时间

  • 为队列设置过期时间则队列中的所有消息都有相同的生存时间，通过参数x-message-ttl 可以为队列设置过期时间，单位为毫秒

  • Map<String,Objext> args = new Map<String,Object>();
    //设置队列的TTL为1分钟
    args.put("x-message-ttl",60000)
    channel.queueDeclare(queueName,durable,exclusive,autoDelete,args);
    

• 单条消息的过期时间

  • 除了可以为队列设置过期时间之外，还可以为消息单独设置过期时间，这样在队列中的不同消息的生存时间就不同了，为消息设置过期时间可以通过设置属性expiration 来实现

  • AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
    builder.expiration("60000")
    AMQP.BasicProperties properties = builder.build();
    channel.basicPublish(exchangeName, routingKey, mandatory, properties,messagel.getBytes());
    

  • AMQP.BasicProperties properties = new AMQP.basicProperties();
    properties.setExpiration("60000");
    channel.basicPublish(exchangeName, routingKey, mandatory,properties, message.getByte());
    

• 如果同时为队列和消息设置过期时间，则以则两者中过期时间较短的为准,如果不设置TTL，表示消息不会过期，如果设置TTL为0，表示此时队列上没有消费者的话就立即把消息删除

死信队列

? DLX，Dead-Letter-Exchange，称为死信交换器，当一个队列中的消息变成死信（消息在队列中的生存时间超过其TTL时，该消息会变成死信 ），死信能够被重新发送到死信交换器，而与死信交换器绑定的队列就叫做死信队列

• 消息变成死信的情况
  • 消息被拒绝（Basic.Reject/Basic.Nack），并且requeue参数为false
  • 消息过期
  • 队列达到最大长度

任何队列都可以指定一个死信交换器，当该队列中存在死信时，RabbitMQ会自动将该死信转发到死信交换器上，再由该死信交换器路由到死信队列中

• 为队列指定死信交换器

• channel.exchangeDeclare("dlx-exchange","direct");
  Map<String,Object> args = new Map<>();
  //指定哪个交换器为死信交换器
  args.put("x-dead-letter-exchange","dlx-exchange");
  //死信交换器和普通交换器没什么不同，可以绑定多个队列
  //指定死信队列的路由键，DLK
  args.put("x-dead-letter-routing-key", "dlx-routing-key");
  //为队列声明其死信交换器
  channel.queueDeclare(queueName,durable,exclusive,autoDelete,args);
  

延迟队列

? 延迟队列中存放的消息并不会被消费者立刻消费，而是等待指定时间后，消费者才能获取到延迟队列中的消息，延迟队列可以使用死信队列和TTL来实现

? 延迟队列的实现步骤：

1. 将消息发送到普通队列中，并为消息设置其过期时间TTL，该过期时间就是延迟时间
2. 为该普通队列设置死信交换器，当普通队列中的消息过期时会被转发到死信交换器中，进而路由到死信队列中
3. 消费者获取死信队列中的消息，就是延迟了一定时间后才获取到的消息了

优先级队列

? 优先级队列中的消息具有优先级属性，优先级高的消息会被先消费掉，可以通过设置队列的x-max-priority 参数来指定某个队列为优先级队列

Map<String,Object> args = new Map<>();
//指定最大优先级为10
args.put("x-max-priority", 10);
channel.queueDeclare(queueName,durable,exclusive,autoDelete,args);
//生产者发布消息时为消息指定优先级
AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
//设置消息的优先级为5
builder.priority(5);
AMQP.BasicProperties properties = builder.build();
channel.BasicPublish(exchangeName,routingKey,properties,message.getByte());

RPC实现

? RPC，Remote Procedure Call远程过程调用，让调用远程计算机上的服务就像调用本地计算机上的服务一样，抽象了网络底层技术

? 在RabbitMQ中实现RPC架构，客户端发送请求消息到RPC队列 ，指定一个回调队列用来存储服务器的响应信息以及一个标识请求-响应的关联IdcorrelationId 用来标识该响应信息对应于哪个请求，而服务器时刻监听RPC队列，每当RPC队列中有请求消息时就处理它，并把响应消息发送到回调队列中，而客户端时刻监听回调队列，当服务器响应消息中的关联Id与自己的请求消息Id相同时，即该响应消息是发送给自己的，就将该消息获取处理消费

//获取一个队列的名称，该队列由RabbitMQ自动生成
String callbackQueue = channel.queueDeclare(),getQueue();
BasicProperties props = new
  //设置关联IdcorrelationId和回调队列
BasicProperties.builder().correlationId(corrId).replyTo(callbackQueue).build();
channel.BasicPublish("",rpc_queue,props,message.getByte());

持久化

? 持久化可以提供RabbitMQ的可靠性，将RabbitMQ中的交换器、队列或消息持久化到磁盘中，这样当RabbitMQ宕机时也能够保证消息不会丢失

? RabbitMQ的持久化有三部分：交换器持久化，队列持久化和消息持久化

• 交换器持久化：在声明交换器时将参数durable 设置为true 即可将交换器持久化到磁盘中
• 队列持久化：同样的，在声明队列时将参数durable 设置为true 即可将队列持久化到磁盘中，但如果消息没有设置为持久化，该队列中的消息并不会进行持久化，只有该队列的元数据持久化
• 消息持久化：在发布消息时可以将发送模式Basicproperties.deliveryMode 属性设置为2，来实现消息的持久化，消息是存放在队列中的，也就是说，如果队列没有进行持久化设置，那么消息持久化就没有意义了，通常将队列和消息的持久化一起设置
• 并不是说将交换器，队列和消息都设置为持久化就能保证数据百分百不会丢失了，当消费者设置自动应答时autoAck=true ，在消费者自动应答后，消费消息前异常，消息就丢失了，并且在持久化时通过会将要持久化的数据先保存到缓冲区中，当还没有持久化到磁盘时，RabbitMQ宕机也会导致数据丢失

生产者确认

? 生产者将消息发布到RabbitMQ中怎么确认该消息是否真的到达RabbitMQ中的呢，需要通过确认机制来告知生产者，有两种确方式来实现：事务机制和发布确认机制

• 事务机制

  • RabbitMQ客户端与事务有关的方法有三个：channel.txSelect 、channel.txCommit 和channel.txRollback ，RabbitMQ开启事务后，如果事务提交成功，则表明消息一定到达RabbitMQ，如果在发送过程中出现异常等情况则需要进行事务回滚

    • channel.txSelect ：用于将当前信道传输模式设置成事务模式

    • channel.txCommit ：用于提交事务

    • channel.txRollbaxk：用于回滚事务

    • try{
        //在信道中开启事务
        channel.txSelect();
        channel.basicPublish(exchangeName,routingKey,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getByte());
        ...
          //提交事务
        channel.txCommit();
      }catch(Exception e){
        //出现异常则回滚事务
        channel.txRollback();
      }
      

    • 事务机制虽然能够解决消息是否真正发送到RabbitMQ的问题，但事务机制本身十分低效，事务机制在发送一条消息后会阻塞客户端，需要等待RabbitMQ的回应后才能继续发送下一条消息，推荐使用发布确认机制来确定消息是否送达RabbitMQ

• 发布确认模式

  • 发布确认publlisher confirm 模式需要客户端将信道设置成confirm 确认模式，一旦信道进入confirm 模式，所有在该信道的消息都会被指派一个唯一的ID，一旦该消息被正确的发送到RabbitMQ上，RabbitMQ会返回一个确认Basic.Ack （包含消息的ID）给生产者来告知该消息已经到达RabbitMQ，如果设置了消息和队列时持久化的，则RabbitMQ需要在将消息和队列持久化后才会返回确认消息Basic.Ack ，而RabbitMQ返回给生产者的Basic.Ack 消息中的deliveryTag 包含有消息的ID，也可以设置channel.BasicAck 方法中的multiple 参数来批量确认消息，multiple 表示在消息Id之前的消息都确认，消息的Id序号时递增的，序号从1开始

  • 生产者通过channle.confirmSelect 方法来将信道设置为确认confirm 模式，RabbitMQ会返回Confirm.Select-OK 的确认消息表示信道模型设置成功，在confirm 模型信道中发送的消息会被RabbitMQ确认ack 或未确认nack 返回给生产者，用来告知生产者哪些消息成功送达RabbitMQ和哪些消息丢失（通过消息的唯一ID） ，而生产者能够添加一个监听器来监听RabbitMQ服务器返回的ack 和nack 消息做后事处理

  • channel.confirmSelect();
    //发送消息
    while(true){
      //获取消息的唯一ID 
      long nextSeqNo = channel.getNextPublishSeqNo();
      channel.basicPublish(exchangeName,routingKey,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getByte());
      confirmSet.add(nextSeqNo);
    }
    //添加确认消息的监听器
    channel.addConfirmListener(new ConfirmListener(){
      //消息确认的回调函数
      public void handleAck(long deliveryTag, boolean multiple) throws IOException{...};
      //消息为确认的回调函数
      public void handleNack(long deliveryTag, boolean multiple) throws IOException{...};
    });
    
    

消息分发

? 当RabbitMQ队列上存在多个消费者时，队列中的消息会轮询分发给消费者，但这样会导致当一个消息者的消费能力过强或过弱时而消费者进程出现空闲或繁忙，导致效率下降

? 可以通过channel.basicQos(int prefetchCount) 方法给信道上设置消费者运行保存的未消费的消息数量大小，这个方法就是给信道上设置一个列表来当消息缓冲区，将消息队列中的消息先发送的该列表上，而消费者从列表中取消息消费，这样就可以解决问题了

? channel.basicQos(int prefetchCount, boolean global) 中的参数global表示该信道上允许的为确认消息的总量，即所欲消费者未确认的消息

五、RabbitMQ管理

1.rabbitmqctl：rabbitmqctl [-n node] [-t timeout] [-q] [command] [command options...]

多用户与权限

? 每一个RabbitMQ服务器都能够创建虚拟的消息服务器，称之为虚拟主机vhost ，每一个vhost 本质上是一个独立的小型RabbitMQ服务器，拥有自己的队列、交换器。多个vhost 间是相对独立的，可以避免独立和较强的名称冲突

? RabbitMQ默认创建的虚拟主机为/ ，可以通过命令rabbitmqctl add_vhost vhostname 来创建虚拟主机

• rabbitmqctl list_vhosts [name] [tracing] ：来罗列虚拟主机的相关信息，name 表示以虚拟主机的名称来罗列，tracing ：表示是否使用了RabbitMQ的trace功能
• rabbitmqctl delete_vhost vhostname ：通过虚拟主机名称删除虚拟主机
• rabbitmqctl set_permissions [-p vhost] user conf write read ：RabbitMQ中用户的权限是基于虚拟主机的，创建一个用户时，通常会为该用户值某个虚拟主机上指定权限
  • conf ：一个用于匹配用户在哪些资源上拥有配置权限的正则表达式
  • write ：一个用于匹配用户在哪些资源上拥有可写权限的正则表达式（发布消息）
  • read ：一个用于匹配用户在哪些资源上拥有可读权限的正则表达式（消费消息）
• rabbitmqctl clear_permissions [-o vhost] [username] ：清除某用户在某虚拟主机上的权限
• rabbitmqctl list_permissions [-p vhost] ：虚拟主机下的用户权限
• rabbitmqctl list_user_permissions [user]：用户在哪些虚拟主机上拥有哪些权限

用户管理

? 在RabbitMQ中，用户的访问控制（Access Control）的基本单元，且单个用户可以跨虚拟主机进行授权

• rabbitmqctl add_user username password ：添加一个用户
• rabbitmqctl change_password username newpassword ：更改用户密码
• rabbitmqctl clear_password username ：清除用户密码，这样该用户就不能使用密码验证
• rabbitmqctl authenticate_user username password ：验证用户和密码的正确性
• rabbitmqctl delete_user username：删除用户
• rabbitmqctl list_users：罗列所有用户
• 用户角色
  • none：新创建的用户默认为none，没有任何角色
  • management：可以访问Web页面
  • policymaker：包含management的所有权限，并且可以管理策略Policy和参数Parameter
  • monitoring：包含management所有权限，并且可以看到所有客户端连接、信道的相关信息
  • administrator：管理员，拥有最高的权限
• rabbitmqctl set_user_tag username tag... ：为用户设置用户角色标签，设置后用户之前的角色会被清除

Web管理

? RabbitMQ management插件提供了Web管理界面用来管理RabbitMQ，需要有management权限以上的用户才能访问Web界面

• rabbitmq-plugins enable rabbitmq-management ：开启RabbitMQ management管理插件，可用于Web界面
• rabbitmq-plugins disable rabbitmq-management：关闭RabbitMQ management插件
• rabbitmq-plugins list：查看查询列表

六、RabbitMQ集群

集群搭建

1. 设置hosts：让集群中的各节点知道其他节点所在的ip地址
2. 设置相同的Cookie：设置集群中的节点都拥有相同的Cookie值，用于集群各节点访问验证的信息
3. 加入集群：将运行的RabbitMQ节点stop，并进行重置rabbitmqctl reset ，然后将该节点加入到另外的节点rabbitmqctl join_cluster rabbit@node1，其中node1为集群节点中某一节点的名称（主机名）， ，然后再启动该节点，启动后该节点就加入集群中了

RabbitMQ 对网络敏感，一般将集群搭建在同一局域网中，而不同地区的RabbitMQ消息同步使用Federation或者Shovel来代替