一、RabbitMQ介绍

消息队列（Message Queue，简称MQ），从字面意思上看，本质是个队列，FIFO先入先出，只不过队列中存放的内容是message而已。
其主要用途：不同进程Process/线程Thread之间通信。

产生消息队列几个原因：

• 不同进程（process）之间传递消息时，两个进程之间耦合程度过高，改动一个进程，引发必须修改另一个进程，为了隔离这两个进程，在两进程间抽离出一层（一个模块），所有两进程之间传递的消息，都必须通过消息队列来传递，单独修改某一个进程，不会影响另一个；
• 不同进程（process）之间传递消息时，为了实现标准化，将消息的格式规范化了，并且某一个进程接受的消息太多无法处理完，并且也有先后顺序，必须对收到的消息进行排队，因此诞生了事实上的消息队列；

关于消息队列的详细介绍请参阅：

• 《Java帝国之消息队列》
• 《一个故事告诉你什么是消息队列》
• 《到底什么时候该使用MQ》

MQ框架非常之多，比较流行的有RabbitMq、ActiveMq、ZeroMq、kafka，以及阿里开源的RocketMQ。

1.RabbitMQ简介

RabbitMQ 是实现 AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。 RabbitMQ 主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。

RabbitMQ 是一个开源的 AMQP 实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP 等，支持 AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

AMQP，即 Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP 的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ是一个消息代理：它接受和转发消息。你可以把它想象成一个邮局：当你把你想要发布的邮件放在邮箱中时，你可以确定邮差先生最终将邮件发送给你的收件人。在这个比喻中，RabbitMQ是邮政信箱，邮局和邮递员。 
RabbitMQ和邮局的主要区别在于它不处理纸张，而是接受，存储和转发二进制数据块 - 消息。

• Git-Hub地址：https://github.com/rabbitmq
• 官方文档：https://www.rabbitmq.com/getstarted.html

2.Rabbitmq应用场景

异步处理

场景说明：用户注册后，需要发注册邮件和注册短信,传统的做法有两种1.串行的方式;2.并行的方式

1.串行方式：将注册信息写入数据库后,发送注册邮件,再发送注册短信,以上三个任务全部完成后才返回给客户端。 这有一个问题是,邮件,短信并不是必须的,它只是一个通知,而这种做法让客户端等待没有必要等待的东西.

2.并行方式：将注册信息写入数据库后,发送邮件的同时,发送短信,以上三个任务完成后,返回给客户端,并行的方式能提高处理的时间。

3.消息队列

引入消息队列后，把发送邮件,短信不是必须的业务逻辑异步处理

由此可以看出,引入消息队列后，用户的响应时间就等于写入数据库的时间+写入消息队列的时间(可以忽略不计)，引入消息队列处理后，响应时间是串行的3倍,是并行的2倍。

应用解耦

场景：双11是购物狂节,用户下单后,订单系统需要通知库存系统,传统的做法就是订单系统调用库存系统的接口。

这种做法有一个缺点:

• 当库存系统出现故障时,订单就会失败
• 订单系统和库存系统高耦合

引入消息队列

• 订单系统:用户下单后,订单系统完成持久化处理,将消息写入消息队列,返回用户订单下单成功。

• 库存系统:订阅下单的消息,获取下单消息,进行库操作。

就算库存系统出现故障,消息队列也能保证消息的可靠投递,不会导致消息丢失。

流量削峰

流量削峰一般在秒杀活动中应用广泛

• 场景：秒杀活动，一般会因为流量过大导致应用挂掉，为了解决这个问题一般在应用前端加入消息队列。
• 作用:
  • 可以控制活动人数，超过此一定阀值的订单直接丢弃
  • 可以缓解短时间的高流量压垮应用(应用程序按自己的最大处理能力获取订单)

1.用户的请求,服务器收到之后,首先写入消息队列,加入消息队列长度超过最大值,则直接抛弃用户请求或跳转到错误页面.
2.秒杀业务根据消息队列中的请求信息，再做后续处理.

3.名词解释

