- 同步通讯
- 异步通讯
- 什么是MQ
- 一、RabbitMQ概述
- 二、安装
- 三、管理页面介绍
- 四、常见消息模型
- 基本消息队列（BasicQueue）：HelloWord案例
- 五、SpringAMQP
- 1、AMQP(Advanced Message Queuing Protocol)
- 2、Spring AMQP
- 工程中引入spring-amqp的依赖
- 配置YML文件
- 3、基本消息队列（BasicQueue）：HelloWord案例(SpringAMQP实现)
- 1、发送消息
- 2、接收消息
- 4、工作消息队列（WorkQueue）
- 1、Work Queue，工作队列，可以提高消息处理速度，避免队列消息堆积
- 2、Work模型：
- 消息预取机制：
- 消息预取限制：
- 3、广播（Fanout Exchange）
- 4、路由（Direct Exchange）
- 5、主题（Topic Exchange）
- 五、SpringAMQP-消息转换器
同步通讯
- 所谓的同步通信就是各个步骤都必须在同一时间内完成，所有操作是实时的，时效性较强
- 同步方式，存在一些问题：
- 耦合度高：
在我们开发过程中，总有需求的变动、需求的升级等问题，那如果是同步调用的方式，每次需求的变动或升级都需要在原来的代码上进行修改，这就完全违背了软件开发中的开闭原则 - 性能下降
调用者需要等待的服务提供者响应，调用链路过长则响应的时间等于调用时间之和
3.资源浪费
调用链中的每个服务在等待响应过程中，不能释放请求占用的资源，高并发场景下会极度浪费资源 - 级联失败
如果服务提供者出现问题，整个调用链路都会受牵连跟着出问题，如同多米诺骨牌一样，导致整个微服务集群故障。
异步调用常见实现就是事件驱动模式
异步通讯的各步骤可以不在同一时间内完成，操作不是实时的，时效性较低，依赖于Broker的可靠性，安全性，吞吐能力，加构复杂，业务没有明显的流程先；
异步通讯的优势：
- 服务解耦
以支付模块来说，如果使用同步方式，需要在同一时间完成与当前订单的所有的相关业务。
而异步方式，支付服务只需要关注本身的业务，支付成功后向Broker发布支付成功事件，而其他相关业务就可以在约定的时间或达到某个条件后再执行后续业务 - 性能提升，吞吐量提高
以支付模块来说，使用异步只需要本身的支付服务完成后就可返回结果，其他业务不在此时完成，从而极大的减少服务调用链路的时间
3.服务之间没有强依赖，不担心级联失败的问题
以支付模块来说，支付成功后相当于当前支付已完成，至于其他相关业务或服务在后续执行，不影响当前支付服务的运行状态（解耦带来的作用）
4.流量削峰
随着业务量增加，请求量会变大，Broker可以起到一个缓存的作用，SpringBoot可以根据服务器处理量从Broker中一个一个的拿出请求处理。将一瞬间的大量请求分摊到一段较长的时间去处理，以达到削峰
MQ（MessageQueue）,翻译为 消息队列，字面看来就是存放消息的队列，也就是事件驱动中的Broker
RabbitMQ是基于Erlang语言开发的开源消息通信中间件，官网地址:点击跳转 https://rabbitmq.com
二、安装- 在安装docker的linux上使用如下命令拉取RabbitMq镜像
docker pull rabbitmq:3-management
- 使用如下命令基于RabbitMQ镜像启动容器
docker run #启动命令 -e RABBITMQ_DEFAULT_USER=itcast #设置一个用户名 -e RABBITMQ_DEFAULT_PASS=123321 #设置密码 --name mq #容器名称 --hostname mq1 #集群启动时需要配置的 服务名称 -p 15672:15672 #管理端口映射 -p 5672:5672 #连接端口映射 -d #后台启动 rabbitmq:3-management #镜像名称三、管理页面介绍
- Overview：节点详细信息
- Connections：连接
- Channels：通道(操作MQ的工具)
- Exchanges：交换机（路由消息到队列中）
- Queues：队列（缓存消息）
- Admin：管理（其中的 Virtual Hosts 虚拟主机，是队列（queue）、交换机（exchange）等资源的逻辑分组 ）
MQ的官方文档中给出了5个MQ的Demo示例，对应了几种不同的用法：文档地址
基本消息队列（BasicQueue）：HelloWord案例官方的HelloWord是基于最基础的消息队列模型来实现的，只包括三个角色
- publisher：消息发布者，将消息发送到队列 queue
- queue：消息队列，负责接收并缓存消息
- consume：订阅队列，处理队列中的消息
- 基本消息队列的消息发送流程：
public void testSendMessage() throws IOException, TimeoutException {
// 1.建立连接,建立Connerection
ConnectionFactory factory = new ConnectionFactory();
// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
factory.setHost("192.168.157.130");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("itcast");
factory.setPassword("123321");
// 1.2.建立连接
Connection connection = factory.newConnection();
// 2.创建通道 Channel
Channel channel = connection.createChannel();
// 3.创建队列
String queueName = "simple.queue";
channel.queueDeclare(queueName, false, false, false, null);
// 4.发送消息
String message = "hello, rabbitmq!";
channel.basicPublish("", queueName, null, message.getBytes());
System.out.println("发送消息成功：【" + message + "】");
// 5.关闭通道和连接
channel.close();
connection.close();
}
- 基本消息队列的消息接收流程：
public static void main(String[] args) throws IOException, TimeoutException {
// 1.建立连接,建立Connerection
ConnectionFactory factory = new ConnectionFactory();
// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
factory.setHost("192.168.157.130");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("itcast");
factory.setPassword("123321");
// 1.2.建立连接
Connection connection = factory.newConnection();
// 2.创建通道 Channel
Channel channel = connection.createChannel();
// 3.创建队列
String queueName = "simple.queue";
channel.queueDeclare(queueName, false, false, false, null);
// 4.订阅消息，定义consumer的消费行为 handleDelivery(),利用channel将消费者与队列绑定
channel.basicConsume(queueName, true, new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
// 5.处理消息
String message = new String(body);
System.out.println("接收到消息：【" + message + "】");
}
});
System.out.println("等待接收消息。。。。");
}
五、SpringAMQP
1、AMQP(Advanced Message Queuing Protocol)
一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有RabbitMQ等。
2、Spring AMQPSpring AMQP是基于AMQP写一定义的一套API规范，提供了模板来接收和发送消息。
-
其中 spring-amqp是基础抽象
-
spring-rabbit是底层的默认是实现
特征： -
Listener container:用于异步处理入站消息的
-
RabbitTemplate:用于发送和接收消息的
-
RabbitAdmin:用于自动声明队列、交换和绑定
org.springframework.boot
spring-boot-starter-amqp
配置YML文件
如果是微服务架构，则需要在每个服务都进行此配置
spring:
rabbitmq:
host: 192.168.157.130 #rabbitMq的ip地址
port: 5672 #端口
username: itcast #用户名
password: 123321 #密码
virtual-host: / #虚拟主机
3、基本消息队列（BasicQueue）：HelloWord案例(SpringAMQP实现)
1、发送消息
使用RabbitTemplate 发送
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSendMSG() {
String queueName = "simple.queue";
String message = "hello, rabbitmq!";
rabbitTemplate.convertAndSend(queueName, message);
}
2、接收消息
@RabbitListener 注解 queues:队列名称
@Component
public class SpringRabbitListener {
@RabbitListener(queues = "simple.queue")
public void listSimpleQueue(String msg) {
System.out.println(msg);
}
}
4、工作消息队列（WorkQueue）
1、Work Queue，工作队列，可以提高消息处理速度，避免队列消息堆积
2、Work模型：
- 多个消费者绑定一个队列，一个消息只会被一个消费者处理
- 读取同一队列的不同消费者拿无视处理消息的性能各自拿去全部消息的平均值。
修改yml配置文件，这只preFetch这个值，可以控制预取消息的上线
spring:
rabbitmq:
listener:
simple:
prefetch: 1 #消息预取限制
发布（Publish），订阅（Subscribe）
- 发布订阅模式与其他模式的区别为：允许将同一消息发送给多个消费者。实现方式是加入了又Exchange（交换机）
- 接收pubilsher发送的消息
- 将消息按照规则路由到与至绑定的队列
- 注意： Exchange负责消息路由，而不是存储，路由失败则消息丢失
根据交换机类型不同分为三种：
- Fanout Exchange会将接收到的消息路由到每一个跟其绑定的queue(队列)
- 在消费者服务中，声明队列（Queue），交换机(Exchange)，并将两者绑定
- springAMQP提供了声明交换机，队列，绑定关系的API，例如：
- 此处为bean声明方式，在下面有通过**@RabbitListener**声明
@Configuration
public class FanoutConfig {
@Bean
public FanoutExchange fanoutExchange(){
return new FanoutExchange("itcast.fanout");
}
@Bean
public Queue fanoutQueue1(){
return new Queue("fanout.queue1");
}
@Bean
public Binding fanoutBinding1(Queue fanoutQueue1,FanoutExchange fanoutExchange){
return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
}
- 在消费者服务中，编写两个消费者方法，分别监听 fanout.queue1、fanout.queue2
@RabbitListener(queues = "fanout.queue1")
public void listFanoutQueue1(String msg) {
System.out.println("消费者接收到fanout.queue1"+msg);
}
@RabbitListener(queues = "fanout.queue2")
public void listFanoutQueue2(String msg) {
System.out.println("消费者接收到fanout.queue2"+msg);
}
- 在生产者服务中 向创建好的Fanout Exchange（广播）发送消息
@Test
public void testSendMSGFanoutExchange(){
//交换机名称
String exchangeName = "itcast.fanout";
//消息
String message = "hello, every one!";
//发送消息
rabbitTemplate.convertAndSend(exchangeName,"", message);
}
4、路由（Direct Exchange）
- 会将接收到的消息根据规则路由到指定的Queue,因此称为路由模式（routes）.
- 每一个Queue(队列)都与Exchange设置一个BindingKey
- 发布者发送消息时，指定消息的RoutingKey
- Exchange将消息路由到BindingKey与消息RoutingKey一直的队列
- 在消费者服务中，编写消费者方法，监听Queue(队列)
- @RabbitListener声明Exchange,Queue,RoutingKey
@RabbitListener(bindings = @QueueBinding( //声明队列并绑定路由并监听，设置RoutingKey
value = @Queue(name = "direct.queue2"), //声明队列
//声明交换机itcast.direct 交换机类型为：ExchangeTypes.DIRECT（路由）
exchange = @Exchange(name ="itcast.direct",type = ExchangeTypes.DIRECT),
key = {"red","yellow"})) //设置RoutingKey（可以有多个）。
public void listenDirectQueue2(String msg) {
System.out.println("消费者接收到"+msg);
}
- 在生产者中编写方法，向创建好的 Direct Exchange （路由）发送消息
public void testSendMSGDirectExchange(){
//交换机名称
String exchangeName = "itcast.direct";
//消息
String message = "hello";
//发送消息,red为RoutingKey
rabbitTemplate.convertAndSend(exchangeName,"red", message);
//发送消息,yellow为RoutingKey
rabbitTemplate.convertAndSend(exchangeName,"yellow", message);
}
5、主题（Topic Exchange）
-
Topic Exchange与Direct Exchange类似，区别在于Topic Exchange的RoutingKey必须是多个单词的列表，并且以 “ , ”(逗号) 分割
-
队列与交换机指定BindingKey是可以使用通配符：
#(井号) ：代表0个或多个单词
*(星号) ：代指一个单词
- china.news：代表中国的新闻消息
- china.weather：代表中国的天气消息
- USA.news：代表美国的新闻消息
- USA.weather：代表美国的天气消息
- 在消费者服务中，编写消费者方法，监听Queue(队列)
@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);
}
- 在生产者中编写方法，向创建好的 Topic Exchange （主题）发送消息
@Test
public void testSendMSGTopicExchange(){
//交换机名称
String exchangeName = "itcast.topic";
//消息
String message = "hello, blue!";
//发送消息，此处 china.news 与两个消费者的BindingKey都匹配，所以两个消费者都会接收到此消息
rabbitTemplate.convertAndSend(exchangeName,"china.news", message);
}
五、SpringAMQP-消息转换器
在 SpringAMQP 的发送方法中，接收消息的类型是Object，也就是说我们可以发送任意对象类型的消息，SpringAMQP会帮我们序列化为字节后发送。
- Spring的对消息对象的处理是由MessageConverter处理的，而默认实现是SimpleMessageConverter,基于JDK的ObjectOutPutStream完成序列化，这样会使消息序 列化后 复杂化
- 注意发送方与接收方必须使用相同的MessageConverter
SimpleMessageConverter序列化效果：
- 如果需要修改，我们要重新定义一个MessageConverter类型的Bean即可，推荐使用JSON方式序列化，步骤如下：
- 在需要修改的服务中导入依赖
com.fasterxml.jackson.dataformat jackson-dataformat-xml
- 在生产者服务中声明MessageConverter
@Configuration
public class MessageConverterConfig {
@Bean
public MessageConverter jsonMessageConverter(){
return new Jackson2JsonMessageConverter();
}
}
- 修改后的效果
发送Object消息，例：
4. 声明一个队列（只要声明即可，不限制声明方式）
@Bean
public Queue ObjectQueue(){
return new Queue("object.queue");
}
- 发送一条Map类型消息
public void testSendMSGObject(){
//消息
Map map = new HashMap<>();
map.put("name","张三");
map.put("age",18);
//发送消息
rabbitTemplate.convertAndSend("object.queue",map);
}
- 监听此队列
@RabbitListener(queues="object.queue")
public void objectqueue(Map map) {
System.out.println("消费者objectqueue接收到:"+map);
}
- 接收到的消息： ”消费者objectqueue接收到:{name=张三, age=18}“
