同步通信和异步通信

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

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

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

同步通信

我们之前学习的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

消息发送：

  1. 配置MQ地址
spring:
  rabbitmq:
    host: localhost # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: admin # 用户名
    password: 123456 # 密码
  1. 使用RabbitTemplate发送消息
// 队列名称，需要先创建对列
String queueName = "queue001";
// 消息
String message = "hello, spring amqp!";
// 发送消息
rabbitTemplate.convertAndSend(queueName, message);

消息接收：

  1. 配置MQ地址
spring:
  rabbitmq:
    host: localhost # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: admin # 用户名
    password: 123456 # 密码
  1. 创建一个监听类
@Component
public class TestListen {
	//指定监听的对列名称
    @RabbitListener(queues = "queue001")
    public void listen(String msg){ //消息类型需要与发送的类型对应
        System.out.println(msg);
    }
}
  1. 启动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();
}

原文地址：http://www.cnblogs.com/xxgxs/p/16810829.html

1. 本站所有资源来源于用户上传和网络，如有侵权请邮件联系站长！ 2. 分享目的仅供大家学习和交流，请务用于商业用途! 3. 如果你也有好源码或者教程，可以到用户中心发布，分享有积分奖励和额外收入！ 4. 本站提供的源码、模板、插件等等其他资源，都不包含技术服务请大家谅解！ 5. 如有链接无法下载、失效或广告，请联系管理员处理！ 6. 本站资源售价只是赞助，收取费用仅维持本站的日常运营所需！ 7. 如遇到加密压缩包，默认解压密码为"gltf",如遇到无法解压的请联系管理员！ 8. 因为资源和程序源码均为可复制品，所以不支持任何理由的退款兑现，请斟酌后支付下载 声明：如果标题没有注明"已测试"或者"测试可用"等字样的资源源码均未经过站长测试.特别注意没有标注的源码不保证任何可用性