首先先来了解一下两个名词

JMS

​Java Message Service，Java定义的一套消息服务标准，符合JMS标准规范的，都是通用的Java消息服务

MOM

​Message Oriented Middleware，面向消息的中间件符合消息开发标准规范的中间件产品，例如ActliveMQ、RabbitMQ、Kafka等。可以提供消息存储机制、提供消息的发送和消费服务，提供消息的缓存处理等功能的中间件产品。符合MOM规范的产品，同时可以依托JMS标准规范访问的产品，可以称为JMS Provider

MQ的好处

• 解耦

  ​在多个组件之间原本进行网络调用的方式现在换成MQ的方式来进行消息的异步通讯，倘若一个消费端系统下线，影响也仅仅是消息积压在MQ中没有被消费而已

• 可恢复性

  系统的一部分组件失效时，不会影响到整个业务系统。MQ能够保证加入队列的消息仍然可以在组件恢复后继续被消费处理

• 异步通信

  ​消息队列提供了异步处理机制，能够很好的提升用户的体验度（订单系统为例，用户下订单后，订单系统直接返回下订单成功的结果，然后将数据封装到MQ中，最后进入数据库）

• 峰值处理

  ​MQ能够使关键组件在访问高峰时顶住巨大的压力（订单系统为例，订单会被存储到MQ队列中，不会直接访问消费端，消费端通过拉取的方式控制处理速度，使流量趋于平稳，达到了削峰填谷的目的）

• 扩展性

  ​由于组件之间的耦合度很低，所以增大消息入队和处理频率是很方便的

RabbitMQ的原理

• Broker

  ​ 接受客户端连接，实现AMQP消息队列和路由功能的进程（就是我们启动的RabbitServer）

• Vhost

  ​ 虚拟主机，一个VH里面可以有多个Exchange和Queue。当多个不同用户使用同一个RabbitMQ服务（Broker）时，会划分出多个VH

• Exchange

  ​ 接受发送方的消息，根据Binding规则将消息路由到不同的Queue中

• Queue

  ​ 消息存储的地方。多个发送方可以向一个队列发送消息，多个消费方可以消费一个队列的消息，本质上是一个缓冲区，遵循FIFO（先进先出）的处理机制，在RabbitMQ的Queue中，可以设置消息持久化或自动删除

• Channel

  ​ 由于大量建立TCP连接不现实，所以AMQP（高级消息队列协议）规定：消息都必须经过信道发送出去

绑定策略

• Direct

  ​ 是一种点对点的交换器，发送方发送消息到MQ中，MQ的direct交换器接收到消息后，会根据Routiong Key来决定消息将会被发送到哪一个队列中；而接收方则需要负责监视一个队列（通过注册队列监听器），当队列状态发生变化时消费消息。注册队列监听器需要提供交换器信息、队列信息和路由键信息

  发送方

  yaml

  server:port: 8282spring:rabbitmq:host: 192.168.49.142port: 5672username: guestpassword: guestvirtual-host: /
  

  pom

  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId>
  </dependency>
  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>
  

  direct

  @Component
  public class DirectMessagePush {/*** 注入RabbitMQ逻辑模板*/@Resourceprivate AmqpTemplate amqpTemplate;public void sendMessage(Order order) {/*** order-exchange 具体的交换器* order 路由键*/this.amqpTemplate.convertAndSend("order-exchange", "order", order);}}
  

  controller

  @Controller
  public class Send {@Resourceprivate DirectMessagePush directMessagePush;@GetMapping("/order")@ResponseBodypublic String sendOrder() {Order order = new Order();order.setId(1);order.setTotalPrice(18000.00);List<Item> items = new ArrayList<>();for (int i = 0; i < 5; i++) {Item item = new Item();item.setId(1 + i);item.setPrice(2000.00 + i * 200);item.setProductName("Apple 16");item.setRemark("2019 16寸");items.add(item);}order.setItems(items);this.directMessagePush.sendMessage(order);return "SUCCESS !";}}
  

  接收方

  yaml

  server:port: 8181spring:rabbitmq:host: 192.168.49.142port: 5672username: guestpassword: guestvirtual-host: /listener:direct:retry:enabled: truemax-attempts: 10
  

  pom

  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId>
  </dependency>
  <!-- AMQP插件组，用于开发spring boot访问符合AMQP协议的MQ产品的依赖启动器 -->
  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>
  

  consumer

  /*** @RabbitListener RabbitMQ的监听器类* 		bindings 绑定策略* @QueueBinding 声明具体的绑定策略* 		value 具体绑定的队列* 		exchange 队列对应的交换器* 		key 绑定的具体路由键* @Queue 具体的队列描述* 		name 队列名称，消费者关注，发布者不关注* 		autoDelete* 			“true” 当队列没有被任何消费者监听的时候，RabbitMQ自动删除该队列*			“false” 只要队列创建，永不删除，RabbitMQ保存未被消费的消息等待其它消费监听者处理* @Exchange 具体的交换器* 		name 交换器名称*	    autoDelete 当没有队列与交换器绑定时，是否删除该交换器* 		type 交换器的类型*/
  @RabbitListener(bindings = {@QueueBinding(value = @Queue(name = "order-queue", autoDelete = "false"),exchange = @Exchange(name = "order-exchange", type = "direct", autoDelete = "false"),key = "order")
  })
  @Component
  public class OrderMassage {/*** @param order 消息* @RabbitHandler 标记当前方法是消费消息的方法* 该方法将会被会注册到MQ上，监听MQ的队列，当队列中出现消息的时候自动消费* <p>* 消费端方法不能有返回值！*/@RabbitHandlerpublic void doSomething(Order order) {/*** 相关业务*/System.out.println("order = ------------------>>> " + order);}}
  

