MQ

同步调用的问题

微服务间基于Feign的调用就属于同步方式，存在一些问题。

• 耦合度高：每次加入新的需求，都要修改原来的代码
• 性能下降：调用者需要等待服务提供者响应，如果调用链过长则响应时间等于每次调用的时间之和。
• 资源浪费：调用链中的每个服务在等待响应过程中，不能释放请求占用的资源，高并发场景下会极度浪费系统资源
• 级联失败：如果服务提供者出现问题，所有调用方都会跟着出问题，如同多米诺骨牌一样，迅速导致整个微服务群故障

异步调用方案

异步调用常见实现就是事件驱动模式

• 耦合度低
• 吞吐量提升
• 故障隔离
• 流量削峰

异步通信的缺点：

• 依赖于Broker的可靠性、安全性、吞吐能力
• 架构复杂了，业务没有明显的流程线，不好追踪管理

什么是MQ

MQ （MessageQueue），中文是消息队列，字面来看就是存放消息的队列。也就是事件驱动架构中的Broker。

a.安装RabbitMQ

1.在线拉取

docker pull rabbitmq:3-management

2.执行下面的命令来运行MQ容器：

docker run \-e RABBITMQ_DEFAULT_USER=root \-e RABBITMQ_DEFAULT_PASS=root \--name mq \--hostname mq1 \-p 15672:15672 \-p 5672:5672 \-d \rabbitmq:3-management

15672 控制台端口号

5672 消息通信端口

RabbitMQ中的几个概念：

• channel：操作MQ的工具
• exchange：路由消息到队列中
• queue：缓存消息
• virtual host：虚拟主机，是对queue、exchange等资源的逻辑分组

b.消息模型

MQ的官方文档中给出了5个MQ的Demo示例，对应了几种不同的用法：

• 基本消息队列（BasicQueue）
• 工作消息队列（WorkQueue）
• 发布订阅（Publish、Subscribe），又根据交换机类型不同分为三种：
  • Fanout Exchange：广播
  • Direct Exchange：路由
  • Topic Exchange：主题

官方的HelloWorld是基于最基础的消息队列模型来实现的，只包括三个角色：

• publisher：消息发布者，将消息发送到队列
• queuequeue：消息队列，负责接受并缓存消息
• consumer：订阅队列，处理队列中的消息

基本消息队列的消息发送流程：

1.建立connection

2.创建channel

3.利用channel声明队列

4.利用channel向队列发送消息

基本消息队列的消息接收流程：

1.建立connection

2.创建channel

3.利用channel声明队列 (避免队列不存在，所以双方都声明)

4.定义consumer的消费行为handleDelivery()

5.利用channel将消费者与队列绑定

c.SpringAMQP发送和接收

案例：利用SpringAMQP实现HelloWorld中的基础消息队列功能

流程如下：

1.在父工程中引入spring-amqp的依赖

<!--AMQP依赖，包含RabbitMQ-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2.在publisher中编写测试方法，向simple.queue发送消息

• 在publisher服务中编写application.yml，添加mq连接信息：

spring:rabbitmq:host: 192.168.58.128 # rabbitMq的ip地址port: 5672 # 端口username: rootpassword: rootvirtual-host: /

• 在publisher服务中新建一个测试类，编写测试方法：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testSendMessage2SimpleQueue(){String queueName = "simple.queue";String message = "Hello, spring amqp!";rabbitTemplate.convertAndSend(queueName, message);}
}

3.在consumer中编写消费逻辑，监听simple.queue

• 在consumer服务中编写application.yml，添加mq连接信息：

spring:rabbitmq:host: 192.168.58.128port: 5672username: rootpassword: rootvirtual-host: /

• 在consumer服务中新建一个类，编写消费逻辑 (添加监听注解**@RabbitListener**)：

@Component
public class SpringRabbitListener {@RabbitListener(queues = {"simple.queue"})public void listenSimpleQueue(String msg){System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");}
}

d.WorkQueue模型

Work queue，工作队列，可以提高消息处理速度，避免队列消息堆积

案例：模拟WorkQueue，实现一个队列绑定多个消费者

1.在publisher服务中定义测试方法，每秒产生50条消息，发送到simple.queue

@Test
public void testSendMessage2WorkQueue() throws Exception {String queueName = "simple.queue";String message = "Hello, message__";for (int i = 1; i <= 50; i++) {rabbitTemplate.convertAndSend(queueName, message + i);Thread.sleep(20);}
}

2.在consumer服务中定义两个消息监听者，都监听simple.queue队列

3.消费者1每秒处理50条消息，消费者2每秒处理10条消息

@Component
public class SpringRabbitListener {@RabbitListener(queues = {"simple.queue"})public void listenWorkQueue1(String msg) throws Exception {System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());Thread.sleep(20);}@RabbitListener(queues = {"simple.queue"})public void listenWorkQueue2(String msg) throws Exception {System.out.println("消费者2...接收到消息：【" + msg + "】" + LocalTime.now());Thread.sleep(200);}
}

消费预取限制

修改application.yml文件，设置preFetch这个值，可以控制预取消息的上限：

spring:rabbitmq:host: 192.168.58.128port: 5672username: rootpassword: rootvirtual-host: /listener:simple:prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

可以提高整个消息队列的速度，避免消息堆积

e.发布订阅模型

发布（ Publish ）、订阅（ Subscribe ）

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。实现方式是加入了exchange（交换机）。

常见exchange类型包括：

• Fanout：广播
• Direct：路由
• Topic：话题

