Rabbitmq学习

MQ

虽然同步调用时效性强，能够立即得到结果，但是存在以下问题：

• 耦合度高：每次加入新需求都要修改原来的代码
• 性能下降：如果调用链长，对服务性能影响非常大
• 资源浪费：调用链中每个服务在等待响应过程中，不能释放请求占用的资源，高并发场景下会极度浪费系统资源
• 级联失败：服务提供者出现问题，消费者也会跟着出现问题，最终整个微服务群瘫痪

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

当一个服务完成后需要调用其他服务，这个服务不需要主动调用其他服务，可以向事件代理者（Broker）发布事件，事件代理者就会通知相应服务执行相关业务，这个服务也不需要等待其他服务执行完直接返回结果给用户即可。

• 服务解耦
• 性能提升，吞吐量高
• 服务无强依赖，无级联问题
• 流量削峰：并发量高时，Broker可以起到缓冲的作用

缺点：依赖Broker的可靠性，安全性，吞吐能力，架构复杂，不好追踪管理

MQ（消息队列），也就是事件驱动架构中的Broker

RabbitMQ

基于Erlang（面向并发的编程语言）语言开发

安装

1. 拉取镜像

   docker pull rabbitmq:3-management
   

2. 运行容器

   docker run \
    -e RABBITMQ_DEFAULT_USER=root \
    -e RABBITMQ_DEFAULT_PASS=123321 \
    --name mq \
    --hostname mq1 \ #配置主机名，将来做集群必须要配，单机不配也可以
    -p 15672:15672 \ #管理平台ui端口
    -p 5672:5672 \ #消息通信端口
    -d \
    rabbitmq:3-management
    #-e：给mq设置环境变量
   

概念

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

消息模型

• 基本消息队列
  

• 工作消息队列：一个队列绑定多个消费者，提高消息处理速度，避免消息堆积
  

• 发布订阅：和前面的区别是允许将同一消息发送给多个队列，实现方式是加入了exchange交换机
  生产者不用管发送给哪个队列，由交换机处理，只用管将消息发送给交换机

  exchange只负责接收消息和消息路由，而不是存储，路由失败则消息丢失
  根据交换机类型又分为三种

1. 广播（Fanout Exchange）将接收到的消息路由到每个跟他绑定的queue
   

2. 路由（DirectExchange）将接受到的消息根据规则路由到指定queue

   • 每个queue都与交换机设置一个bindingkey可以是多个
   • 发布者发送消息时，指定消息的RoutingKey
   • 交换机将消息路由到BindingKey与消息RoutingKey一致的队列（RoutingKey只要与BindingKey中的一个匹配成功，就可以路由到这个队列）

   

3. 主题（TopicExchange）与DirectExchange相似，区别在于RoutingKey必须是多个单词的列表，并且以.分割
   Queue与Exchange指定BindingKey时可以使用通配符：#指0或多个单词 *：指一个单词

   

最基础的消息队列模型包含三个角色

• publisher
• queue：由rabbitmq管理
• consumer

SpringAMOP

一般交换机和队列定义在消费者端

简化rabbitmq复杂的api

AMQP（Advanced Message Queuing Protocol）：该协议与语言和平台无关，符合微服务中独立性的要求，用于在应用程序或之间传递业务消息的开放标准

Spring AMQP基于这个协议定义的一套api规范，提供模版来发送和接受消息

• spring-amqp：基础抽象
• spring-rabbit：底层默认实现

消息一旦消费就会从队列中删除，Rabbitmq没有消息回溯功能

消费逻辑的编写需要新建一个类

FanoutExchange

1. 消费者服务中，利用代码声明队列，交换机，并将两者绑定
   创建一个配置类完成队列交换机的声明和绑定（用bean去声明）
2. 消费者服务中，编写两个消费者方法，分别监听fanout.queue1和fanout.queue2
3. 在publish中编写测试方法，向交换机发送消息

DirectExchange

1. 利用@RabbitListener注解声明交换机，队列，RoutingKey
2. 消费者服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
3. 在publish中编写测试方法，向交换机发送消息

TopicExchange

1. 利用@RabbitListener注解声明交换机，队列，RoutingKey
2. 消费者服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2
3. 在publish中编写测试方法，向交换机发送消息

消息转换器

如果我们向消息队列中发送Java对象，spring对消息对象的处理是由MessageConverter来处理的，而默认实现是SimpleMessageConverter，会基于jdk的ObjectOutputStream完成序列化

我们只需要定义一个MessageConverter类型的Bean即可，推荐使用json序列化

引入依赖

<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-xml</artifactId>
</dependency>

声明MessageConverter

在启动类声明也可以，生产者和消费者的MessageConverter必须一致

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