• Broker：它提供一种传输服务，它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输。
• Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
• Queue：消息的载体，每个消息都会被投到一个或多个队列。
• Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
• Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
• vhost：虚拟主机，一个broker里可以有多个vhost，用作不同用户的权限分离。
• Producer：消息生产者，就是投递消息的程序。
• Consumer：消息消费者，就是接受消息的程序。
• Channel：消息通道，在客户端的每个连接里，可建立多个channel。

二、RabbitMQ安装

官网下载地址：https://www.rabbitmq.com/download.html

1.Windows安装

• 安装Erlang

  • 下载地址：http://www.erlang.org/download/otp\_win64\_17.3.exe

• 安装RabbitMQ

• 启用管理工具

  启动了管理工具，可以试一下命令： 停止：net stop RabbitMQ 启动：net start RabbitMQ

• 在浏览器中输入地址查看：http://127.0.0.1:15672/

• 使用默认账号登录：guest/ guest

2.Linux安装

3.Mac安装

手动安装

下载地址：http://www.rabbitmq.com/install-standalone-mac.html

官网提供了包含erlang的完整包 ，直接下载安装包

• 解压

• 配置环境变量

  lihuandeMacBook-Pro:~ lihuan$ vi ~/.bash_profile

  改完执行，生效 lihuandeMacBook-Pro:~ lihuan$ source ~/.bash_profile

  export RABBIT_HOME=$PATH:/Users/lihuan/Documents/opt/rabbitmq_server-3.7.17 export PATH=${PATH}:${RABBIT_HOME}/sbin

• 启动服务

  lihuandeMacBook-Pro:~ lihuan$ lihuandeMacBook-Pro:~ lihuan$ rabbitmq-server

  ## RabbitMQ 3.7.17. Copyright (C) 2007-2019 Pivotal Software, Inc.

  ########## Licensed under the MPL. See https://www.rabbitmq.com/

  ########## Logs: /Users/lihuan/Documents/opt/rabbitmq_server-3.7.17/var/log/rabbitmq/rabbit@lihuandeMacBook-Pro.log /Users/lihuan/Documents/opt/rabbitmq_server-3.7.17/var/log/rabbitmq/rabbit@lihuandeMacBook-Pro_upgrade.log Starting broker... completed with 0 plugins.

• 关闭服务

  前台启动直接ctrl+c结束退出

  后台启动的话，使用 rabbitmqctl stop

• 查看rabbitmq运行状态

  rabbitmqctl status

• 查看运行日志

  tail -f rabbitmq_server-home(rabbitserver解压目录)/var/log/rabbitmq/rabbit@***.log

• RabbitMQ 启动插件

  lihuandeMacBook-Pro:~ lihuan$ rabbitmq-plugins enable rabbitmq_management Enabling plugins on node rabbit@lihuandeMacBook-Pro: rabbitmq_management The following plugins have been configured: rabbitmq_management rabbitmq_management_agent rabbitmq_web_dispatch Applying plugin configuration to rabbit@lihuandeMacBook-Pro... The following plugins have been enabled: rabbitmq_management rabbitmq_management_agent rabbitmq_web_dispatch

  started 3 plugins. lihuandeMacBook-Pro:~ lihuan$

 $: sudo /sbin/rabbitmq-plugins enable rabbitmq_management（执行一次以后不用再次执行）

• 登陆管理界面 (http://localhost:15672/)

• 账号密码初始默认都为guest

启动服务后的常用操作:

rabbitmqctl list_users  #查看当前所有用户

lihuandeMacBook-Pro:~ lihuan$ rabbitmqctl list_users
Listing users ...
user    tags
guest    [administrator]
lihuandeMacBook-Pro:~ lihuan$

由于rabbitmq的默认用户名和密码都是guest,为了安全起见,可以先删除guest账户

rabbitmqctl delete_user guest 删除guest账户

rabbitmqctl add_user admin admin123 然后创建新用户.admin 密码admin123 用户名密码自定义

rabbitmqctl set_user_tags admin administrator 分配admin到administrator分组

rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"  分配admin账户有所有权限

rabbitmqctl list_user_permissions admin 查看用户权限

三、RabbitMq管理

1.添加admin用户

lihuandeMacBook-Pro:~ lihuan$ rabbitmqctl list_users
Listing users ...
user    tags
admin    [administrator]
guest    [administrator]
lihuandeMacBook-Pro:~ lihuan$

2.用户角色

1、超级管理员(administrator)
可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。
2、监控者(monitoring)
可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)
3、策略制定者(policymaker)
可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。
4、普通管理者(management)
仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。
5、其他
无法登陆管理控制台，通常就是普通的生产者和消费者。

