Java阶段五Day12

问题解析

顺序消息

• 生产者代码核心逻辑：

  • 按照订单Id，绑定一个固定的队列，按照生成消息的时间，做顺序发送，然后做顺序消费

• 消费者代码核心逻辑：

  • 按顺序消费，消费逻辑主要想要展现有消费成功，有消费失败

事务消息

参考今日内容

Rocket核心概念

Keys

发送消息的时候，消息可以携带一个属性properties，主要作用就是方便消息的查询

在业务代码中，一般都用来绑定业务数据，比如orderId，orderSn，userId等

Tags

每个消息允许绑定一个标签，作用是在消费的时候对于主题消费大量消息来进行过滤使用的

如果过滤掉，消费端代码consumeMessage()对于这条过滤的消息，不执行方法

根据这个业务场景，做简单测试

• producer： 两个组
• consumer：两个组
• 约定tag: group1=ORDER group2=CART

Springboot整合RocketMQ

• 整合软件技术： redis ，mybatis， es ，nacos ，dubbo
• 依赖：**-starter
• yaml文件配置： 对应底层的**Properties类属性，封装配置类的
• 自定义配置类或者注解使用（RocketMQTemplate）

案例使用

• 依赖

<!--整合的依赖-->
<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.2</version>
</dependency>
<!--starter-web-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId>
</dependency>

• 属性配置yaml

rocketmq:# namesrv地址name-server: localhost:9876# 当前应用程序的默认生产者和消费者分组# 代码中可以覆盖producer:group: my-rocket-group-prodconsumer:group: my-rocket-group-consume

• 在代码中编写生产者和消费者逻辑

准备案例环境

HelloController + HelloService实现案例调用功能

代码架构

调整架构，从HelloController同步调用，修改成引入rocketmq的异步调用

生产端发送消息

发送消息的代码位置，注入一个RocketMQTemplate

HelloController

