AMQP

相关概念介绍

1. AMQP是一个提供统一消息服务的应用层标准高级消息队列协议, 是应用层协议的一个开放标准，为面向消息的中间件设计.
2. AMQP是一个二进制协议, 具有多信道, 协商式, 异步, 安全, 多平台, 高效等优点.
3. RabbitMQ是AMQP协议的Erlang实现.
4. 相关概念介绍
   • Connection: 连接, 一个网络连接, 如tcp/ip套接字连接.
   • Session: 会话, 端点之间的命名对话. 在一个会话上下文中, 保证“恰好传递一次”.
   • Channel: 信道, 多路复用连接中的一条全双工数据流通道, 为会话提供传输介质.
     • 
   • Client: 客户端, AMQP连接或会话的发起者, AMQP是非对称的, 客户端生成和消费消息, 服务器存储和路由这些信息.
   • Broker: 服务节点, 一般情况下可以将一个RabbitMQ Broker看作一RabbitMQ服务器
   • 端点: AMQP对话的任意一方, 如一个AMQP连接包括两个端点(客户端/服务器)
   • Consumer: 消费者, 一个从消息队列里请求消息的客户端程序.
   • Producer: 生产者, 一个向交换机发布消息的客户端应用程序.

RabbitMQ运转流程

• 在简单模式中
  • 

1. 生产者发送消息
   • 生产者创建连接, 开启一个信道, 连接到RabbitMQ Broker.
   • 声明队列并设置属性: 如是否持久化, 是否独占信道, 是否自动删除.
   • 将路由键(空串)与队列绑定.
   • 发送消息至RabbitMQ Broker
   • 关闭信道, 关闭连接.
2. 消费者接收消息
   • 消费者创建连接, 开启一个信道, 连接到RabbitMQ Broker.
   • 向Broker请求消费队列中的消息, 设置相应的回调函数.
   • 等待Broker回应闭关投递响应队列中的消息, 消费者接收消息.
   • 自动确认接收到的消息.
   • RabbitMQ从队列中删除已被确认的消息.
   • 关闭信道, 关闭连接.

生产者流转过程

1. 客户端与代理服务器Broker建立连接
   • 调用newConnection()方法, 该方法会进一步封装Protocol Header 0-9-1的报文头发给Broker, 以此通知Broker交互采用AMQPO-9-1协议.
   • 紧接着Broker返回Connection.Start来建立连接, 在连接建立过程中涉及Connection.Start/.Start-OK, Connection.Tune/.Tune-Ok, Connection.Open/ .Open-Ok这六个命令的交互.
2. 客户端调用connection.createChannel()方法.
   • 该方法开启信道, 其包装的channel.open命令发给Broker
3. channel.basicPublish方法发消息, 其对应的AMQP命令为Basic.Publish, 该命令包含了contentHeader和contentBody
   • • contentHeader包含消息体的属性, 如投递模式, 优先级.
     • contentBody包含了消息体内容.
4. 客户端发送完消息要关闭资源的时候, 涉及Channel.Close和Channel.Close-OK, 以及Connection.Close,Connection.Close-OK命令的交互.

 消费者流转过程说明

1. 消费者客户端与代理服务器Broker建立连接.
   • 过程与生产者建立连接一样.
2. 消费者客户端调用connection.createChannel方法.
3. 在真正消费之前, 消费者客户端要向Broker发送Basic.Consume命令, 将Channel置为接受模式, 之后Broker回执Basic.Consume-OK命令告诉消费者已经准备好了消费消息.
4. Broker向消费者客户端推送消息, 即Basic.Deliver命令, 它一样会携带contentHeader和contentBody.
5. 消费者接收到消息并正确消费后, 会向Broker发送确认, 即Basic.Ack命令.
6. 客户端接受完消息关闭资源.

 RabbitMQ的工作模式

Work queues工作队列模式

1. 工作模式说明
   • 和简单模式相比, 多了一个或一些消费者, 多个消费端共同消费同一个队列中的信息.
   • 
2. 应用场景: 对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度.
3. 多个消费者之间采取轮询的方式去消费.
4. 代码和简单模式几乎一样, 只是多了几个消费者客户端.

订阅模式类型

• 模式示例图
  • 