3.创建Virtual Hosts

像mysql有数据库的概念并且可以指定用户对库和表等操作的权限。RabbitMQ也有类似的权限管理。在RabbitMQ中可以虚拟消息服务器VirtualHost，每个VirtualHost相当用一个相对独立的RabbitMQ服务器，每个VirtualHost之间是相互隔离的。exchange、queue、message不能互通。 在RabbitMQ中无法通过AMQP创建VirtualHost，可以通过以下命令来创建。

4.管理界面配置

• 新建虚拟主机
• 新建Exchange

• 新建Queue

Total代表队列中的消息总条数，Ready代表消费者还可以读到的条数，Unacked:代表还有多少条没有被应答

• 绑定Exchange和Queue

代码里边需要绑定Exchange和Queue，管理平台也需要手动去绑定Exchange和Queue

四、消费模式

1.简单队列

• 导入RabbitMQ的客户端依赖

  com.rabbitmq amqp-client 3.4.1

• 获取MQ的连接

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

  /**

  • @author Huan Lee
  • @version 1.0
  • @date 2019-08-27 14:48
  • @describtion 业精于勤，荒于嬉；行成于思，毁于随。

  */ public class RabbitmqFactory {

  public static Connection getConnection() throws Exception {
      //定义连接工厂
      ConnectionFactory factory = new ConnectionFactory();
      //设置服务地址
      factory.setHost("localhost");
      //端口
      factory.setPort(5672);
      //设置账号信息，用户名、密码、vhost
      factory.setVirtualHost("testhost");
      factory.setUsername("admin");
      factory.setPassword("admin");
      // 通过工程获取连接
      Connection connection = factory.newConnection();
      return connection;
  }
  

  }

• 生产者发送消息到队列

  import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.service.config.RabbitmqFactory;

  /**

  • @author Huan Lee
  • @version 1.0
  • @date 2019-08-27 14:51
  • @describtion 业精于勤，荒于嬉；行成于思，毁于随。

  */ public class RabbitmqSender {

  private final static String QUEUE_NAME = "q_test_01";
  
  public static void main(String[] argv) throws Exception {
      // 获取到连接以及mq通道
      Connection connection = RabbitmqFactory.getConnection();
      // 从连接中创建通道
      Channel channel = connection.createChannel();
  
      // 声明（创建）队列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  
      // 消息内容
      String message = "Hello World!";
      channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
      System.out.println(" [x] Sent '" + message + "'");
      //关闭通道和连接
      channel.close();
      connection.close();
  }
  

  }

• 管理工具中查看消息

点击上面的队列名称，查询具体的队列中的信息：

• 消费者从队列中获取消息

  import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.QueueingConsumer; import com.rabbitmq.service.config.RabbitmqFactory;

  /**

  • @author Huan Lee
  • @version 1.0
  • @date 2019-08-27 15:11
  • @describtion 业精于勤，荒于嬉；行成于思，毁于随。

  */ public class RabbitmqConsumer {

  private final static String QUEUE_NAME = "q_test_01";
  
  public static void main(String[] argv) throws Exception {
  
      // 获取到连接以及mq通道
      Connection connection = RabbitmqFactory.getConnection();
      // 从连接中创建通道
      Channel channel = connection.createChannel();
      // 声明队列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  
      // 定义队列的消费者
      QueueingConsumer consumer = new QueueingConsumer(channel);
  
      // 监听队列
      channel.basicConsume(QUEUE_NAME, true, consumer);
  
      // 获取消息
      while (true) {
          QueueingConsumer.Delivery delivery = consumer.nextDelivery();
          String message = new String(delivery.getBody());
          System.out.println(" [x] Received '" + message + "'");
      }
  }
  

  }

