07MQ基础

1.同步通讯与异步通讯

  • 同步通讯：就如同打视频电话，双方的交互都是实时的。因此同一时刻你只能跟一个人打视频电话。
  • 异步通讯：就如同发微信聊天，双方的交互不是实时的，你不需要立刻给对方回应。因此你可以多线操作，同时跟多人聊天。

1.1同步通信与其问题

同步调用的方式我们已经学过了，之前的OpenFeign调用就是，那么这种方式存在哪些问题呢？

以余额支付功能为例

业务执行流程是这样的：

  • 支付服务需要先调用用户服务完成余额扣减
  • 然后支付服务自己要更新支付流水单的状态
  • 然后支付服务调用交易服务，更新业务订单状态为已支付

这其中就存在3个问题：
1.拓展性差，如果我要在此基础上添加“短信通知”、“积分添加”等功能，这个业务模块就需要继续添加，最终会很臃肿

2.性能下降，OpenFeign是同步通讯，会一步步执行全流程（支付、用户、交易、通知、积分、……）时间太长，性能会很差

3.级联失败，这个分布式事务，采用了seata来保证事务一致性，如果一处失败，全部都回滚。但是现在支付成功、交易成功，但是通知失败，我们需要回滚吗？很明显不需要，我就是通知失败，我不想把用户支付的钱再还给用户，通知失败可以以后再想办法通知就行了。

1.2使用异步通讯解决同步通信的问题

异步调用方式其实就是基于消息通知的方式，一般包含三个角色：

  • 消息发送者：投递消息的人，就是原来的调用方
  • 消息Broker：管理、暂存、转发消息，你可以把它理解成微信服务器
  • 消息接收者：接收和处理消息的人，就是原来的服务提供方

现在我们把支付消息放到Broker中，然后把不需要事务强一致性的部分从同步通信改成异步通信，这样就能防止这部分调用占用大量时间：

2.RabbitMQ

实现异步通信，需要使用Message Queue（消息队列，就是上面那样的东西），简称MQ，现在主流MQ是RabbitMQ

2.1创建RabbitMQ

在宝塔终端创建RabbitMQ容器：

docker run \
 -e RABBITMQ_DEFAULT_USER=itheima \
 -e RABBITMQ_DEFAULT_PASS=123321 \
 -v mq-plugins:/plugins \
 --name mq \
 --hostname mq \
 -p 15672:15672 \
 -p 5672:5672 \
 --network hm-net\
 -d \
 rabbitmq:3.8-management

可以看到在安装命令中有两个映射的端口：

  • 15672：RabbitMQ提供的管理控制台的端口
  • 5672：RabbitMQ的消息发送处理接口

访问http://你的宝塔地址（也就是你的虚拟机地址）:15672

RabbitMQ对应的架构如图：

其中包含几个概念：

  • publisher：生产者，也就是发送消息的一方
  • consumer：消费者，也就是消费消息的一方
  • queue：队列，存储消息。生产者投递的消息会暂存在消息队列中，等待消费者处理
  • exchange：交换机，负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。
  • virtual host：虚拟主机，起到数据隔离的作用。每个虚拟主机相互独立，有各自的exchange、queue

MQ入门-06.RabbitMQ-快速入门_哔哩哔哩_bilibili 这个是RabbitMQ图形胡界面的基本使用方法介绍

2.2交换机类型

Fanout交换机，广播类型，发送消息到所有连接到Fanout交换机的队列

Direct交换机，有选择地发送消息，队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKey
比如说我的队列A绑定red、yellow两个key，队列B绑定red、blue两个key。如果我使用direct交换机发送消息，规定只能发送到绑定了red这个key的队列，AB都能接收到，但是只发送到绑定了yellow这个key的队列，只有A接收到。

Topic交换机，类似Direct交换机，但是可以在确定key时使用通配符，通配符规则：

  • #：匹配一个或多个词
  • *：匹配不多不少恰好1个词

举例：
item.*：只能匹配item.spu这个key
item.#：能够匹配item.spu.insert 或者 item.spu这两个key

3.SpringAMQP