• Fanout

  ​ 广播交换器。将接收到的消息广播发送到绑定匹配的所有队列中，这个过程交换器不会匹配Routing Key，所以消息中不需要提供路由键信息；接收方则需要负责监视一个队列（通过注册队列监听器），当队列状态发生变化时消费消息。注册队列监听器需要提供交换器信息、队列信息

  发送方

  yaml

  server:port: 8282spring:rabbitmq:host: 192.168.49.142port: 5672username: guestpassword: guestvirtual-host: /
  

  pom

  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId>
  </dependency>
  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>
  

  fanout

  @Component
  public class FanoutPublisher {@Resourceprivate AmqpTemplate amqpTemplate;public void sendStr(String str) {// 中间占位参数不能缺省！this.amqpTemplate.convertAndSend("fanout-exchange", "", str);}}
  

  controller

  @Controller
  public class Send {@Resourceprivate FanoutPublisher fanoutPublisher;@GetMapping("/fanout")@ResponseBodypublic String sendStr() {this.fanoutPublisher.sendStr("广播通知：... ...");return "OK";}}
  

  接收方

  yaml

  server:port: 8181spring:rabbitmq:host: 192.168.49.142port: 5672username: guestpassword: guestvirtual-host: /listener:direct:retry:enabled: truemax-attempts: 10
  

  pom

  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId>
  </dependency>
  <!-- AMQP插件组，用于开发spring boot访问符合AMQP协议的MQ产品的依赖启动器 -->
  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>
  

  consumer1

  /*** 广播队列的消费者1*/
  @RabbitListener(bindings = {@QueueBinding(value = @Queue(name = "log-fanout1", autoDelete = "false"),exchange = @Exchange(name = "fanout-exchange", autoDelete = "false", type = "fanout"))
  })
  @Component
  public class Person1 {@RabbitHandlerpublic void fanoutHandler(String srt) {System.out.println("srt1 ----------------------------> " + srt);}}
  

  consumer2

  /*** 广播队列的消费者2*/
  @RabbitListener(bindings = {@QueueBinding(value = @Queue(name = "log-fanout2", autoDelete = "false"),exchange = @Exchange(name = "fanout-exchange", autoDelete = "false", type = "fanout"))
  })
  @Component
  public class Person2 {@RabbitHandlerpublic void fanoutHandler(String srt) {System.out.println("srt2 ----------------------------> " + srt);}}
  

• Topic

  ​ 主题交换器。也称之为规则匹配交换器。通过自定义的匹配规则来决定消息存储到哪些队列中，MQ中的交换器会根据Routing Key来决定消息应该发送到某具体队列；接收方则需要负责监视一个队列（通过注册队列监听器），当队列状态发生变化时消费消息。注册队列监听器需要提供交换器信息、队列信息和路由键信息

  发送方

  yaml

  server:port: 8282spring:rabbitmq:host: 192.168.49.142port: 5672username: guestpassword: guestvirtual-host: /
  

  pom

  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId>
  </dependency>
  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>
  

  topic

  /*** 发送日志消息*/
  @Component
  public class LogPublisher {@Resourceprivate AmqpTemplate amqpTemplate;public void sendLog(String log) {Random random = new Random();int num = random.nextInt(10000);String routingKey = "";if (num % 5 == 0) {routingKey = "char.log.info";}if (num % 5 == 1) {routingKey = "char.log.warn";}if (num % 5 == 2) {routingKey = "char.log.error";} else {routingKey = "char.log.char";}System.out.println("routingKey --------------------> " + routingKey);this.amqpTemplate.convertAndSend("topic-exchange", routingKey, log);}}
  

  接收方

  yaml

  server:port: 8181spring:rabbitmq:host: 192.168.49.142port: 5672username: guestpassword: guestvirtual-host: /listener:direct:retry:enabled: truemax-attempts: 10
  

  pom

  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId>
  </dependency>
  <!-- AMQP插件组，用于开发spring boot访问符合AMQP协议的MQ产品的依赖启动器 -->
  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>
  

  consumer1

  /*** 消费所有级日志*/
  @RabbitListener(bindings = {@QueueBinding(value = @Queue(name = "topic-log-all", autoDelete = "false"),exchange = @Exchange(name = "topic-exchange", autoDelete = "false", type = "topic"),key = "*.log.*")
  })
  @Component
  public class All {@RabbitHandlerpublic void logHandler(String info) {System.out.println("all --------------------------->" + info);}}
  

  consumer2

  /*** 消费error级日志*/
  @RabbitListener(bindings = {@QueueBinding(value = @Queue(name = "topic-log-error", autoDelete = "false"),exchange = @Exchange(name = "topic-exchange", autoDelete = "false", type = "topic"),key = "*.log.error")
  })
  @Component
  public class Error {@RabbitHandlerpublic void logHandler(String error) {System.out.println("error --------------------------->" + error);}}
  

  consumer3

  …

  consumer4

  …

  注意事项：开发消息消费端是时候，消费方法不能带有返回值（也就是说@RabbitHandler描述的方法不具备返回值）