• 管理工具中查看消费消息

2.Work模式

一个生产者、2个消费者。一个消息只能被一个消费者获取。

• 生产者消息到mq

  import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.service.config.RabbitmqFactory;

  /**

  • @author Huan Lee
  • @version 1.0
  • @date 2019-08-27 15:32
  • @describtion 业精于勤，荒于嬉；行成于思，毁于随。

  */ public class RabbitmqSender {

  private final static String QUEUE_NAME = "test_queue_work";
  
  public static void main(String[] argv) throws Exception {
      // 获取到连接以及mq通道
      Connection connection = RabbitmqFactory.getConnection();
      Channel channel = connection.createChannel();
  
      // 声明队列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  
      for (int i = 0; i < 100; i++) {
          // 消息内容
          String message = "" + i;
          channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
          System.out.println(" [x] Sent '" + message + "'");
  
          Thread.sleep(i * 10);
      }
  
      channel.close();
      connection.close();
  }
  

  }

• Rabbitmq管理平台查看发送消息

• 消费者1消费消息

  import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.QueueingConsumer; import com.rabbitmq.service.config.RabbitmqFactory;

  /**

  • @author Huan Lee
  • @version 1.0
  • @date 2019-08-27 15:29
  • @describtion 业精于勤，荒于嬉；行成于思，毁于随。

  */ public class RabbitmqConsumer1 {

  private final static String QUEUE_NAME = "test_queue_work";
  
  public static void main(String[] argv) throws Exception {
  
      // 获取到连接以及mq通道
      Connection connection = RabbitmqFactory.getConnection();
      Channel channel = connection.createChannel();
  
      // 声明队列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  
      // 同一时刻服务器只会发一条消息给消费者
      //channel.basicQos(1);
  
      // 定义队列的消费者
      QueueingConsumer consumer = new QueueingConsumer(channel);
  
      // 监听队列，false表示手动返回完成状态，true表示自动
      channel.basicConsume(QUEUE_NAME, true, consumer);
  
      // 获取消息
      while (true) {
          QueueingConsumer.Delivery delivery = consumer.nextDelivery();
          String message = new String(delivery.getBody());
          System.out.println(" [y] Received '" + message + "'");
          //休眠
          Thread.sleep(10);
          // 返回确认状态，注释掉表示使用自动确认模式
          //channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
      }
  }
  

  }

• 消费者2消费消息

  import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.QueueingConsumer; import com.rabbitmq.service.config.RabbitmqFactory;

  /**

  • @author Huan Lee
  • @version 1.0
  • @date 2019-08-27 15:31
  • @describtion 业精于勤，荒于嬉；行成于思，毁于随。

  */ public class RabbitmqConsumer2 {

  private final static String QUEUE_NAME = "test_queue_work";
  
  public static void main(String[] argv) throws Exception {
  
      // 获取到连接以及mq通道
      Connection connection = RabbitmqFactory.getConnection();
      Channel channel = connection.createChannel();
  
      // 声明队列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  
      // 同一时刻服务器只会发一条消息给消费者
      //channel.basicQos(1);
  
      // 定义队列的消费者
      QueueingConsumer consumer = new QueueingConsumer(channel);
      // 监听队列，false表示手动返回完成状态，true表示自动
      channel.basicConsume(QUEUE_NAME, true, consumer);
  
      // 获取消息
      while (true) {
          QueueingConsumer.Delivery delivery = consumer.nextDelivery();
          String message = new String(delivery.getBody());
          System.out.println(" [x] Received '" + message + "'");
          // 休眠1秒
          Thread.sleep(1000);
          //下面这行注释掉表示使用自动确认模式
          //channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
      }
  }
  

  }

• Rabbitmq管理平台查看消费消息

测试结果：
1、消费者1和消费者2获取到的消息内容是不同的，同一个消息只能被一个消费者获取。
2、消费者1和消费者2获取到的消息的数量是相同的，一个是消费奇数号消息，一个是偶数。

