标签：Exchange -- springcloud day4 绑定队列交换机消息 public

同步通信和异步通信

微服务间通讯有同步和异步两种方式：

同步通讯：就像打电话，需要实时响应。

异步通讯：就像发邮件，不需要马上回复。

同步通信

我们之前学习的Feign调用就属于同步方式，虽然调用可以实时得到结果，但存在下面的问题：

优点：

时效性较强，可以立即得到结果

缺点：

耦合度高
性能和吞吐能力下降
有额外的资源消耗
有级联失败问题

异步通信

异步通信在消息的发送者和接受者之间加入了一个中间人，发送者只需要将消息传递给中间人，然后接收者从中间人订阅消息，这种处理方式可以解除事件的发布者与接收者之间的耦合

Broker 是一个像数据总线一样的东西，所有的服务要接收数据和发送数据都发到这个总线上，这个总线就像协议一样，让服务间的通讯变得标准和可控。

优点：

吞吐量提升：无需等待订阅者处理完成，响应更快速
故障隔离：服务没有直接调用，不存在级联失败问题
调用间没有阻塞，不会造成无效的资源占用
耦合度极低，每个服务都可以灵活插拔，可替换
流量削峰：不管发布事件的流量波动多大，都由Broker接收，订阅者可以按照自己的速度去处理事件

缺点：

架构复杂了，业务没有明显的流程线，不好管理
需要依赖于Broker的可靠、安全、性能

常用的异步通信技术 MQ

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

几种常见MQ的对比：

	RabbitMQ	ActiveMQ	RocketMQ	Kafka
公司/社区	Rabbit	Apache	阿里	Apache
开发语言	Erlang	Java	Java	Scala&Java
协议支持	AMQP，XMPP，SMTP，STOMP	OpenWire,STOMP，REST,XMPP,AMQP	自定义协议	自定义协议
可用性	高	一般	高	高
单机吞吐量	一般	差	高	非常高
消息延迟	微秒级	毫秒级	毫秒级	毫秒以内
消息可靠性	高	一般	高	一般

追求可用性：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求消息低延迟：RabbitMQ、Kafka

RabbitMQ

MQ的基本结构：

RabbitMQ中的一些角色：

publisher：生产者
consumer：消费者
exchange个：交换机，负责消息路由
queue：队列，存储消息
virtualHost：虚拟主机，隔离不同租户的exchange、queue、消息的隔离

RabbitMQ消息模型

RabbitMQ官方提供了5个不同的Demo示例，对应了不同的消息模型：

使用SpringAMQP实现消息模型

直接使用RabbitMQ来实现消息的发送和接收步骤非常繁琐，我们可以直接使用SpringAMQP，其底层实现的还是RabbitMQ

使用SpringAMQP需要先引入依赖

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

声明对列和交换机

使用配置类声明

@Configuration
public class ExchangeConfig {

    //声明交换机，返回类型即为交换机的类型 fanout01 表示交换机的名称
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("fanout01");
    }
    //声明对列 queue01
    @Bean
    public Queue queue01(){
        return new Queue("fanout.queue01");
    }
    //绑定对列queue01 到交换机 FanoutExchange
    @Bean
    public Binding fanout01Binding01(FanoutExchange fanoutExchange,Queue queue01){
        return BindingBuilder.bind(queue01).to(fanoutExchange);
    }

使用注解声明

说明：

@RabbitListener ：声明消费者

bindings 属性：该消费者的绑定关系

value ：表示消费者绑定的对列信息

exchange：表示对列绑定的交换机信息

key：表示路由key

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "direct.queue02"),
        exchange = @Exchange(name = "direct",type = ExchangeTypes.DIRECT),
        key = {"red","yellow"}
))
public void listenDirectQueue02(String msg){
    System.out.println("directQueue02:"+msg);
}

基础模型

BasicQueue

消息发送：

配置MQ地址

spring:
  rabbitmq:
    host: localhost # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: admin # 用户名
    password: 123456 # 密码

使用RabbitTemplate发送消息

// 队列名称，需要先创建对列
String queueName = "queue001";
// 消息
String message = "hello, spring amqp!";
// 发送消息
rabbitTemplate.convertAndSend(queueName, message);

消息接收：

配置MQ地址

spring:
  rabbitmq:
    host: localhost # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: admin # 用户名
    password: 123456 # 密码

创建一个监听类

@Component
public class TestListen {
	//指定监听的对列名称
    @RabbitListener(queues = "queue001")
    public void listen(String msg){ //消息类型需要与发送的类型对应
        System.out.println(msg);
    }
}

启动springboot程序，即可接收 queue001 对列中的消息

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

WorkQueue

Work queues，也被称为（Task queues），任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。

当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。

此时就可以使用work 模型，多个消费者共同处理消息处理，速度就能大大提高了。

消息接收：使用两个消费者来监听同一个对列

@RabbitListener(queues = "queue001")
public void listen(String msg){
    System.out.println("消费者111："+msg);
}

@RabbitListener(queues = "queue001")
public void listen2(String msg){
    System.out.println("消费者222："+msg);
}

消息预存机制

消费者在获取对列中的消息时，会先将消息预存，再进行处理，而预存的消息数量默认是无上限的；此时，若一个消费者处理速度快，而另一个消费者处理速度慢，通过消息预存机制他们会分配到相同数量的消息，这显然是不合理的

我们可以通过配置预存的消息数量来解决这个问题

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

发布/订阅模型

发布订阅的模型如图：

可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化：

Publisher：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有以下3种类型：
- Fanout：广播，将消息交给所有绑定到交换机的队列
- Direct：定向，把消息交给符合指定routing key 的队列
- Topic：通配符，把消息交给符合routing pattern（路由模式）的队列
Consumer：消费者，与以前一样，订阅队列，没有变化
Queue：消息队列也与以前一样，接收消息、缓存消息。

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

Fanout

Fanout，英文翻译是扇出，我觉得在MQ中叫广播更合适。

Fanout 消息发送流程

1）可以有多个队列
2）每个队列都要绑定到Exchange（交换机）
3）生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定
4）交换机把消息发送给绑定过的所有队列
5）订阅队列的消费者都能拿到消息

Direct

在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

在Direct模型下：

队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
消息的发送方在向 Exchange发送消息时，也必须指定消息的 RoutingKey。
Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息

Topic

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：

#：匹配一个或多个词

*：匹配不多不少恰好1个词

举例：

item.#：能够匹配item.spu.insert 或者 item.spu

item.*：只能匹配item.spu

消息转换器

之前说过，Spring会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。

只不过，默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：

数据体积过大
有安全漏洞
可读性差

配置 json 转换器

使用JSON方式来做序列化和反序列化。

在publisher和consumer两个服务中都引入依赖：

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

配置消息转换器，使用 Jackson2JsonMessageConverter 覆盖默认的消息转换器

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

标签：Exchange,--,springcloud,day4,绑定,队列,交换机,消息,public
From： https://www.cnblogs.com/xxgxs/p/16810829.html

springcloud学习记录day4 -- 消息队列RubbitMQ