1. 前面两种模式, 只有三个角色, 消费者, 生产者, 消息队列. 
2. 而在订阅模式中, 多了一个exchange交换机角色, 并且过程略有变化.
   • P: 生产者, 发送消息, 但不再发给消息队列, 而是发给X(交换机)
   • C: 消费者, 消息的接收者, 会一直等待消息到来.
   • Queue: 消息队列, 接收消息, 缓存消息.
   • Exchange: 交换机, 一方面, 会接收生产者发送的消息, 另一方面, 会处理生产者发出的消息, 如递交给某个特定队列, 或是丢弃某消息. 到底如何操作取决于Exchange的类型(常见3种)
     • Fanout: 广播, 把消息交给所有绑定到交换机的队列.
     • Direct: 定向, 把消息交给符合指定routing key的队列.
     • Topic: 通配符, 把消息交给符合routing pattern(路由模式)的队列.
3. 注意: 交换机只负责转发消息, 不具备储存消息的能力.

Publish/Subscribe发布/订阅模式

• 模式说明
  1. 每个消费者监听自己的队列
  2. 生产者将消息发给broker, 由交换机将消息转发到绑定此交换机的每个队列, 每个与交换机绑定的队列都能接收到消息.
• 生产者代码
  • 代码

    /*
    发布/订阅使用的交换机类型为: fanout
     */
    public class Producer {
    
        //交换机名称
        static final String FANOUT_EXCHAGE = "fanout_exchange";
        //队列名称
        static final String FANOUT_QUEUE_1 = "fanout_queue_1";
        //队列名称
        static final String FANOUT_QUEUE_2 = "fanout_queue_2";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            //创建连接
            Connection connection = ConnectionUtil.getConnection();
    
            // 创建频道
            Channel channel = connection.createChannel();
    
            /**
             * 声明交换机
             * 参数1：交换机名称
             * 参数2：交换机类型，fanout、topic、direct、headers(不常用)
             */
            channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
    
            channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null);
            channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null);
    
            //队列绑定交换机
            channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHAGE, "");
            channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHAGE, "");
    
            //发送消息
            for (int i = 0; i < 10; i++) {
                String message = "RabbitMQ" + Math.random();
                /**
                 * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
                 * 参数2：路由key,简单模式可以传递队列名称
                 * 参数3：消息其它属性
                 * 参数4：消息内容
                 */
                channel.basicPublish(FANOUT_EXCHAGE, "", null, message.getBytes());
            }
            System.out.println("已发送消息");
    
            //关闭资源
            channel.close();
            connection.close();
        }
    }

  • 如图所示, 两个消息队列均有生产者发出的消息.
  • 
• 消费者代码

  public class Consumer1 {
  
      public static void main(String[] args) throws IOException, TimeoutException {
          Connection connection = ConnectionUtil.getConnection();
  
          // 创建频道
          Channel channel = connection.createChannel();
  
          //声明交换机
          channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
  
          //声明队列1
          channel.queueDeclare(Producer.FANOUT_QUEUE_1, true, false, false, null);
  
          //队列绑定交换机
          channel.queueBind(Producer.FANOUT_QUEUE_1, Producer.FANOUT_EXCHAGE, "");
  
          //创建消费者；并设置消息处理
          DefaultConsumer consumer = new DefaultConsumer(channel){
              @Override
              /**
               * consumerTag 消息者标签，在channel.basicConsume时候可以指定
               * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
               * properties 属性信息
               * body 消息
               */
              public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  
                  System.out.println("消费者1-接收到的消息为：" + new String(body));
              }
          };
  
          //监听消息
          /**
           * 参数1：队列名称
           * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
           * 参数3：消息接收到后回调
           */
          channel.basicConsume(Producer.FANOUT_QUEUE_1, true, consumer);
      }
  }
  
  
  public class Consumer2 {
  
      public static void main(String[] args) throws IOException, TimeoutException {
          Connection connection = ConnectionUtil.getConnection();
  
          // 创建频道
          Channel channel = connection.createChannel();
  
          //声明交换机
          channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
  
          //声明队列2
          channel.queueDeclare(Producer.FANOUT_QUEUE_2, true, false, false, null);
  
          //队列绑定交换机
          channel.queueBind(Producer.FANOUT_QUEUE_2, Producer.FANOUT_EXCHAGE, "");
  
          //创建消费者；并设置消息处理
          DefaultConsumer consumer = new DefaultConsumer(channel){
              @Override
              public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  
                  System.out.println("消费者2-接收到的消息为：" + new String(body));
              }
          };
  
          //监听消息
          channel.basicConsume(Producer.FANOUT_QUEUE_2, true, consumer);
      }
  }

