RabbitMQ的基本概念以及绑定策略的简单演示

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

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描述的方法不具备返回值）