我们不能只通过图形化界面进行发送消息、接收消息。我们需要在spring项目中使用Java代码发送消息、接收消息，SpringAMQP较为简单地实现了这个功能。

SpringAMQP提供了三个功能：

  • 自动声明队列、交换机及其绑定关系（创建队列）
  • 基于注解的监听器模式，异步接收消息（接收消息）
  • 封装了RabbitTemplate工具，用于发送消息（发送消息）

我们以现在的pay-service（发送消息）和trade-service（接收消息，更新订单状态）为例，现在改造成异步更新订单

3.1配置MQ

不管是生产者（pay-service）还是消费者(trade-service)，都需要配置MQ的基本信息。分为两步：

1）添加依赖：

  <!--消息发送-->
  <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>

2）配置MQ地址：

然后在pay-service和trade-service的bootstrap中添加

          – dataId: shared-jdbc.yaml # 共享mybatis配置

          – dataId: shared-log.yaml # 共享日志配置

          – dataId: shared-swagger.yaml # 共享日志配置

          – dataId: shared-seata.yaml # 共享seata配置

          – dataId: rabbitMQ.yaml # 共享rabbitmq配置（这个是新增的MQ配置）

3.2发送消息

修改pay-service服务下的com.hmall.pay.service.impl.PayOrderServiceImpl类中的tryPayOrderByBalance方法：添加一个rabbitTemplate，然后不再使用tradeClient，而是使用rabbitTemplate修改订单状态，发送消息使用convertAndSend，传入三个参数：交换机名称，key，消息

private final RabbitTemplate rabbitTemplate;

@Override
@Transactional
public void tryPayOrderByBalance(PayOrderDTO payOrderDTO) {
    // 1.查询支付单
    PayOrder po = getById(payOrderDTO.getId());
    // 2.判断状态
    if(!PayStatus.WAIT_BUYER_PAY.equalsValue(po.getStatus())){
        // 订单不是未支付，状态异常
        throw new BizIllegalException("交易已支付或关闭！");
    }
    // 3.尝试扣减余额
    userClient.deductMoney(payOrderDTO.getPw(), po.getAmount());
    // 4.修改支付单状态
    boolean success = markPayOrderSuccess(payOrderDTO.getId(), LocalDateTime.now());
    if (!success) {
        throw new BizIllegalException("交易已支付或关闭！");
    }
    // 5.修改订单状态
    // tradeClient.markOrderPaySuccess(po.getBizOrderNo());
    try {
        rabbitTemplate.convertAndSend("pay.direct", "pay.success", po.getBizOrderNo());
    } catch (Exception e) {
        log.error("支付成功的消息发送失败，支付单id：{}， 交易单id：{}", po.getId(), po.getBizOrderNo(), e);
    }
}

3.3接收消息

在trade-service服务中定义一个消息监听类：

package com.hmall.trade.listener;

import com.hmall.trade.service.IOrderService;
import lombok.RequiredArgsConstructor;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RequiredArgsConstructor
public class PayStatusListener {

    private final IOrderService orderService;

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "trade.pay.success.queue", durable = "true"),
            exchange = @Exchange(name = "pay.direct"),
            key = "pay.success"
    ))
    public void listenPaySuccess(Long orderId){
        orderService.markOrderPaySuccess(orderId);
    }
}

4.优化消息传输过程

Spring的消息发送代码接收的消息体是一个Object：

而在数据传输时，它会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。

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

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

显然，JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高，因此可以使用JSON方式来做序列化和反序列化。

我们的项目有spring-boot-starter-web依赖，无需再引入Jackson依赖，如果没有spring-boot-starter-web依赖需要引入：

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

只需要在启动类中加入一个Bean即可：

@Bean
public MessageConverter messageConverter(){
    // 1.定义消息转换器
    Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
    // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
    jackson2JsonMessageConverter.setCreateMessageIds(true);
    return jackson2JsonMessageConverter;
}

当然读者也可以把这个bean加入common的config模块，然后再在spring.factories文件中加入com.hmall.common.config.MqConfig（本项目传入Github的分支就是采用的这种方法）

项目下载地址

gupengzu/high-concurrency-project at 07MQ-basic

发表评论

您的邮箱地址不会被公开。 必填项已用 * 标注