• 小结
  1. 交换机需要与队列绑定, 绑定后, 一个消息可被多个消费者都收到.
  2. 发布/订阅模式与工作队列模式的区别
     • 工作队列模式不用定义交换机, 而发布/订阅模式需要定义交换机.
     • 发布/订阅模式的生产方是面向交换机发送消息, 工作队列模式的生产方是面向队列发送消息(底层使用默认交换机).
     • 发布/订阅模式需要设置队列和交换机的绑定, 工作队列模式不需要设置, 实际上工作队列模式会将队列绑定到默认的交换机.

Routing路由模式

• 路由模式特点
  • 队列与交换机的绑定, 不是任意绑定了, 而要指定一个RoutingKey.
  • 消息的发送方在向Exchange发送消息时, 也必须指定消息的RoutingKey.
  • Exchange不再把消息交给每一个绑定的队列, 而是根据消息的RoutingKey进行判断, 只有队列的RoutingKey与消息的RoutingKey完全一致, 才会接受到消息.
• 图解
  • 
  • P: 生产者, 在向Exchange发消息时, 会指定一个routing key.
  • X: 交换机, 接收生产者的消息, 然后把消息递交给与routing key完全匹配的队列.
  • C1: 消费者, 其所在队列指定了需要routing key为error的消息.
  • C2: 消费者, 其所在队列指定了需要routing key为info, error, warning的消息.

1. 生产者代码
   • 代码

     /**
      * 路由模式的交换机类型为：direct
      */
     public class Producer {
     
         //交换机名称
         static final String DIRECT_EXCHAGE = "direct_exchange";
         //队列名称
         static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";
         //队列名称
         static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";
     
         public static void main(String[] args) throws IOException, TimeoutException {
             //创建连接
             Connection connection = ConnectionUtil.getConnection();
             // 创建频道
             Channel channel = connection.createChannel();
             //声明交换机
             channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
             //声明队列
             channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);
             channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);
             //队列绑定交换机
             /**
              * 参数1: 队列名
              * 参数2: 交换机名
              * 参数3: Routing Key
              */
             channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHAGE, "insert");
             channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHAGE, "update");
     
             // 发送信息
             String message = "routing key为insert";
             /**
              * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
              * 参数2：路由key,简单模式可以传递队列名称
              * 参数3：消息其它属性
              * 参数4：消息内容
              */
             channel.basicPublish(DIRECT_EXCHAGE, "insert", null, message.getBytes());
             System.out.println("已发送消息: " + message);
     
             // 发送信息
             message = "routing key为update";
             channel.basicPublish(DIRECT_EXCHAGE, "update", null, message.getBytes());
             System.out.println("已发送消息: " + message);
     
             // 关闭资源
             channel.close();
             connection.close();
         }
     }

   • 如图, 消息队列只会收到特定的消息
   • 
2. 消费者代码

   public class Consumer1 {
   
       public static void main(String[] args) throws IOException, TimeoutException {
           Connection connection = ConnectionUtil.getConnection();
           // 创建频道
           Channel channel = connection.createChannel();
           //声明交换机
           channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
   
           //创建队列
           channel.queueDeclare(Producer.DIRECT_QUEUE_INSERT, true, false, false, null);
   
           //队列绑定交换机
           channel.queueBind(Producer.DIRECT_QUEUE_INSERT, Producer.DIRECT_EXCHAGE, "insert");
   
           //创建消费者, 并设置消息处理
           DefaultConsumer consumer = new DefaultConsumer(channel) {
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                   System.out.println(new String(body, "utf-8"));
               }
           };
   
           channel.basicConsume(Producer.DIRECT_QUEUE_INSERT, true, consumer);
       }
   }
   
   
   public class Consumer2 {
   
       public static void main(String[] args) throws IOException, TimeoutException {
           Connection connection = ConnectionUtil.getConnection();
           Channel channel = connection.createChannel();
           channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
           channel.queueDeclare(Producer.DIRECT_QUEUE_UPDATE, true, false, false, null);
           channel.queueBind(Producer.DIRECT_QUEUE_UPDATE, Producer.DIRECT_EXCHAGE, "update");
           DefaultConsumer consumer = new DefaultConsumer(channel) {
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                   System.out.println(new String(body, "utf-8"));
               }
           };
           channel.basicConsume(Producer.DIRECT_QUEUE_UPDATE, true, consumer);
       }
   }

• 小结: Routing模式要求生产者指定一个RoutingKey给交换机, 消息队列绑定交换机时也指定RoutingKey, 这样消息会转发到符合routingKey的队列.

Topics通配符模式

