MQ
1. 初始MQ
异步通讯的优缺点
优点
耦合度低吞吐量提升故障隔离流量削峰
缺点
依赖于Broker的可靠性、安全性、吞吐能力架构复杂了，业务没有明显的流程线，不好追踪管理
什么是MQ?
2. RabbitMQ快速入门
RabbitMQ概述
channel:操作MQ的工具exchange:路由信息到队列中queue:缓存消息virtual host:虚拟主机，是对queue、exchange等资源的逻辑分组
消息模型介绍
简单队列模型
3. SpringAMQP
3.1 什么是SpringAMQP
SpringAmqp的官网地址: https://spring.io/projects/spring-amqp
AMQP: 是用于在应用程序或之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求。
Spring AMQP: 是基于AMQP协议定义的一套API规范，提供了模板来发送和接收消息。包含两部分，其中spring-amqp是基础抽象，spring-rabbit是底层的默认实现
3.2 入门案例的消息发送和消息接收
引入AMQP依赖
在Publisher服务中编写application.yml,添加mq连接消息：
spring:
rabbitmq:
host: 192.168.198.138 # 端口
port: 5672 # 端口
username: itcast # 用户名
password: 123321 # 密码
virtual-host: / # 虚拟主机
在publisher服务中新建一个测试类，编写测试方法
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSendMessage2SimpleQueue() {
String queueName = "simple.queue";
String message = "hello,Spring amqp!";
rabbitTemplate.convertAndSend(queueName,message);
}
}
在consumer服务中编写application.yml，添加mq连接信息
spring:
rabbitmq:
host: 192.168.198.138 # 端口
port: 5672 # 端口
username: itcast # 用户名
password: 123321 # 密码
virtual-host: / # 虚拟主机
在consumer服务中新建一个类，编写消费逻辑
@Component
public class SpringRabbitListener {
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(String msg) {
System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
}
}
3.3 Work Queue 工作队列
作用:可以提高消息处理速度，避免队列消息堆积同时发送50条消息，我们想让处理快的消费者多去处理数据，但默认的情况下是，所有的消费者平分数据，这会拖慢我们的处理速度
解决方法：消费预取限制（这样）
spring:
rabbitmq:
host: 192.168.198.138
port: 5672
username: itcast
password: 123321
virtual-host: /
listener:
simple:
prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个信息
3.4 发布订阅模型介绍
交换机的作用
接收publisher发送的消息将消息按照规则路由到与之绑定的队列不能缓存消息，路由失败，消息丢失FanoutExchange的会将消息路由到每个绑定的队列
3.4.1 FanoutExchange
案例实现
在consumer服务常见一个类，添加@Configuration注解，并声明FanoutExchange、Queue和绑定关系对象Binding
@Configuration
public class FanoutConfig {
// itcast.fanout
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange("itcast.fanout");
}
// fanout.queue1
@Bean
public Queue fanoutQueue1() {
return new Queue("fanout.queue1");
}
// 绑定队列1到交换机
@Bean
public Binding fanoutBinging1(Queue fanoutQueue1,FanoutExchange fanoutExchange) {
return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
}
// fanout.queue2
@Bean
public Queue fanoutQueue2() {
return new Queue("fanout.queue2");
}
// 绑定队列2到交换机
@Bean
public Binding fanoutBinging2(Queue fanoutQueue2,FanoutExchange fanoutExchange) {
return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
}
}
在consumer服务的SpringRabbitListener类中，添加两个方法，分别监听fanout.queue1和fanout.queue2
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 + "】");
}
在publisher服务的SpringAmqpTest类中添加测试方法
@Test
public void testSendFanoutExchange() {
// 交换机名称
String exchangeName = "itcast.fanout";
// 消息
String message = "hello,every one!";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName,"",message);
}
测试结果
3.4.2 DirectExchange
案例实现
在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
并利用@RabbitListener声明Exchange、Queue、Routingkey
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue1"),
exchange = @Exchange(name = "itcast.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 = "itcast.direct", type = ExchangeTypes.DIRECT),
key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg) {
System.out.println("消费者接收到direct.queue2的消息【" + msg + "】");
}
在publisher服务的SpringAmqpTest类中添加测试方法
@Test
public void testSendDirectExchange() {
// 交换机名称
String exchangeName = "itcast.direct";
// 消息
String message = "hello,blue!";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName,"blue",message);
}
3.4.3 TopicExchange
案例实现
在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2
并利用@RabbitListener声明Exchange、Queue、RoutingKey
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue1"),
exchange = @Exchange(name = "itcast.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 = "itcast.topic",type = ExchangeTypes.TOPIC),
key = "#.news"
))
public void listenTopicQueue2(String msg) {
System.out.println("消费者接收到topic.queue2的消息【" + msg + "】");
}
在publisher服务的SpringAmqpTest类中添加测试方法
@Test
public void testSendTopicExchange() {
// 交换机名称
String exchangeName = "itcast.topic";
// 消息
String message = "今天天气不错，我的心情好极了";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName,"china.weather",message);
}
3.5 消息转换器