import com.tarena.csmall.rocketmq.demo.service.HelloService;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;/*** @author liner* @version 1.0*/
@RestController
public class HelloController {@Autowiredprivate RocketMQTemplate rocketMQTemplate;@GetMapping("/hello")public String sayHi(String name){//发送消息 Message是rocketMQTemplate支持发送的消息参数//和底层api方法的Message不是同一个类,相当于将底层Message包装了一次.Message message=//payLoad在和,就是bodyMessageBuilder.withPayload(name).setHeader("age",18).build();SendResult sendResult = rocketMQTemplate.syncSend("rocket-topic-a:tagA", message);//rocketMQTemplate.receive();//发送消息return "success";}
}

无论RocketMQTemplate 用哪种发送send消息，最终都会调用doSend实现方法，其他所有方法syncSend，send都是重载 + 外部调用

doSend方法，将Message对象中的payLoad做了序列化，存储到rocketmq message的body中。将header存储到header头中，方便消费的时候做反序列化

消费端（push）

注意：如果使用pull消费，继续使用RocketMQTemplate调用receive方法，每调用一次就从对应目标队列中拿到一条消息

push的消费端，处理步骤：

1. 准备消费端component bean对象
2. 实现消费端push的接口，定义消息的泛型（涉及到spring框架如何将消息反序列化解析）实现方法
3. 配置注解，提供消费属性（监听绑定队列，过滤tag，消费者组）

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;/**topic:消费端绑定主题consumerGroup:消费者分组selectorExpression: 顾虑的标签*/
@Component
@RocketMQMessageListener(topic = "rocket-topic-a",consumerGroup = "${rocketmq.consumer.group}",selectorExpression = "*")
public class MyConsumerListener implements RocketMQListener<String> {/*** 每个listener自定义的对象,底层都会开启一个消费进程 绑定这个listerner* 在底层消费者中,监听consumerMessage方法里,调用这个类的onMessage;* 调用之前,已经实现了对象消息数据的转化* 接口有泛型,底层方法逻辑,会根据泛型,将消息message进行反序列化和数据封装* @param name 根据泛型反序列化的body对象* 对于消费成功还是失败,spring整合rocketmq: 只要抛异常,就返回失败,不抛异常就是正常*/@Overridepublic void onMessage(String name) {System.out.println("消费端接收到消息:"+name);}
}

异步下单操作

将同步的调用关系，转化成异步调用关系，可以引入rocketmq消息中间件

• BusinessService ——> IOrderService 同步关系
• OrderService ——> ICartService，OrderService ——> IStockService 同步关系

考虑：是不是所有的同步，都有必要转化成异步

1. Business ——> 调用OrderService

可以将同步转化成异步，这样做的好处，提升请求并发qps，缺点是不知道订单到底是成功还是失败。（业务处理落地方案选型在这里是需要平衡的，并发和业务用户体验）

2. Order ——> ICartService

可以异步，只要订单新增成功，说明库存够用，删除购物车，可以不在当前业务同步执行，降低订单处理时长，提升RT效率

3. Order ——> IStockService

不可以异步，必须同步（银行账号，支付平台账号划款，转账到当前系统的用户账户金额中）

将Business调用Order的过程实现异步下单：

• Producer： BusinessServiceImpl生产消息，发送到队列
• Consumer： Order的 web 应用实现消息的接收，调用OrderServiceImpl实现消费逻辑

Business生产端

• 依赖 rocketmq

<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.2</version>
</dependency>

• yaml 配置属性

#添加rocket配置
rocketmq:name-server: localhost:9876producer:group: business-producerconsumer:group: business-consumer

• 注入template发送消息，根据发送结果，处理业务逻辑

import cn.tedu.csmall.all.service.IBusinessService;
import cn.tedu.csmall.all.service.IOrderService;
import cn.tedu.csmall.commons.exception.CoolSharkServiceException;
import cn.tedu.csmall.commons.pojo.order.dto.OrderAddDTO;
import cn.tedu.csmall.commons.restful.ResponseCode;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;@Service
@Slf4j
public class BusinessServiceImpl implements IBusinessService {@Autowiredprivate RocketMQTemplate rocketMQTemplate;/* @DubboReference(loadbalance = "roundrobin")private IOrderService dubboOrderService;*/@Overridepublic void buy() {// 模拟触发购买业务// 先实例化一个用于新增订单的DTOOrderAddDTO orderAddDTO=new OrderAddDTO();orderAddDTO.setUserId("UU100");orderAddDTO.setCommodityCode("PC100");//订单 快照orderAddDTO.setMoney(100);orderAddDTO.setCount(2);// 暂时只能进行输出,后期有微服务支持可以调用其他模块log.info("新增订单信息为:{}",orderAddDTO);// dubbo调用order模块新增订单的方法// 将上面实例化的orderAddDTO当做参数,让它在数据库中生效/*dubboOrderService.orderAdd(orderAddDTO);*///替换成异步生单逻辑 发送订单新增的消息//消息的携带信息,消息的封装特点. 消息一定要精简(足够小的占用空间)准确(足够用处理业务逻辑)Message message= MessageBuilder.withPayload(orderAddDTO).build();SendResult sendResult = rocketMQTemplate.syncSend("business-order-topic:orderAdd", message);if (!sendResult.getSendStatus().toString().equals("SEND_OK")){throw new CoolSharkServiceException(ResponseCode.BAD_REQUEST,"订单消息发送失败");}}
}

Order消费端

参考以下步骤，实现消费端

Order-adapter整合 rocketmq

• 依赖

• yaml

• 代码：消费端

消费逻辑步骤

• 消费监听，实现接口
• spring bean对象
• 注解提供消费详细信息（主题，分组，过滤tag）

获取消息之后的业务逻辑

• 注入IOrderServiceImpl实现，再调用orderAdd方法

• 消息类型是什么OrderAddDTO直接接收

import cn.tedu.csmall.all.service.IOrderService;
import cn.tedu.csmall.commons.exception.CoolSharkServiceException;
import cn.tedu.csmall.commons.pojo.order.dto.OrderAddDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;/*** @author liner* @version 1.0*/
@Component
@RocketMQMessageListener(topic = "business-order-topic",consumerGroup = "${rocketmq.consumer.group}",selectorExpression = "orderAdd")
@Slf4j
public class OrderAddConsumerListener implements RocketMQListener<OrderAddDTO> {@Autowiredprivate IOrderService orderService;@Overridepublic void onMessage(OrderAddDTO orderAddDTO) {//调用service,执行orderAdd方法//异常处理逻辑,消息消费失败的处理逻辑try{orderService.orderAdd(orderAddDTO);}catch (CoolSharkServiceException e){//业务异常,说明订单新增业务性失败,比如库存没了log.error("库存减少失败,库存触底了:{},异常信息:{}",orderAddDTO,e.getMessage());}}
}

• MQ的功能之一： 异步解耦
• MQ的功能之二： 消峰填谷

场景分析

分布式系统架构中，队列是分布式的，生产端是分布式集群，消费端也是分布式集群。相当于有多个消费端同时监听队列，同时减库存，写入订单。

面试题：如何处理消息重复消费的问题，重复消费大部分场景，需要解决的

引入2个概念来解决： 幂等的业务方法，和消息的分布式锁

消息重复消费的问题

消息队列重复消费问题，是深入询问的面试题之一，设计到的2个概念，本章节详细介绍一下

方法的幂等

结论： 一个方法的一次业务逻辑调用和N次调用的结果是一致的，我们称这种方法就是幂等。

案例中一旦重复消费，一定要把消费的业务逻辑方法（orderAdd）设计成幂等的

• 幂等的方法