• 模式说明: Topic类型和Direct相比, 都可以根据RoutingKey把消息路由到不同的队列, 但Topic类型的交换机可以让队列在绑定RoutingKey的时候使用通配符.

1. RoutingKey一般都是由一个或多个单词组成, 多个单词之间由"."分隔, 如item.insert
2. 通配符规则
   • #: 匹配一个或多个词.
   • *: 只能匹配一个词.

     item.* 只能匹配item.insert
     item.# 能匹配item.insert, item.insert.abc

3. 图解
   • 
   • 红色Queue, 绑定的是usa.#, 所以凡是以usa.开头的RoutingKey, 都会被匹配到.
   • 黄色的Queue: 绑定的是#.news, 所以凡是以.news结尾的RoutingKey, 都会被匹配的.
4. 生产者代码

   public class Producer {
   
       //交换机名称
       static final String TOPIC_EXCHAGE = "topic_exchange";
       //队列名称
       static final String TOPIC_QUEUE_1 = "topic_queue_1";
       //队列名称
       static final String TOPIC_QUEUE_2 = "topic_queue_2";
   
       public static void main(String[] args) throws IOException, TimeoutException {
           //创建连接
           Connection connection = ConnectionUtil.getConnection();
           // 创建频道
           Channel channel = connection.createChannel();
           //声明交换机
           channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
   
           //发送信息
           String message = "Topic: routing key为item.insert";
           channel.basicPublish(TOPIC_EXCHAGE, "item.insert", null, message.getBytes());
   
   
           message = "Topic: routing key为item.update";
           channel.basicPublish(TOPIC_EXCHAGE, "item.update", null, message.getBytes());
   
           message = "Topic: routing key为item.delete";
           channel.basicPublish(TOPIC_EXCHAGE, "item.delete", null, message.getBytes());
   
           channel.close();
           connection.close();
       }
   }

5. 消费者代码

   public class Consumer1 {
   
       public static void main(String[] args) throws IOException, TimeoutException {
           Connection connection = ConnectionUtil.getConnection();
   
           // 创建频道
           Channel channel = connection.createChannel();
   
           //声明交换机
           channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
   
           channel.queueDeclare(Producer.TOPIC_QUEUE_1, true, false, false, null);
   
           //队列绑定交换机
           channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.update");
           channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.delete");
   
           //创建消费者；并设置消息处理
           DefaultConsumer consumer = new DefaultConsumer(channel) {
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                   System.out.println(new String(body, "utf-8"));
               }
           };
   
           //监听消息
           channel.basicConsume(Producer.TOPIC_QUEUE_1, true, consumer);
       }
   }
   
   public class Consumer2 {
   
       public static void main(String[] args) throws IOException, TimeoutException {
   
           Connection connection = ConnectionUtil.getConnection();
   
           // 创建频道
           Channel channel = connection.createChannel();
   
           //声明交换机
           channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
   
           channel.queueDeclare(Producer.TOPIC_QUEUE_2, true, false, false, null);
   
           //队列绑定交换机
           channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHAGE, "item.*");
   
           //创建消费者；并设置消息处理
           DefaultConsumer consumer = new DefaultConsumer(channel){
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                   System.out.println(envelope.getRoutingKey());
                   System.out.println(envelope.getExchange());
                   System.out.println(envelope.getDeliveryTag());
                   System.out.println(new String(body, "utf-8"));
               }
           };
   
           //监听消息
           channel.basicConsume(Producer.TOPIC_QUEUE_2, true, consumer);
       }
   }

• 小结: Topic相比起Routing模式, 多了通配符, 使用上更灵活, 若不用通配符, 则和Routing路由模式一样.

模式总结

1. 简单模式: 一个生产者, 一个消费者, 不用自己设置交换机(用默认交换机)
2. 工作队列模式: 一个生产者, 多个消费者(轮询, 竞争), 不用自己设置交换机(用默认交换机)
3. 发布/订阅模式: 需要设置类型为fanout的交换机, 并且交换机和队列进行绑定，当发送消息到交换机后, 交换机会将消息发送到绑定的队列.
4. 路由模式: 需要设置类型为direct的交换机, 交换机和队列进行绑定, 并且指定routing key, 当发送消息到交换机后, 交换机会根据routing key将消息发送到对应的队列.
5. 通配符模式: 需要设置类型为topic的交换机, 交换机和队列进行绑定, 并且指定通配符方式的routing key, 当发送消息到交换机后, 交换机会根据routing key将消息发送到对应的队列.