其实，这样是不合理的，因为消费者1线程停顿的时间短。应该是消费者1要比消费者2获取到的消息多才对。
RabbitMQ 默认将消息顺序发送给下一个消费者，这样每个消费者会得到相同数量的消息。即轮询（round-robin）分发消息。

怎样才能做到按照每个消费者的能力分配消息呢？联合使用 Qos 和 Acknowledge 就可以做到。
basicQos 方法设置了当前信道最大预获取（prefetch）消息数量为1。消息从队列异步推送给消费者，消费者的 ack 也是异步发送给队列，从队列的视角去看，总是会有一批消息已推送但尚未获得 ack 确认，Qos 的 prefetchCount 参数就是用来限制这批未确认消息数量的。设为1时，队列只有在收到消费者发回的上一条消息 ack 确认后，才会向该消费者发送下一条消息。prefetchCount 的默认值为0，即没有限制，队列会将所有消息尽快发给消费者。

2个概念

轮询分发 ：使用任务队列的优点之一就是可以轻易的并行工作。如果我们积压了好多工作，我们可以通过增加工作者（消费者）来解决这一问题，使得系统的伸缩性更加容易。在默认情况下，RabbitMQ将逐个发送消息到在序列中的下一个消费者(而不考虑每个任务的时长等等，且是提前一次性分配，并非一个一个分配)。平均每个消费者获得相同数量的消息。这种方式分发消息机制称为Round-Robin（轮询）。

公平分发 ：虽然上面的分配法方式也还行，但是有个问题就是：比如：现在有2个消费者，所有的奇数的消息都是繁忙的，而偶数则是轻松的。按照轮询的方式，奇数的任务交给了第一个消费者，所以一直在忙个不停。偶数的任务交给另一个消费者，则立即完成任务，然后闲得不行。而RabbitMQ则是不了解这些的。这是因为当消息进入队列，RabbitMQ就会分派消息。它不看消费者为应答的数目，只是盲目的将消息发给轮询指定的消费者。

为了解决这个问题，我们使用basicQos( prefetchCount = 1)方法，来限制RabbitMQ只发不超过1条的消息给同一个消费者。当消息处理完毕后，有了反馈，才会进行第二次发送。
还有一点需要注意，使用公平分发，必须关闭自动应答，改为手动应答。

Work模式的“能者多劳”

打开上述代码的注释：

// 同一时刻服务器只会发一条消息给消费者
channel.basicQos(1);

//开启这行 表示使用手动确认模式
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

同时改为手动确认：

// 监听队列，false表示手动返回完成状态，true表示自动
channel.basicConsume(QUEUE_NAME, false, consumer);

测试：
消费者1比消费者2获取的消息更多。

消息的确认模式

消费者从队列中获取消息，服务端如何知道消息已经被消费呢？

模式1：自动确认
只要消息从队列中获取，无论消费者获取到消息后是否成功消息，都认为是消息已经成功消费。
模式2：手动确认
消费者从队列中获取消息后，服务器会将该消息标记为不可用状态，等待消费者的反馈，如果消费者一直没有反馈，那么该消息将一直处于不可用状态。

3.订阅模式

解读：
1、1个生产者，多个消费者
2、每一个消费者都有自己的一个队列
3、生产者没有将消息直接发送到队列，而是发送到了交换机
4、每个队列都要绑定到交换机
5、生产者发送的消息，经过交换机，到达队列，实现，一个消息被多个消费者获取的目的
注意：一个消费者队列可以有多个消费者实例，只有其中一个消费者实例会消费

• 消息的生产者--向交换机中发送消息

  import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.service.config.RabbitmqFactory;

  /**

  • @author Huan Lee
  • @version 1.0
  • @date 2019-08-27 15:57
  • @describtion 业精于勤，荒于嬉；行成于思，毁于随。

  */ public class RabbitmqSender {

  private final static String EXCHANGE_NAME = "test_exchange_fanout";
  
  public static void main(String[] argv) throws Exception {
      // 获取到连接以及mq通道
      Connection connection = RabbitmqFactory.getConnection();
      Channel channel = connection.createChannel();
  
      // 声明exchange
      channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
  
      // 消息内容
      String message = "Hello World!";
      channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
      System.out.println(" [x] Sent '" + message + "'");
  
      channel.close();
      connection.close();
  }
  

  }