  • GET方法： 查询方法，天生幂等

  • DELETE方法： 删除方法，天生幂等

  • PUT方法： 修改,并不是天生幂等，需要设计

    • 减少库存：

      • update stock_tbl set stock=stock-#{stock} where id=#{id};
        

        （不是幂等）

      • select * from stock_log where order_id=#{orderId};
        

        （查询日志，判断是否已经减过库存了），没有数据

        update stock_tbl set stock=stock-#{stock} where id=#{id};
        insert into stock_log (字段) values (订单id,商品减库存信息);
        

        （这样设计就幂等了，依然有问题）

  • POST方法： 新增，并不是天生幂等，需要设计

    • 新增订单：

    • insert into order_tbl (order_id,order_item_id,count,user_id) values (各种属性);
      

      如果使用唯一属性校验，作用在order_id order_sn(编号)

      同一张订单，这个字段值是相同（幂等满足，没做幂等不满足）

• 当前orderAdd方法设计幂等的解决思路（之一）

  • 使用订单id 或者订单编号，userId+商品id （这个只满足当前的案例特点，不满足实际场景）查询订单，如果已经存在了，库存不减少，订单不增了，购物车不用删除了
  • 补充一个查询方法

  @Override
  public void orderAdd(OrderAddDTO orderAddDTO) {//幂等设计思路: 利用userId和commodityCode 查询,如果已经存在了订单,方法直接执行结束//如果结果不存在,减库存,生单,删除购物车int count=orderMapper.selectExists(orderAddDTO);if (count>0){log.debug("订单已经新增了");return;}StockReduceCountDTO countDTO=new StockReduceCountDTO();countDTO.setCommodityCode(orderAddDTO.getCommodityCode());countDTO.setReduceCount(orderAddDTO.getCount());// 利用Dubbo调用stock模块减少库存的业务逻辑层方法实现功能stockService.reduceCommodityCount(countDTO);// 2.从购物车中删除用户选中的商品(调用Cart模块删除购物车中商品的方法)// 利用dubbo调用cart模块删除购物车中商品的方法实现功能Order order=new Order();BeanUtils.copyProperties(orderAddDTO,order);// 下面执行新增 假设insert是幂等的.orderMapper.insertOrder(order);log.info("新增订单信息为:{}",order);cartService.cartDelete(orderAddDTO);
  }
  

​ 也在OrderMapper补充了一个sql语句，查询存在的订单

@Select("select count(id) from " +
"order_tbl where user_id=#{userId} and commodity_code=#{commodityCode}")
int selectExists(OrderAddDTO orderAddDTO);

分布式锁

当前分布式消费架构

即使，将方法设计成幂等，这个架构中，消息重复消费

满足线程安全问题的所有因素

• 并发 / 多线程
• 写操作
• 共享数据

只要解决其中一点，线程安全问题就消失了

• 并发多线程 ——> 串行
• 写操作 ——> 避免写（不能满足当前案例，必须写）
• 共享数据 ——> 个体数据（不能满足，重复消费，重复订单是前提）

分布式线程安全问题的解决方案——分布式锁

错误思路： 引入synchronized同步锁，不能解决分布式场景下，多个进程的并发线程安全问题

概念： 分布式场景下，多进程，多线程并发的抢锁机制。抢到资源锁，执行业务逻辑，抢不到等待或者放弃执行。能够避免对同一个资源出现并发多线程操作的解决方案

和synchronized的区别在于synchronizeds本地锁。管理一个进程中的多线程，分布式锁是管理多个进程中的多线程

分布式锁当前落地方案： redis setnx命令

Redis五种数据类型

目标：

• 了解 redis 五种类型以及应用场景
• 了解 redis 五种类型的操作命令
• 理解 redis 的分布式锁命令setnx机制

和类型无关的命令

• set / get 非常常用的字符串类型数据的写 / 覆盖，和读

以下的命令是不区分数据类型（key-value类型是总类型，在 redis 中对于value数据结构是严格的区分的，存在五种不同的value数据类型）

• keys {patterns}

keys *

这个命令表示要查询 * 匹配的当前 redis 节点的所有key值，将已有的数据返回，没有数据时，返回空。这里返回的都是内存中保存的数据key值

*不能在线上系统redis使用keys ，会造成redis阻塞

不支持cluster分布式结构的，不能通过一个keys * 从一个 redis 服务中查看其它 redis 数据

• exists key

exists user

查看一个key值是否存在，如果存在返回1，如果不存在返回0，可以查看多个key，不可以使用get这种读操作来代替exists判断存在的操作，使用读判断存在会浪费读数据的带宽

exists key1 key2 ...

• expire / pexpire key time

expire user 1000

给 user 数据在 redis 设置超时时间1000秒

pexpire user 1000

给 user 数据在 redis 设置超时时间1000毫秒

对于使用springboot客户端直接应用 redis 的程序代码，区别不大，没有使用相关超时的数据写入时，默认是永久数据。

• ttl / pttl

ttl name

查看 name 这个key值在 redis 剩余秒数

pttl name

查看 name 这个key值在 redis 剩余毫秒数

永久数据返回 -1，超时 / 删除数据返回 -2

• del key

del name

删除一个叫做 name 的 key值

• type key

set name wanglaoshi
type name

使用type可以查询当前 redis 存储这个key使用都的类型名称

lpush list01 100 200 300
type list01

• flushall

flushall

冲刷所有，清空当前 redis 实例中所有记录的数据，将当前 redis 服务的内存数据和持久化文件中的数据全部清空。

所以这个命令不能在生产环境上线的系统使用，经常在开发和测试环境使用

• save

save

redis 支持持久化，将内存数据，输出到持久化文件，内存数据保存在磁盘上。redis 重新启动时自动加载保存的持久化文件，将数据恢复回来。save命令的调用，就是将内存数据输出到持久化文件中保存。

redis 默认给你提供save命令的间隔调用时间

• 900 1
• 300 10
• 60 10000

上述配置的含义，表示三个定时扫描的逻辑。前面的数值是时间秒，后面表示判断数据变动的次数。如果满足则调用save（定时的趋势，数据变动越频繁，save调用的时间间隔越短）

如果是非正常关机，非正常断电导致 redis 进程消失没有save的数据，就丢失了

基本类型——String

String 常用命令

• set key value
  • EX：可以在 set 时直接设置超时秒数
  • PX：可以在 set 时直接设置超时毫秒数
  • NX：在执行 set 时，会判断 redis 中有没有该key值，如果有则无法set，没有则可以set成功。表示，只有第一个set数据的客户端可以成功，后续都会失败。
  • XX：在执行 set 时，会判断 redis 中有没有该key值，有则会set成功，没有则不成功。表示，使用XX的客户端没有新建的权限。

set bomb tnt EX 50set age 22 NX
set name wanglaoshi NXset gender male XX
set age 55 XX

redis 可以对字符串类型进行写操作调用 set 命令，也可以在已有数据时，对数据覆盖操作。

• get key

get age

从 redis 中读取key值的value数据。在 redis 中value最大数据长度1GB。

• incr / incrby，decr / decrby

incr age
decr age
incrby age 10
decrby age 20

执行计步器，可以增加数值，减少数值。对应value字符串数据必须是纯数字

常见的应用使用计步器：

• 记录排队人数（拿号，自增，叫号后，前剩余人数自减）；
• 在线人数统计（每秒钟上下变动）

String 应用场景

一般使用String类型的value数据实现 缓存的功能。并且可以利用代码的序列化和反序列化的方法，将对象序列化为字符串（user ——>{“userName”:“wanglaoshi”}）。在easymall中使用序列化将product对象变成json，以商品 id 作为唯一key值操作商品在 redis 的缓存数据。