注意：exchange负责消息路由，而不是存储，路由失败则消息丢失

1) FanoutExchange

Fanout Exchange 会将接收到的消息路由给每一个跟其绑定的queue

案例：利用SpringAMQP演示FanoutExchange的使用

1.在consumer服务中，利用代码声明队列、交换机，并将两者绑定

在consumer服务常见一个类，添加@Configuration注解，并声明FanoutExchange、Queue和绑定关系对象Binding

@Configuration
public class FanoutConfig {// abcd.fanout@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("abcd.fanout");}// fanout.queue1@Beanpublic Queue fanoutQueue1(){return new Queue("fanout.queue1");}// 绑定队列1到交换机@Beanpublic Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}// fanout.queue2@Beanpublic Queue fanoutQueue2(){return new Queue("fanout.queue2");}// 绑定队列2到交换机@Beanpublic Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}

2.在consumer服务中，编写两个消费者方法，分别监听fanout.queue1和fanout.queue2

@Component
public class SpringRabbitListener {@RabbitListener(queues = {"fanout.queue1"})public void listenFanoutQueue1(String msg){System.out.println("消费者接收到fanout.queue1的消息：【" + msg + "】");}@RabbitListener(queues = {"fanout.queue2"})public void listenFanoutQueue2(String msg){System.out.println("消费者接收到fanout.queue2的消息：【" + msg + "】");}
}

3.在publisher中编写测试方法，向abcd.fanout发送消息

@Test
public void testSendFanoutExchange() {// 交换机名称String exchangeName = "abcd.fanout";//消息String message = "Hello, every one!";// 发送消息: 参数依次为：交换机名称，RoutingKey，消息rabbitTemplate.convertAndSend(exchangeName, "", message);
}

2) DirectExchange

Direct Exchange 会将接收到的消息根据规则路由到指定的Queue，因此称为路由模式（routes）。

• 每一个Queue都与Exchange设置一个BindingKey
• 发布者发送消息时，指定消息的RoutingKey
• Exchange将消息路由到BindingKey与消息RoutingKey一致的队列

案例：利用SpringAMQP演示DirectExchange的使用

1.利用@RabbitListener声明Exchange、Queue、RoutingKey

2.在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2

@Component
public class SpringRabbitListener {@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue1"),exchange = @Exchange(name = "abcd.direct", type = ExchangeTypes.DIRECT),key = {"red", "blue"}))public void listenDirectQueue1(String msg){System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】");}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue2"),exchange = @Exchange(name = "abcd.direct", type = ExchangeTypes.DIRECT),key = {"red", "yellow"}))public void listenDirectQueue2(String msg){System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】");}
}

3.在publisher中编写测试方法，向abcd. direct发送消息

@Test
public void testSendDirectExchange(){// 交换机名称String exchangeName = "abcd.direct";// 消息String message = "Hello, blue!";// 发送消息，参数依次为：交换机名称，RoutingKey，消息rabbitTemplate.convertAndSend(exchangeName, "red", message);
}

3) TopicExchange

TopicExchange与DirectExchange类似，区别在于routingKey必须是多个单词的列表，并且以. 分割。Queue与Exchange指定BindingKey时可以使用通配符：

• #：代指0个或多个单词
• *：代指一个单词

案例：利用SpringAMQP演示TopicExchange的使用

1.利用@RabbitListener声明Exchange、Queue、RoutingKey

2.在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2

@Component
public class SpringRabbitListener {@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue1"),exchange = @Exchange(name = "abcd.topic", type = ExchangeTypes.TOPIC),key = {"china.#"}))public void listenTopicQueue1(String msg){System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】");}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue2"),exchange = @Exchange(name = "abcd.topic", type = ExchangeTypes.TOPIC),key = {"*.news"}))public void listenTopicQueue2(String msg){System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】");}
}

3.在publisher中编写测试方法，向abcd. topic发送消息

@Test
public void testSendTopicExchange(){// 交换机名称String exchangeName = "abcd.topic";// 消息String message = "Hello, new!";// 发送消息，参数依次为：交换机名称，RoutingKey，消息rabbitTemplate.convertAndSend(exchangeName, "hello.news", message);
}

f.消息转换器

在SpringAMQP的发送方法中，接收消息的类型是Object，也就是说我们可以发送任意对象类型的消息，SpringAMQP会帮我们序列化为字节后发送。

案列：测试发送Object类型消息

1.在consumer中利用@Bean声明一个队列：

@Configuration
public class FanoutConfig {@Beanpublic Queue objectQueue() {return new Queue("object.queue");}
}

2.在publisher中发送消息以测试：

@Test
public void testSendObjectQueue(){Map<String, Object> msg = new HashMap<>();msg.put("name", "张三");msg.put("age", 21);rabbitTemplate.convertAndSend("object.queue", msg);
}

消息转换器

Spring的对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。如果要修改只需要定义一个MessageConverter 类型的Bean即可。推荐用JSON方式序列化

1.在publisher服务引入依赖

<dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId>
</dependency>

2.在publisher服务声明MessageConverter：

@Bean
public MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();
}

3.在consumer服务引入Jackson依赖：

• 与步骤1一样

4.在consumer服务定义MessageConverter：

• 与步骤2一样

5.定义一个消费者，监听object.queue队列并消费消息：

@RabbitListener(queues = "object.queue")
public void listenObjectQueue(Map<String, Object> msg){System.out.println("接收到object.queue的消息：" + msg);
}