注意：消息发送到没有队列绑定的交换机时，消息将丢失，因为，交换机没有存储消息的能力，消息只能存在在队列中。

• 消息消费者1

  import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.QueueingConsumer; import com.rabbitmq.service.config.RabbitmqFactory;

  /**

  • @author Huan Lee
  • @version 1.0
  • @date 2019-08-27 16:08
  • @describtion 业精于勤，荒于嬉；行成于思，毁于随。

  */ public class RabbitmqConsumer1 {

  private final static String QUEUE_NAME = "test_queue_work1";
  
  private final static String EXCHANGE_NAME = "test_exchange_fanout";
  
  public static void main(String[] argv) throws Exception {
  
      // 获取到连接以及mq通道
      Connection connection = RabbitmqFactory.getConnection();
      Channel channel = connection.createChannel();
  
      // 声明队列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  
      // 绑定队列到交换机
      channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
  
      // 同一时刻服务器只会发一条消息给消费者
      channel.basicQos(1);
  
      // 定义队列的消费者
      QueueingConsumer consumer = new QueueingConsumer(channel);
      // 监听队列，手动返回完成
      channel.basicConsume(QUEUE_NAME, false, consumer);
  
      // 获取消息
      while (true) {
          QueueingConsumer.Delivery delivery = consumer.nextDelivery();
          String message = new String(delivery.getBody());
          System.out.println(" [Recv] Received '" + message + "'");
          Thread.sleep(1000);
  
          channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
      }
  }
  

  }

• 消息消费者2

  import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.QueueingConsumer; import com.rabbitmq.service.config.RabbitmqFactory;

  /**

  • @author Huan Lee
  • @version 1.0
  • @date 2019-08-27 16:10
  • @describtion 业精于勤，荒于嬉；行成于思，毁于随。

  */ public class RabbitmqConsumer2 {

  private final static String QUEUE_NAME = "test_queue_work2";
  
  private final static String EXCHANGE_NAME = "test_exchange_fanout";
  
  public static void main(String[] argv) throws Exception {
  
      // 获取到连接以及mq通道
      Connection connection = RabbitmqFactory.getConnection();
      Channel channel = connection.createChannel();
  
      // 声明队列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  
      // 绑定队列到交换机
      channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
  
      // 同一时刻服务器只会发一条消息给消费者
      channel.basicQos(1);
  
      // 定义队列的消费者
      QueueingConsumer consumer = new QueueingConsumer(channel);
      // 监听队列，手动返回完成
      channel.basicConsume(QUEUE_NAME, false, consumer);
  
      // 获取消息
      while (true) {
          QueueingConsumer.Delivery delivery = consumer.nextDelivery();
          String message = new String(delivery.getBody());
          System.out.println(" [Recv2] Received '" + message + "'");
          Thread.sleep(1000);
  
          channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
      }
  }
  

  }

测试结果：
同一个消息被多个消费者获取。一个消费者队列可以有多个消费者实例，只有其中一个消费者实例会消费到消息。

• 管理工具中查看队列和交换机的绑定关系

4.路由模式

五、RabbitMq解析

1.任务分发机制

循环分发

RabbbitMQ的分发机制非常适合扩展,而且它是专门为并发程序设计的,如果现在load加重,那么只需要创建更多的Consumer来进行任务处理。

消息确认

为了保证数据不被丢失,RabbitMQ支持消息确认机制,为了保证数据能被正确处理而不仅仅是被Consumer收到,那么我们不能采用no-ack，而应该是在处理完数据之后发送ack。
在处理完数据之后发送ack,就是告诉RabbitMQ数据已经被接收,处理完成,RabbitMQ可以安全的删除它了.
如果Consumer退出了但是没有发送ack,那么RabbitMQ就会把这个Message发送到下一个Consumer，这样就保证在Consumer异常退出情况下数据也不会丢失。

RabbitMQ它没有用到超时机制.RabbitMQ仅仅通过Consumer的连接中断来确认该Message并没有正确处理，也就是说RabbitMQ给了Consumer足够长的时间做数据处理。
如果忘记ack,那么当Consumer退出时,Mesage会重新分发,然后RabbitMQ会占用越来越多的内存。

2.消息持久化

RabbitMQ默认情况下的交换机和队列以及消息是非持久化的，也就是说在服务器重启或者宕机恢复后，之前创建的交换机和队列都将不复存在，之前未消费的消息也就消失不见了。原因在于每个队列和交换机的durable属性。该属性默认情况是false，它决定了RabbitMQ是否需要在崩溃或者重启之后重新创建队列（或者交换机）。

将交换机和队列的durable属性设置为true，这样你就不需要在服务器断电后重新创建队列和交换机了。你也许会认为把队列和交换机的durable属性设置为true就足够可以让消息幸免于重启后丢失了，真的是这样吗？队列和交换机当然必须被设置为true，但光这样做还不够。 
能从AMQP服务器崩溃中恢复的消息，我们称之为持久化消息。在消息发布前，通过把它的“投递默认”（ delivery mode）选项设置为2（AMQP客户端可能会使用人性化的常量来代替数值）来把消息标记成持久化。到目前为止，消息还只是被表示为持久化的，但是它还必须被发布到持久化的交换机中，并到达持久化的队列中才行。如果不是这样的话，则包含持久化消息的队列（或者交换机）会在Rabbit崩溃重启后不复存在，从而导致消息丢失。

因此，如果消息想要从Rabbit崩溃中恢复，那么消息必须满足以下条件：

1. 把它的投递默认选项设置为持久化
2. 发送到持久化的交换机
3. 到达持久化的队列

做到以上三点，你就不需要担心发送到Rabbit服务器的消息因服务器崩溃等其它原因而丢失了。

要持久化队列queue的持久化需要在声明时指定durable=True;

队列和交换机有一个创建时候指定的标志durable,durable的唯一含义就是具有这个标志的队列和交换机会在重启之后重新建立,它不表示说在队列中的消息会在重启后恢复。

消息持久化包括3部分

• exchange持久化,在声明时指定durable => true

  hannel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);//声明交换机，且为可持久化的

• queue持久化,在声明时指定durable => true

  channel.QueueDeclare(QueueName, durable: true, exclusive: false, autoDelete: false, arguments: null);//声明消息队列，且为可持久化的

• 消息持久化,在投递时指定delivery_mode => 2(1是非持久化)

  channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());

如果exchange和queue都是持久化的,那么它们之间的binding也是持久化的,如果exchange和queue两者之间有一个持久化，一个非持久化,则不允许建立绑定。
注意：一旦创建了队列和交换机,就不能修改其标志了,例如,创建了一个non-durable的队列,然后想把它改变成durable的,唯一的办法就是删除这个队列然后重现创建。

RabbitMQ确保持久性消息能从服务器重启中恢复的方式是，将它们写入磁盘上的一个持久化日志文件。当发布一个持久性消息到持久交换机上时，Rabbit会在消息提交到日志文件后才发送响应。记住，之后这条消息如果路由到了非持久队列的话，它会自动从持久性日志中移除，并且无法从服务器重启中恢复。如果你使用持久性消息的话，则确保之前提到的持久性消息的那三点都必须做到位。一旦你从持久性队列中消费了一个持久性消息的话（并且确认了它），RabbitMQ会在持久化日志中把这条消息标记为等待垃圾收集。在你消费持久性消息前，如果RabbitMQ重启的话，服务器会自动重建交换机和队列（以及绑定），重播持久性日志文件的消息到合适的队列或者交换机上（取决于Rabbit服务器宕机的时候，消息处在路由过程的哪个环节）。

虽然持久化消息可以做到消息的不丢失，但持久化的消息在进入队列前会被写到磁盘，这个过程比写到内存慢得多，所以会严重的影响性能，可能导致消息的吞吐量降低10倍不止。所以，在做消息持久化前，一定要认真考虑性能和需求之间的平衡关系。

3.公平分发

你可能也注意到了,分发机制不是那么优雅,默认状态下,RabbitMQ将第n个Message分发给第n个Consumer。n是取余后的,它不管Consumer是否还有unacked Message，只是按照这个默认的机制进行分发.

那么如果有个Consumer工作比较重,那么就会导致有的Consumer基本没事可做,有的Consumer却毫无休息的机会,那么,Rabbit是如何处理这种问题呢?

通过basic.qos方法设置prefetch_count=1，这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message，换句话说,在接收到该Consumer的ack前,它不会将新的Message分发给它

channel.basic_qos(prefetch_count=1) 

注意，这种方法可能会导致queue满。当然，这种情况下你可能需要添加更多的Consumer，或者创建更多的virtualHost来细化你的设计。

4.分发到多个Consumer

Exchange

• Direct Exchange：直接匹配,通过Exchange名称+RountingKey来发送与接收消息。
• Fanout Exchange：广播订阅,向所有的消费者发布消息,但是只有消费者将队列绑定到该路由器才能收到消息,忽略Routing Key。
• Topic Exchange：主题匹配订阅,这里的主题指的是RoutingKey,RoutingKey可以采用通配符,如:*或#，RoutingKey命名采用.来分隔多个词,只有消息这将队列绑定到该路由器且指定RoutingKey符合匹配规则时才能收到消息。
• Headers Exchange：消息头订阅,消息发布前,为消息定义一个或多个键值对的消息头,然后消费者接收消息同时需要定义类似的键值对请求头:(如:x-mactch=all或者x_match=any)，只有请求头与消息头匹配,才能接收消息,忽略RoutingKey.

默认的exchange:如果用空字符串去声明一个exchange，那么系统就会使用”amq.direct”这个exchange，我们创建一个queue时,默认的都会有一个和新建queue同名的routingKey绑定到这个默认的exchange上去。

channel.BasicPublish("", "TaskQueue", properties, bytes);

因为在第一个参数选择了默认的exchange，而我们申明的队列叫TaskQueue，所以默认的，它在新建一个也叫TaskQueue的routingKey，并绑定在默认的exchange上，导致了我们可以在第二个参数routingKey中写TaskQueue，这样它就会找到定义的同名的queue，并把消息放进去。

如果有两个接收程序都是用了同一个的queue和相同的routingKey去绑定direct exchange的话，分发的行为是负载均衡的，也就是说第一个是程序1收到，第二个是程序2收到，以此类推。
如果有两个接收程序用了各自的queue，但使用相同的routingKey去绑定direct exchange的话，分发的行为是复制的，也就是说每个程序都会收到这个消息的副本。行为相当于fanout类型的exchange。

Bindings 绑定

绑定其实就是关联了exchange和queue，或者这么说:queue对exchange的内容感兴趣,exchange要把它的Message deliver到queue。

Direct exchange

Driect exchange的路由算法非常简单:通过bindingkey的完全匹配。

Exchange和两个队列绑定在一起,Q1的bindingkey是orange，Q2的binding key是black和green.
当Producer publish key是orange时,exchange会把它放到Q1上,如果是black或green就会到Q2上,其余的Message被丢弃.

Multiple bindings

多个queue绑定同一个key也是可以的,对于下图的例子,Q1和Q2都绑定了black,对于routing key是black的Message，会被deliver到Q1和Q2，其余的Message都会被丢弃.

Topic exchange

对于Message的routing_key是有限制的，不能使任意的。格式是以点号“.”分割的字符表。比如：”stock.usd.nyse”, “nyse.vmw”, “quick.orange.rabbit”。你可以放任意的key在routing_key中，当然最长不能超过255 bytes。

对于routing_key，有两个特殊字符

• *(星号)代表任意一个单词
• #(hash)0个或多个单词

https://blog.csdn.net/whoamiyang/article/details/54954780

参考文章：

• https://blog.csdn.net/hellozpc/article/details/81436980
• Java 帝国之消息队列
• 一个故事告诉你什么是消息队列