谷粒商城(RabbitMq简介、订单服务、分布式事务)思路详解_谷粒商城分布式事物怎么设计的

1.RabbitMq简介

1.应用场景

①串行处理注册成功后发送邮件和发送短信太慢，可以通过消息队列，把注册成功信息放进去，然后发送邮件和发送短信都订阅了消息队列只要一有信息就能够进行处理。而且用户不需要等待

②解耦：主要就是订单系统生成订单后去调用库存系统，如果库存系统api更新，那么修改幅度大，可以通过消息队列来保存订单消息，库存根据消息来进行api的调用

③流量控制：其实就是秒杀活动的时候，用户请求放入队列，并且根据秒杀业务的速度来控制请求出来的速度。

2.两种类型的mq

1. JMS：支持java平台，不跨语言，本质就是javaAPI，相当于就是一个规范。提供大量的message结构
2. AMQP：跨语言跨平台，是一个网络线级协议。只提供byte数组传输。

3.rabbitmq的原理

1. 生产者：主要就是生产消息，与消息队列建立连接通过多个信道来传输消息
2. 消息：头+体还有就是路由键
3. 交换机：就是通过消息的路由键来决定发送给哪一个队列。
4. 虚拟主机：主要就是用于间隔两种语言平台设置的，他们各自有各自的消息代理
5. 消费者：与消息代理建立连接，监听消息队列，等待消息并进行处理。

4.安装rabbitmq

4369、25672：Erlang发现和集群端口

5672、5671：AMQP高级消息队列协议端口

15672：web后台端口

61613/61614：STOMP协议端口

安装之后直接访问http端口 15672。

 docker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p 25672:25672 -p 15671:15671 -p 15672:15672 rabbitmq:management
1

docker update rabbitmq --restart=always
1

5.交换机的类型

1. direct：点对点，也就是只能发给一个队列
2. tanout：广播所有的队列
3. topic：根据队列路由键接收规则和消息路由键来发送消息，可以一发多个队列。

接着就是创建一个点对点的交换机和经典的队列。测试点对点的消息推送和消息接收。然后就是测试fanout交换机，无论是什么路由键的message最后都会广播到所有的消息队列上

绑定思路

①创建队列和交换机

②交换机绑定队，并且设定好跳转到这个队列的路由键

2.整合Springboot

1.RabbitmqAutoConfiguration

组件

• RabbitTeamplate
• AmqpAdmin：需要一个连接对象
• CachingConnectionFactory：连接创建工厂
• RabbitMessagingTemplate：操作消息的类

首先就是创建AmqpAdmin，并且这个时候需要配置一下连接工厂的配置信息，主要就是host、port和virtual-host虚拟主机

2.创建exchange、binding、queue

参数

• exchange:name、durable、autodelete
• queue：name、durable、excusive（排他，其它连接是否能够传输信息进来）、autodelete

@Test
    void exchange() {
        DirectExchange directExchange = new DirectExchange("hello.java.exchange",true,false);
        amqpAdmin.declareExchange(directExchange);
    }

    @Test
    void queue() {
        Queue queue = new Queue("hello.java.queue",true,false,true);
        amqpAdmin.declareQueue(queue);
    }

    @Test
    void binding() {
        Binding binding = new Binding("hello.java.queue", Binding.DestinationType.QUEUE,"hello.java.exchange","hello",new HashMap<>());
        amqpAdmin.declareBinding(binding);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

3.发送消息

思路

①第一个就是要知道在传送消息的时候是需要依靠模板rabbitTemplate，它里面有一个messageConverter，这个转换器主要就是能够把消息变成对应的形式。通常用的是simple，这个转换器就是序列化转换成byte数组来传输，但是我们可以增加自己配置一个Json的Config转换器。发送消息用的是json状态

②接着就是通过模板来发送

 @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    public void sendMessage(){
        MemberResVo memberResVo = new MemberResVo();
        memberResVo.setCity("sss");
        memberResVo.setId(123L);
        rabbitTemplate.convertAndSend("hello.java.exchange","hello",memberResVo);
    }
1
2
3
4
5
6
7
8
9
10

4.接收消息

思路

①开启注解EnableRabbit才能够使用@RabbitListener来监听，并且指定监听队列可以是多个。

②然后就发送消息进行测试

③rabbitHandler主要只能对方法起作用，listener类和方法都可以，但是handler有利于方法的重载。

 @RabbitHandler
    public void acceptMes(MemberResVo resVo){
        System.out.println(resVo);
    }

    @RabbitHandler
    public void acceptMes(OrderEntity orderEntity){
        System.out.println(orderEntity);
    }
1
2
3
4
5
6
7
8
9

5.消息可靠投递

思路

①第一种方法就是事务消息串行发送，等接收到再发送下一条信息，这种消息消耗大量的性能所以不用

②第二种就是通过回调函数机制。

服务收到消息：发送方->brocker代理这里会调用confirmCallback，

消息正确抵达队列：exchange->queue的时候调用的是returnCallback，这两个需要设置在rabbitTemplate里面，所以我们可以把rabbitTem放到config中，然后自己配置

③在配置文件开启这两个调用方法

④消费端的ack机制:其实就是通过设置yml来手动确认消息已经收到，如果没有手动确认那么就一直会留到队列中。它是默认自动签收的。

 rabbitmq:
    host: 192.168.56.10
    port: 5672
    virtual-host: /
    publisher-returns: true
    publisher-confirm-type: correlated
     listener:
      simple:
        acknowledge-mode: manual

1
2
3
4
5
6
7
8
9
10

  @RabbitHandler
    public void acceptMes(MemberResVo resVo, Message message, Channel channel){
        System.out.println(resVo);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        System.out.println("接收到货物->"+deliveryTag);
        try {
            if(deliveryTag%2==0){
                channel.basicAck(deliveryTag,false);
                System.out.println("货物"+deliveryTag+"通过");
            }else{
                channel.basicNack(deliveryTag,false,false);
                System.out.println("货物"+deliveryTag+"拒绝通过");
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

 @PostConstruct
    public void setCallback(){
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             *
             * @param correlationData 唯一标识
             * @param b 是否抵达
             * @param s 错误信息
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                System.out.println("data["+correlationData+"]==>b["+b+"]==>s["+s+"]");
            }
        });

        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            /**
             *
             * @param message 回调失败的信息
             * @param  replyCode 回复的状态码
             * @param replyText 回复的文本内容
             * @param exchange 当时消息发给那个交互机
             * @param routeKey 通过什么路由键进行的处理
             */
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routeKey) {
                System.out.println("message["+message+"]==>replyCode["+replyCode+"]==>replyText["+replyText+"]==>exchange["+exchange+"]==>routeKey["+routeKey+"]");
            }
        });
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

3.订单服务

1.配置环境

2.整合SpringSession

思路

①导入redis和session依赖

②配置host和port

③引入线程config和线程配置

④开启session注解

⑤引入session的config类

拓展与坑

这里如果没有引进session的config类问题就是取session的时候这个id是不一样的，导致没有取出来。

3.订单简介

处理信息流、资金流、物流，订单需要处理的模块非常多。包括订单，商品、支付、促销、物流、用户。

4.订单登录拦截

思路

①写一个toTrade跳到订单确认页面

②登录拦截，主要就是如果没有登录那么就跳转到登录页面并且显示没有登录，如果登录那么就直接可以跳转到订单确认页面

拓展与坑

①记得要把拦截器加入到webmvcconfigurer里面，也就是mvc的IOC容器。才会起作用。

@Controller
public class OrderWebController {
    @GetMapping("toTrade")
    public String toTrade(){
        return "confirm";
    }
}
1
2
3
4
5
6
7

拦截器

@Component
public class LoginUserInterceptor implements HandlerInterceptor {

    public static ThreadLocal<MemberResVo> loginUser=new ThreadLocal<>();
    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        MemberResVo attribute = (MemberResVo) request.getSession().getAttribute(AuthServerConstant.LOGIN_USER);
        if(attribute!=null){
            loginUser.set(attribute);
            return true;
        }else{
            request.getSession().setAttribute("msg","请先登录");
            response.sendRedirect("http://auth.gulimall.com/login.html");
            return false;
        }


    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

5.订单确认页模型抽取

思路

①需要地址信息、商品信息（就是购物车项的信息）、优惠券、总金额、应付金额

@Data
public class OrderConfirmVo {
    //1.地址列表信息
    private List<MemberAddressVo> address;

    //2.商品信息（购物车里面的信息）
    private List<OrderItemVo> items;

    //3.优惠券
    private Integer integration;


    //4.订单总额
    private BigDecimal totalPrice;

    //5.应付价格
    private BigDecimal payPrice;

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

6.订单确认数据获取

思路

①第一个就是远程调用member获取地址

②然后就是远程调用获取cartItem信息，这里只能选择check是true的购物车项，cartitem这里也需要远程调用skuInfo来更新price

③最后就是设置优惠券，并且根据所有的item来计算总价和应付价格

拓展与坑

①最后就是debug调试，出现的问题就是发现没有任何购物车项返回。如果没有购物车项，也没有出错的话，那么问题就可能是没有这个loginUser，远程调用购物车项的逻辑是先看看有没有这个user，如果没有那么就是返回空的，最好就是在这个地方抛出一个异常。

②那么问题出现在什么地方呢？我们调试走进这个远程调用方法，实际上这个地方又开启了一个新的requestTem模板，创建一个新的request，但是这个request是丢失了头部了。导致一个问题就是我们的cookie，也就是sessionid无法传输过去，最后在购物车登录拦截器的时候发现没有sessionid那么也就无法获取session中的loginuser。

解决方案：

其实思路很明确，看看源码的走向，requestTem会经过一系列的拦截器，我们只需要自定义一个拦截器，通过RequestContextHodler来获取当前的request（实际上原理就是ThreadLocal），最后把头部赋值给requestTem，然后再次发送请求就能够取出登录的user，返回它的购物车项的内容。

OrderWebCon

  @GetMapping("toTrade")
    public String toTrade(Model model){

        OrderConfirmVo orderConfirmVo=orderService.confirmOrder();

        model.addAttribute("orderConfirmData",orderConfirmVo);

        return "confirm";
    }
1
2
3
4
5
6
7
8
9

OrderSer

 @Override
    public OrderConfirmVo confirmOrder() {

        OrderConfirmVo orderConfirmVo = new OrderConfirmVo();
        MemberResVo memberResVo = LoginUserInterceptor.loginUser.get();
        //1.远程获取地址
        List<MemberAddressVo> address = memberFeignService.getAddress(memberResVo.getId());
        orderConfirmVo.setAddress(address);
        //2.远程获取购物项信息
        List<OrderItemVo> cartItem = cartFeignService.getCartItem();
        orderConfirmVo.setItems(cartItem);
        
        //3.优惠券
        orderConfirmVo.setIntegration(memberResVo.getIntegration());


        return orderConfirmVo;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

7.异步编排优化订单确认数据获取

思路

主要是远程调用的时候使用异步编排

拓展与坑

异步编排出现的问题就是获取cart和address的线程不同，那么他们就无法获取通过一个threadLocal那么就不能够获取同一个request来给新的requestTem的头部赋值

解决方案

思路就是在分叉的时候可以先取出这个request，然后在处理异步编排的时候set进去

 @Override
    public OrderConfirmVo confirmOrder() throws ExecutionException, InterruptedException {

        OrderConfirmVo orderConfirmVo = new OrderConfirmVo();
        MemberResVo memberResVo = LoginUserInterceptor.loginUser.get();

        //0.先取出这个request，然后set进各个线程的hodler里面
        ServletRequestAttributes requestAttributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
        //1.远程获取地址
        CompletableFuture<Void> memberAddressFuture = CompletableFuture.runAsync(() -> {
           RequestContextHolder.setRequestAttributes(requestAttributes);
            System.out.println("member线程" + Thread.currentThread().getId());
            List<MemberAddressVo> address = memberFeignService.getAddress(memberResVo.getId());
            orderConfirmVo.setAddress(address);
        }, executor);


        CompletableFuture<Void> cartItemFuture = CompletableFuture.runAsync(() -> {
            RequestContextHolder.setRequestAttributes(requestAttributes);
            //2.远程获取购物项信息
            System.out.println("购物车项线程:" + Thread.currentThread().getId());
            List<OrderItemVo> cartItem = cartFeignService.getCartItem();
            orderConfirmVo.setItems(cartItem);
        }, executor);


        //3.优惠券
        orderConfirmVo.setIntegration(memberResVo.getIntegration());

        CompletableFuture.allOf(cartItemFuture,memberAddressFuture).get();


        return orderConfirmVo;
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

8.渲染数据

思路

①直接找到位置，然后遍历循环就可以了

拓展与坑

①这里遇到个问题记得是用orderConfirmData而不是item。而且如果渲染失败，它会重新访问一次网页，这个时候response已经提交，登录拦截器拦截request无法取出这个session会报错。

9.显示有货和无货

思路

①查询购物车项之后才能够获取skuids，然后通过skuIds去获取所有的库存信息。

②把SkuHasStock拉到order这边接收数据，封装成map，skuid-hasStock。最后就是渲染数据

CompletableFuture<Void> cartItemFuture = CompletableFuture.runAsync(() -> {
            RequestContextHolder.setRequestAttributes(requestAttributes);
            //2.远程获取购物项信息
            System.out.println("购物车项线程:" + Thread.currentThread().getId());
            List<OrderItemVo> cartItem = cartFeignService.getCartItem();
            orderConfirmVo.setItems(cartItem);
        }, executor).thenRunAsync(()->{
            List<OrderItemVo> items = orderConfirmVo.getItems();
            List<Long> skuIds = items.stream().map(item -> {
                return item.getSkuId();
            }).collect(Collectors.toList());

            //查询是否有库存信息
            R hassock = wareFeignService.hassock(skuIds);
            if(hassock.getCode()==0){
                List<SkuStockVo> data = hassock.getData(new TypeReference<List<SkuStockVo>>() {
                });
                Map<Long, Boolean> stocks = data.stream().collect(Collectors.toMap(wareSku -> wareSku.getSkuId(), wareSku -> wareSku.getHasStock()));
                orderConfirmVo.setStocks(stocks);
            }
        });
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

10.模拟运费

思路

①先通过wareInfo服务获取fare，其中需要远程访问memberAddress获取地址信息的phone最后一位，作为运费返回给wareInfo。

②前端ajax请求获取运费，并且通过计算得到payPrice。并且渲染

③最后就是处理一下返回的vo，返回price，也返回寄存的地址，最后回显到页面

wareInfoSer

 @Override
    public FareVo getFare(Long addrId) {
        //1.先查出这个地址
        FareVo fareVo = new FareVo();
        R r = memberFeignService.getAddressInfo(addrId);
        if(r.getCode()==0){
            MemberAddressVo memberReceiveAddress = r.getData("memberReceiveAddress", new TypeReference<MemberAddressVo>() {
            });
            String phone = memberReceiveAddress.getPhone();
            String price = phone.substring(phone.length() - 1, phone.length());
            fareVo.setAddress(memberReceiveAddress);
            fareVo.setFare(new BigDecimal(price));

            return fareVo;
        }

        return null;

    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

 @GetMapping("fare")
    public R getFare(@RequestParam("addrId")Long addrId){
        BigDecimal price=wareInfoService.getFare(addrId);
        return R.ok().put("memberAddressPrice",price);
    }
1
2
3
4
5

11.接口幂等性

简介

①如果一个接口无论多少次请求最后结果都是一样的，那么这个就是接口的幂等。

②我们要做的就是保证接口的幂等性，比如提交订单，点击很多次，但是都是提交同一个订单，如果没有进行约束那么就会进行处理很多次。通常解决办法可以是token令牌机制，每次提交订单都需要提交一个验证码与服务器进行对比，如果对比成功那么就处理，并且从redis中取验证码和对比还有删除验证码必须是一个原子操作，不然多个请求进来刚好从redis中取出同一个验证码都对比成功，那么就会加入多个订单。

③还有一种解决方案就是锁的机制，比如乐观锁，主要就是通过更新带version进行对比来决定要不要操作，如果两个操作同时进来，带同一个·version，只要操作一次version+1，那么其中一个处理就会失败。悲观锁相当于就是串行操作。

④还有就是数据库的约束还有防重。防重每次加入一条语句才能够进行操作，如果加入失败那么是不会进行操作的。

12.提交订单

思路

①创建提交订单的对象vo。包括payType、addrId、payPrice这些都是从页面获取的。购物项并不需要获取，因为可以通过loginUser的id来远程调用获取购物车项的内容，而且购物车项的价格是实时更新的可能会发生变化。

②然后就是在获取确认页数据的方法中生成token并且放进redis。这里采用的幂等处理方式就是token。

 //3.优惠券
        orderConfirmVo.setIntegration(memberResVo.getIntegration());

        //4.生成token
        String token = UUID.randomUUID().toString().replace("-", "");
        orderConfirmVo.setOrderToken(token);
        //放进redis
        redisTemplate.opsForValue().set(OrderConstant.USER_ORDER_TOKEN_PREFIX+memberResVo.getId(),token,30, TimeUnit.MINUTES);


        CompletableFuture.allOf(cartItemFuture,memberAddressFuture).get();
1
2
3
4
5
6
7
8
9
10
11

<form action="http://order.gulimall.com/submitOrder" method="post">
				<input id="addrIdInput" name="addrIdInput" type="hidden">
				<input id="orderToken" th:value="${}" name="orderToken" type="hidden">
				<input id="payPriceInput" name="payPriceInput" type="hidden">
				<button class="tijiao">提交订单</button>
			</form>
1
2
3
4
5
6

13.原子校验令牌

思路

①主要通过redisTem可以通过传入script来完成对比和删除token操作。传入一个redisScript，和key还有就是对比的参数。

 @Override
    public OrderResponseVo submitOrder(OrderSubmitVo orderSubmitVo) {
        MemberResVo memberResVo = LoginUserInterceptor.loginUser.get();
        OrderResponseVo response = new OrderResponseVo();
        //1.判断验证码是否正确，使用原子操作
        String script="if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
        String orderToken = orderSubmitVo.getOrderToken();
        Long res= (Long) redisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class),
                Arrays.asList(OrderConstant.USER_ORDER_TOKEN_PREFIX + memberResVo.getId()),
                orderToken);
        if(res==0L){
            //验证失败
            return response;
        }else{
            //验证成功

        }

    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

14.构建订单数据

思路

①创建订单号通过IdWorker的timeId

②创建订单，主要就是靠远程服务ware获取fareVo，里面包含了地址信息和运费信息。

③接着就是远程调用获取购物车项信息，转换成订单项。

④通过ThreadLocal来保存orderSubmitVo，复制订单状态的枚举类。

⑤接着就就是构建订单项list，主要就是远程调用获取购物车项list，然后遍历map构建订单项，其中需要远程调用product获取spu信息，设置好sku信息（通过购物车项），积分信息，订单信息。

拓展与坑

①如果一个方法特别多，那么为了明确思路，可以把各个部分抽出来做另外一个方法，让主体看上去更简洁。

@Override
    public OrderResponseVo submitOrder(OrderSubmitVo orderSubmitVo) {

        MemberResVo memberResVo = LoginUserInterceptor.loginUser.get();
        OrderResponseVo response = new OrderResponseVo();
        //1.判断验证码是否正确，使用原子操作
        String script="if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
        String orderToken = orderSubmitVo.getOrderToken();
        Long res= (Long) redisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class),
                Arrays.asList(OrderConstant.USER_ORDER_TOKEN_PREFIX + memberResVo.getId()),
                orderToken);
        if(res==0L){
            //验证失败
            response.setCode(1);
            return response;
        }else{
            //验证成功
            //1.构建订单号
            OrderCreateTo orderCreateTo = new OrderCreateTo();
            String orderSn = IdWorker.getTimeId();
            //2.远程调用获取地址信息给order赋值,创建订单
            OrderEntity orderEntity = buildOrder(orderSubmitVo, orderSn);

            //3.创建订单列表项
            List<OrderItemEntity> orderItemEntities = buildOrderItems(orderSn);

            //4.



        }

    }

    /**
     * 创建订单项列表
     * @return
     */
    private List<OrderItemEntity> buildOrderItems(String orderSn) {
        List<OrderItemVo> cartItemList = cartFeignService.getCartItem();
        if(cartItemList!=null&&cartItemList.size()>0){
            List<OrderItemEntity> collect = cartItemList.stream().map(cartItem -> {
                OrderItemEntity orderItemEntity = buildOrderItem(cartItem);
                orderItemEntity.setOrderSn(orderSn);
                return orderItemEntity;
            }).collect(Collectors.toList());
            return  collect;
        }
        return null;
    }

    /**
     * 创建订单项
     * @param cartItem
     * @return
     */
    private OrderItemEntity buildOrderItem(OrderItemVo cartItem) {
        OrderItemEntity orderItemEntity = new OrderItemEntity();
        //1.订单信息
        //2.Spu信息
        R spu = productFeignService.getSpu(cartItem.getSkuId());
        SpuInfoVo spuInfo = spu.getData(new TypeReference<SpuInfoVo>() {
        });
        orderItemEntity.setSpuId(spuInfo.getId());
        orderItemEntity.setSpuBrand(spuInfo.getBrandId().toString());
        orderItemEntity.setSpuName(spuInfo.getSpuName());
        orderItemEntity.setCategoryId(spuInfo.getCatalogId());

        //3.Sku信息
        orderItemEntity.setSkuId(cartItem.getSkuId());
        orderItemEntity.setSkuPic(cartItem.getImage());
        orderItemEntity.setSkuName(cartItem.getTitle());
        orderItemEntity.setSkuPrice(cartItem.getPrice());
        String skuAttr = StringUtils.collectionToDelimitedString(cartItem.getSkuAttr(), ";");
        orderItemEntity.setSkuAttrsVals(skuAttr);
        orderItemEntity.setSkuQuantity(cartItem.getCount());
        //4.积分信息
        orderItemEntity.setGiftIntegration(cartItem.getPrice().intValue());
        orderItemEntity.setGiftGrowth(cartItem.getPrice().intValue());
        return orderItemEntity;
    }

    /**
     * 创建订单
     * @param orderSubmitVo
     * @param orderSn
     * @return
     */
    private OrderEntity buildOrder(OrderSubmitVo orderSubmitVo, String orderSn) {
        OrderEntity orderEntity = new OrderEntity();
        orderEntity.setOrderSn(orderSn);

        R fare = wareFeignService.getFare(orderSubmitVo.getAddrId());
        if(fare.getCode()==0){
            FareVo fareVo = fare.getData("fareVo", new TypeReference<FareVo>() {
            });
            orderEntity.setFreightAmount(fareVo.getFare());
            orderEntity.setReceiverCity(fareVo.getAddress().getCity());
            orderEntity.setReceiverDetailAddress(fareVo.getAddress().getDetailAddress());
            orderEntity.setReceiverName(fareVo.getAddress().getName());
            orderEntity.setReceiverPhone(fareVo.getAddress().getPhone());
            orderEntity.setReceiverPostCode(fareVo.getAddress().getPostCode());
            orderEntity.setReceiverProvince(fareVo.getAddress().getProvince());
            orderEntity.setReceiverRegion(fareVo.getAddress().getRegion());
        }
        return orderEntity;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107

15.订单验价

思路

①计算所有的优惠价、积分价和总价的item的和。

②然后赋值到order里面去。

③然后就是提交的payPrice和重新计算好有优惠的总价进行一个比较。

16.库存锁定、保存订单

思路

①保存订单的思路很简单就是，直接插入订单和所有的订单项，并且设置好更新的时间

②接着就是库存锁定，需要知道是哪个商品，多少个。也就是需要skuId和num，直接通过orderItemVo也能够进行传输，还有就是要知道订单号。所以新创建一个WareSkuLockVo来封装锁存商品list和订单号.远程调用仓库锁定库存，并且返回是否锁定成功的信息（需要创建一个库存锁定的返回vo，主要就是skuid，num、还有是否锁定成功）。

保存订单、锁定库存

 //3.保存订单
                saveOrder(orderCreateTo);

                //4.锁定库存
                WareSkuLockVo wareSkuLockVo = new WareSkuLockVo();
                wareSkuLockVo.setOrderSn(orderCreateTo.getOrder().getOrderSn());
                List<OrderItemEntity> orderItems = orderCreateTo.getOrderItems();
                List<OrderItemVo> orderItemVos = orderItems.stream().map(item -> {
                    OrderItemVo orderItemVo = new OrderItemVo();
                    orderItemVo.setSkuId(item.getSkuId());
                    orderItemVo.setTitle(item.getSkuName());
                    orderItemVo.setCount(item.getSkuQuantity());
                    return orderItemVo;
                }).collect(Collectors.toList());
                wareSkuLockVo.setLocks(orderItemVos);

                R r = wareFeignService.lockOrder(wareSkuLockVo);
                if(r.getCode()==200){

                }else{

                }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

需要锁定的订单项.

@Data
public class WareSkuLockVo {
    private String orderSn;//订单号
    private List<OrderItemVo> locks;//需要库存锁定的订单项
}

1
2
3
4
5
6

库存锁定返回结果

@Data
public class LockResult {
    private Long skuId;
    private Integer num;
    private boolean locked;
}

1
2
3
4
5
6
7

17.库存锁定具体操作

思路

①首先就是要查询商品是否有仓库拥有这些商品。注意stock-stocklock才是可以调动的库存

②然后就是封装skuid、wareids（仓库id）、num（锁定的库存数量）到一个skuHasStock对象里面。

③遍历这个skuHasStock集合，并且取出里面仓库ids，遍历ids，查询是否能够锁定库存（stock-lock>=num意思就是必须有空闲的库存>我们请求锁定的库存），如果有那么这个商品遍历结束下一个，如果没有锁定成功，那么整个锁定事务就会失败并且抛出异常。\

WareSkuCon

   @PostMapping("lock/order")
    public R lockOrder(@RequestBody WareSkuLockVo wareSkuLockVo){
        boolean result= false;
        try {
            result = wareSkuService.lockOrder(wareSkuLockVo);
            return R.ok();
        } catch (Exception e) {
            return R.error(BizCodeEnume.NO_STOCK_EXCEPTION.getCode(),BizCodeEnume.NO_STOCK_EXCEPTION.getMsg());
        }

    }
1
2
3
4
5
6
7
8
9
10
11

WareSkuSer

 @Override
    public Boolean lockOrder(WareSkuLockVo wareSkuLockVo) {
        //所有的商品项
        List<OrderItemVo> locks = wareSkuLockVo.getLocks();
        //1.找到有这个商品的仓库
        List<SkuHasStock> skuHasStockList= locks.stream().map(item -> {
            SkuHasStock skuHasStock = new SkuHasStock();
            Long skuId = item.getSkuId();
            skuHasStock.setSkuId(skuId);
            skuHasStock.setNum(item.getCount());
            List<Long> wareIds = this.baseMapper.listWareIdHasSkuStock(skuId);
            skuHasStock.setWareIds(wareIds);
            return skuHasStock;
        }).collect(Collectors.toList());

        //2.遍历仓库id，锁定库存
        for (SkuHasStock skuHasStock : skuHasStockList) {
            List<Long> wareIds = skuHasStock.getWareIds();
            if(wareIds==null||wareIds.size()<=0){
                throw new NoStockException(skuHasStock.getSkuId());
            }
            boolean skuLocks=false;
            for (Long wareId : wareIds) {
               int count= this.baseMapper.lockStock(wareId,skuHasStock.getSkuId(),skuHasStock.getNum());
               if(count==1) {
                   //成功,那么就返回结束这次的循环，因为只需要锁定一个库存可以了
                   skuLocks = true;
                   break;
               }
            }
            //如果商品没有锁定成功那么就抛出异常。只要有一个出问题都不行
            if(skuLocks){
                throw  new NoStockException(skuHasStock.getSkuId());
            }
        }


        return true;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

18.订单提交的debug测试

整个思路总结

①创建订单，主要就是创建订单号，然后通过购物车项的skuid获取商品项，然后给订单项赋值，创建订单项列表。

②然后就是计算订单的总价格，主要就是加上各种优惠之后，所有订单项的总价和优惠、积分合并成订单。

③最后就是验价和锁定库存。

④订单模块需要调用到商品模块的spu和sku，还需要调用到库存wareSku和wareInfo等信息。

拓展与坑

①中间遇到一个bug就是远程调用ware的时候出现调用错误，这个时候可以检查一下两个接口的共用参数是不是错了， 如果是类可以进去看看类的属性是不是发生了变化。我的错误就是订单号一个是long，一个是string类型导致的错误。

orderSer

 @Transactional
    @Override
    public OrderResponseVo submitOrder(OrderSubmitVo orderSubmitVo) {

        orderSubmitVoThreadLocal.set(orderSubmitVo);

        MemberResVo memberResVo = LoginUserInterceptor.loginUser.get();
        OrderResponseVo response = new OrderResponseVo();
        response.setCode(0);
        //1.判断验证码是否正确，使用原子操作
        String script="if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
        String orderToken = orderSubmitVo.getOrderToken();
        Long res= (Long) redisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class),
                Arrays.asList(OrderConstant.USER_ORDER_TOKEN_PREFIX + memberResVo.getId()),
                orderToken);
        if(res==0L){
            //验证令牌失败
            response.setCode(1);
            return response;
        }else{
            //验证成功
            //1.构建订单号
            OrderCreateTo orderCreateTo = createOrder();

            //2.验价
            BigDecimal payPrice = orderSubmitVo.getPayPrice();
            BigDecimal payAmount = orderCreateTo.getOrder().getPayAmount();
            //对比的是提交订单的应付价格和创建订单后的应付价格是否一致
            if(Math.abs(payPrice.subtract(payAmount).doubleValue())<0.01){

                //3.保存订单
                saveOrder(orderCreateTo);

                //4.锁定库存
                WareSkuLockVo wareSkuLockVo = new WareSkuLockVo();
                wareSkuLockVo.setOrderSn(orderCreateTo.getOrder().getOrderSn());
                List<OrderItemEntity> orderItems = orderCreateTo.getOrderItems();
                List<OrderItemVo> orderItemVos = orderItems.stream().map(item -> {
                    OrderItemVo orderItemVo = new OrderItemVo();
                    orderItemVo.setSkuId(item.getSkuId());
                    orderItemVo.setTitle(item.getSkuName());
                    orderItemVo.setCount(item.getSkuQuantity());
                    return orderItemVo;
                }).collect(Collectors.toList());
                wareSkuLockVo.setLocks(orderItemVos);

                //锁定库存
                R r = wareFeignService.lockOrder(wareSkuLockVo);
                if(r.getCode()==0){
                     response.setCode(0);
                     response.setOrder(orderCreateTo.getOrder());
                      return response;
                }else{
                    //库存锁定失败
                   throw  new NoStockException();
//                    response.setCode(3);
//                    return response;

                }


            }else{
                //验价失败
                response.setCode(2);
                return response;
            }



        }

    }

    /**
     * 保存订单
     * @param orderCreateTo
     */
    private void saveOrder(OrderCreateTo orderCreateTo) {
        OrderEntity order = orderCreateTo.getOrder();
        //1.设置更新时间
        order.setModifyTime(new Date());

        //2.保存订单
        this.save(order);

        //3.保存订单项
        List<OrderItemEntity> orderItems = orderCreateTo.getOrderItems();
        orderItemService.saveBatch(orderItems);
    }


    private OrderCreateTo createOrder(){
        OrderSubmitVo orderSubmitVo = orderSubmitVoThreadLocal.get();
        String orderSn = IdWorker.getTimeId();
        //2.远程调用获取地址信息给order赋值,创建订单
        OrderEntity orderEntity = buildOrder(orderSubmitVo, orderSn);

        //3.创建订单列表项
        List<OrderItemEntity> orderItemEntities = buildOrderItems(orderSn);

        //4.计算所有的价格
        computeItem(orderEntity,orderItemEntities);

        OrderCreateTo orderCreateTo = new OrderCreateTo();
        orderCreateTo.setOrder(orderEntity);
        orderCreateTo.setOrderItems(orderItemEntities);

        return orderCreateTo;
    }

    private void computeItem(OrderEntity orderEntity, List<OrderItemEntity> orderItemEntities) {
           //1.遍历所有项计算出总价，总优惠价格
        BigDecimal total=new BigDecimal("0.0");
        BigDecimal coupon=new BigDecimal("0.0");
        BigDecimal integration=new BigDecimal("0.0");
        BigDecimal promotion=new BigDecimal("0.0");
        BigDecimal growth=new BigDecimal("0.0");
        BigDecimal gift=new BigDecimal("0.0");
        for (OrderItemEntity orderItemEntity : orderItemEntities) {
           total= total.add(new BigDecimal(orderItemEntity.getRealAmount().toString()));
           coupon=coupon.add(new BigDecimal(orderItemEntity.getCouponAmount().toString()));
           integration=integration.add(new BigDecimal(orderItemEntity.getIntegrationAmount().toString()));
           promotion=promotion.add(new BigDecimal(orderItemEntity.getPromotionAmount().toString()));
           growth=growth.add(new BigDecimal(orderItemEntity.getGiftGrowth()));
           gift=gift.add(new BigDecimal(orderItemEntity.getGiftIntegration()));
        }
        orderEntity.setCouponAmount(coupon);
        orderEntity.setIntegrationAmount(integration);
        orderEntity.setPromotionAmount(promotion);
        orderEntity.setTotalAmount(total);
        orderEntity.setPayAmount(total.add(new BigDecimal(orderEntity.getFreightAmount().toString())));
        orderEntity.setGrowth(growth.intValue());
        orderEntity.setIntegration(gift.intValue());

        //订单状态


    }

    /**
     * 创建订单项列表
     * @return
     */
    private List<OrderItemEntity> buildOrderItems(String orderSn) {
        List<OrderItemVo> cartItemList = cartFeignService.getCartItem();
        if(cartItemList!=null&&cartItemList.size()>0){
            List<OrderItemEntity> collect = cartItemList.stream().map(cartItem -> {
                OrderItemEntity orderItemEntity = buildOrderItem(cartItem);
                orderItemEntity.setOrderSn(orderSn);
                return orderItemEntity;
            }).collect(Collectors.toList());
            return  collect;
        }
        return null;
    }

    /**
     * 创建订单项
     * @param cartItem
     * @return
     */
    private OrderItemEntity buildOrderItem(OrderItemVo cartItem) {
        OrderItemEntity orderItemEntity = new OrderItemEntity();
        //1.订单信息
        //2.Spu信息
        R spu = productFeignService.getSpu(cartItem.getSkuId());
        SpuInfoVo spuInfo = spu.getData(new TypeReference<SpuInfoVo>() {
        });
        orderItemEntity.setSpuId(spuInfo.getId());
        orderItemEntity.setSpuBrand(spuInfo.getBrandId().toString());
        orderItemEntity.setSpuName(spuInfo.getSpuName());
        orderItemEntity.setCategoryId(spuInfo.getCatalogId());

        //3.Sku信息
        orderItemEntity.setSkuId(cartItem.getSkuId());
        orderItemEntity.setSkuPic(cartItem.getImage());
        orderItemEntity.setSkuName(cartItem.getTitle());
        orderItemEntity.setSkuPrice(cartItem.getPrice());
        String skuAttr = StringUtils.collectionToDelimitedString(cartItem.getSkuAttr(), ";");
        orderItemEntity.setSkuAttrsVals(skuAttr);
        orderItemEntity.setSkuQuantity(cartItem.getCount());
        //4.积分信息
        orderItemEntity.setGiftIntegration(cartItem.getPrice().multiply(new BigDecimal(cartItem.getCount())).intValue());
        orderItemEntity.setGiftGrowth(cartItem.getPrice().multiply(new BigDecimal(cartItem.getCount())).intValue());

        //5.设置订单项价格信息
        orderItemEntity.setCouponAmount(new BigDecimal("0"));
        orderItemEntity.setPromotionAmount(new BigDecimal("0"));
        orderItemEntity.setIntegrationAmount(new BigDecimal("0"));
        BigDecimal origin = orderItemEntity.getSkuPrice().multiply(new BigDecimal(orderItemEntity.getSkuQuantity().toString()));
        BigDecimal realAmount = origin.subtract(orderItemEntity.getCouponAmount())
                .subtract(orderItemEntity.getPromotionAmount())
                .subtract(orderItemEntity.getIntegrationAmount());
        orderItemEntity.setRealAmount(realAmount);



        return orderItemEntity;
    }

    /**
     * 创建订单
     * @param orderSubmitVo
     * @param orderSn
     * @return
     */
    private OrderEntity buildOrder(OrderSubmitVo orderSubmitVo, String orderSn) {
        MemberResVo memberResVo = LoginUserInterceptor.loginUser.get();
        OrderEntity orderEntity = new OrderEntity();
        orderEntity.setOrderSn(orderSn);
        orderEntity.setMemberId(memberResVo.getId());

        R fare = wareFeignService.getFare(orderSubmitVo.getAddrId());
        if(fare.getCode()==0){
            FareVo fareVo = fare.getData("fareVo", new TypeReference<FareVo>() {
            });
            orderEntity.setFreightAmount(fareVo.getFare());
            orderEntity.setReceiverCity(fareVo.getAddress().getCity());
            orderEntity.setReceiverDetailAddress(fareVo.getAddress().getDetailAddress());
            orderEntity.setReceiverName(fareVo.getAddress().getName());
            orderEntity.setReceiverPhone(fareVo.getAddress().getPhone());
            orderEntity.setReceiverPostCode(fareVo.getAddress().getPostCode());
            orderEntity.setReceiverProvince(fareVo.getAddress().getProvince());
            orderEntity.setReceiverRegion(fareVo.getAddress().getRegion());
            //设置订单状态
            orderEntity.setDeleteStatus(0);
            orderEntity.setStatus(OrderStatusEnum.CREATE_NEW.getCode());
            orderEntity.setAutoConfirmDay(7);
        }
        return orderEntity;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231

4.分布式事务

1.本地事务会出现的问题

• 远程服务的假失败，意思其实就是远程服务实际上成功，但是由于网络等原因返回失败，导致本地事务回退。远程服务却已经操作了数据库
• 第二个问题就是远程服务之后出现问题，导致远程服务请求发出之后是无法回退的。

2.本地事务的本质

在方法上面加上@Transaction，那么在本类中调用的时候实际上就是通过cglib继承代理对象来调用，增强，然后调用原来对象的方法。如果嵌套其它加上注解的方法，那么这些方法是不起作用的，原因就是第一个被调用的方法是通过代理对象调用的，但是其他方法是直接调用，没有经过事务的增强调用。

解决方案：可以通过开启aspectj动态代理注解@EnableAspectJAutoProxy(exposeProxy = true)，创建一个代理对象，然后通过代理对象来调用这些方法。那么这个就是可以成功，相当于就是把所有方法增强之后的一个代理对象来调用这些方法。本质上就是通过代理对象来调用。也就是方法已经经过增强了。

 @Transactional
    public void a(){
        OrderServiceImpl o = (OrderServiceImpl) AopContext.currentProxy();
        o.b();
        o.c();
    }

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void b(){
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void c(){
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

3.分布式事务

CAP介绍

1. C就是一致性，也就是每个服务的数据库的信息都有相同备份
2. A就是可用性，数据库集群宕机一个，仍然能够使用
3. P就是分区容错性，意思就是网络出现导致服务分区
4. 通常CA不能一起，原因就是如果要保证一致性，出现了分区错误，那么就需要把断开连接的服务切断，如果需要AP，保证可用性，那么即使那个不能连接的服务出错，也要保持开机状态。

保持一致性的方案

1. 牺牲可用性
2. 可以使用raft算法

raft算法简介

1. 三种状态，随从，领导，候选人
2. 两种时间选举时间（没有领导给节点发送信息，那么就开始倒计时，如果超时，那么就会成为候选人），第二种时间就是心跳时间（更新日志，更新选举时间）
3. 如果没有领导发送心跳，那么等待选举超时，之后就会有节点成为候选人，投票决定谁是leader，leader如果需要保持数据一致性，那么就要先通过心跳带上日志告诉这些节点需要改什么，然后自己提交修改之后再通过心跳来通知节点。
4. 如果发生分区，那么每个分区中又会重新进行选举，来保持一致性。如果没有大多数随从那么就一直不会提交，也就无法更新成功。

4.BASE

BA：基本可用

S：软状态，相当于就是可以容忍一部分数据访问不到，或者是旧的数据。过一段时间再更新。强状态就是要么成，要么败

E：最终一致性：容忍数据访问不到，但过一段时间之后最后一定是一致的。

分布式事务通常以AP为主，放弃了强一致性。

5.分布式事务的解决方案

1. 2PC模式：意思就是一个大的事务管理器连接多个服务，每次要执行业务时都去问服务准备好了吗，如果准备好那么就开始执行。如果没有那么就不执行。问题就是性能不行，每次都要询问。
2. 柔性事务TCC事务补偿：意思就是把业务按照3个接口写入，try，confirm和cancel。先写好业务交给各自的服务数据库try，然后就去confirm是否执行成功，如果没有那么就通过canel来回滚补偿。
3. 最大努力通知：两个服务订阅一个消息队列。大模块调用两个服务但是失败了，就把失败信息交给消息队列，两个服务接收到之后就去回滚。而且为了防止服务不知道，它会多次发送消息给消息队列，直到两个服务都知道了。
4. 可靠消息+最终一致性与最大努力通知相似。

6.Seata

原理

首先是TM（事务管理器，全局事务范围）来给TC发送开启一个事务，然后TM调用第一个服务，然后这个服务的RM（资源管理器）向TC注册分支事务，报告自己的事务状态。然后TM继续调用下一个服务，下一个服务也是这么做。如果下一个服务出现问题，那么TC就会协调其它服务，回滚状态，回滚状态的方法就是通过回滚表

坏境配置

1.下载seata

2.给每个服务数据库加上回滚表

测试

①引入依赖

②引入config

@Configuration
public class MySeataConfig {
    @Autowired
    DataSourceProperties dataSourceProperties;

    @Bean
    public DataSource dataSourceProxy(){
        HikariDataSource dataSource = dataSourceProperties.initializeDataSourceBuilder().type(HikariDataSource.class).build();
        if (StringUtils.hasText(dataSourceProperties.getName())) {
            dataSource.setPoolName(dataSourceProperties.getName());
        }
        return new DataSourceProxy(dataSource);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

③引入file和register文件，并且是使用nacos注册中心，file这个地方需要修改一下service，改成自己的服务名。

registry {
  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
  type = "nacos"

  nacos {
    serverAddr = "localhost:8848"
    namespace = "public"
    cluster = "default"
  }
1
2
3
4
5
6
7
8
9

service {
  #vgroup->rgroup
  vgroup_mapping.gulimall-order-fescar-service-group = "default"
1
2
3

④开启seata-server。

⑤给需要加上事务的方法加上注解@GobalTransaction相当于是开启了事务。其它就要被调用的方法开启小事务@Transaction，相当于是资源管理。而TC就是seata本身。

7.AT方式与可靠性+最终一致性

AT方式

at方式的seata分布式事务处理其实就是提交事务，如果发生问题，那么就可以通过undolog来完成回滚。但是这种问题就是需要一个订单一个订单的处理无法处理高并发情况

可靠性+最终一致性

思路其实就是每次加入库存的时候都记录一条锁存记录，如果后面服务出现问题，那么就可以发送解锁消息到消息队列中。并且可以过一段时间之后再来完成回滚操作，因为你需要确认订单是不是已经被取消了。这种的好处就是可以处理高并发，不需要订单取消和解锁同时完成。需要用到延时队列。

8.延时队列

简介

其实就是放到一个没有被监听的队列中，并且设置好过期时间，消息过期之后会去到死信交换机，然后再去到死信队列。死信队列里面的消息都是已经经过了某段时间的消息是可以被监听使用的。延时队列使用的场景可以是订单下单过30min过期的情景，如果是定时任务消耗资源，而且会有时效性问题（刚好在定时检查的后一秒过期，那么就又要等30min才能取了）。如果是延时队列，那么就可以等待30min之后发现没有支付直接取消，然后库存等40min去检查订单是否取消，如果取消，那么就可以去解。

9.延时队列关单模拟

思路

①通过config来创建这些队列，exchange和各种绑定，然后监听信息

②自己创建一个con来完成发送信息的操作

③整个思路就是先通过exchange发送给延迟队列，消息过期发给exchange，再发给被监听的队列。

 @ResponseBody
    @RequestMapping("send/message")
    public String published(){
        OrderEntity orderEntity = new OrderEntity();
        orderEntity.setOrderSn(UUID.randomUUID().toString());
        System.out.println("发送成功"+orderEntity.getOrderSn());
        rabbitTemplate.convertAndSend("order-event-exchange","order.create.order",orderEntity);
        return "ok";
    }
1
2
3
4
5
6
7
8
9

@Configuration
public class MyMqConfig {

    @RabbitListener
    public void listener(OrderEntity orderEntity, Channel channel, Message message) throws IOException {
        System.out.println("接收到订单号:"+orderEntity.getOrderSn());
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }

    @Bean
    public Queue orderDelayQueue(){
        Map<String,Object> args=new HashMap<>();
        args.put("x-dead-letter-exchange","order-event-exchange");
        args.put("x-dead-letter-routing-key","order.release.order");
        args.put("x-message-ttl",60000);
        return new Queue("order.delay.queue",
                true,
                false,
                false,
                args);
    }

    @Bean
    public Queue orderReleaseQueue(){
        return new Queue("order.release.order.queue",
                true,
                false,
                false);
    }

    @Bean
    public Exchange orderEventExchange(){
       return new TopicExchange("order-event-exchange",
               true,
               false);
    }

    @Bean
    public Binding orderCreateOrderBingding(){
       return  new Binding("order.delay.queue",
               Binding.DestinationType.QUEUE,
               "order-event-exchange",
               "order.create.order",
               null);
    }

    @Bean
    public Binding orderReleaseOrderBinding(){
        return  new Binding("order.release.order.queue",
                Binding.DestinationType.QUEUE,
                "order-event-exchange",
                "order.release.order",
                null);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

10.库存锁定延迟队列

思路

①先把库存锁定消息放到延迟队列，等待2min之后检查订单号是否存在如果存在，那么就不需要去到死信exchange中，如果没有那么就要去到死信exchange转发到死信队列，并且被监听消息取消锁定。

②所以这个地方需要两个队列，两个绑定，一个死信exchange就可以了

拓展与坑

①如果发现没有队列和exchange生成，问题很可能就是没有连接到这个rabbit的服务器。也可能是name写错了

11.监听库存解锁

思路

①在锁定库存之前先把工作单创建，主要就是保存订单号

②如果锁成功，那么就加入工作单项，也就是到底什么商品锁定了多少库存，在什么仓库这样的信息。并且封装商品id和工作单项信息到to对象里面，并且通过rabbitTem来发送消息给延时队列。然后在WareSku中监听这个消息。

③然后在OrderSer中锁库存之后，加上一个异常，创造一个环境就是，订单成功、锁存成功、但是后面服务失败需要回滚的状态。关闭全局的事务管理。提交订单创造工作单和工作单项，但是后面服务出现问题并没有回滚。并且在WareSku中监听并输出来看看是否接收成功

④到了解锁功能，先查询这个detail是否存在数据库，如果存在那么才进行解锁，如果没有那么就不进行操作

⑤接着就是查询工作单中的订单是否存在如果存在那么就解锁，需要传入的是仓库id，商品id和锁定库存，然后修改仓库的值就可以了。并且把rabbitmq改成手动，只有我们通过channel同意的时候才能够放过这些消息

拓展与坑

①去查订单远程调用的时候需要登录，但是有时候做这件事的时候是用户下线，消息才被接收处理，所以需要放行。而且直接通过远程访问的问题就是这个请求并没有带上sessionid无法获取loginUser，最后导致返回的数据是html。所以会一直报错。两种方法，第一种就是通过给请求头加上之前请求头，第二种就是这个请求能够通过拦截器放过去。

②把listener的解锁方法抽取出来，如果没有出现异常那么就是解锁成功，如果不是那么就是解锁失败，通过抓取异常来进行判断

WareSkuSer

@RabbitHandler
    public void handle(StockLockedTo stockLockedTo, Message message){

        System.out.println("收到库存信息");
        StockDetailTo detailTo = stockLockedTo.getDetailTo();
        //1.查询detail是否存在
        Long detailId = detailTo.getId();
        WareOrderTaskDetailEntity detailEntity = wareOrderTaskDetailService.getById(detailId);
        if(detailEntity!=null){
            //如果存在那么就去解锁。

        }else{
            //如果不存在那么就不做任何处理

        }
    }

    @Transactional
    @Override
    public Boolean lockOrder(WareSkuLockVo wareSkuLockVo) {

        //创建工作单
        WareOrderTaskEntity taskEntity = new WareOrderTaskEntity();
        taskEntity.setOrderSn(wareSkuLockVo.getOrderSn());
        wareOrderTaskService.save(taskEntity);


        //所有的商品项
        List<OrderItemVo> locks = wareSkuLockVo.getLocks();
        //1.找到有这个商品的仓库
        List<SkuHasStock> skuHasStockList= locks.stream().map(item -> {
            SkuHasStock skuHasStock = new SkuHasStock();
            Long skuId = item.getSkuId();
            skuHasStock.setSkuId(skuId);
            skuHasStock.setNum(item.getCount());
            List<Long> wareIds = this.baseMapper.listWareIdHasSkuStock(skuId);
            skuHasStock.setWareIds(wareIds);
            return skuHasStock;
        }).collect(Collectors.toList());

        //2.遍历仓库id，锁定库存
        for (SkuHasStock skuHasStock : skuHasStockList) {
            List<Long> wareIds = skuHasStock.getWareIds();
            if(wareIds==null||wareIds.size()<=0){
                throw new NoStockException(skuHasStock.getSkuId());
            }
            boolean skuLocks=false;
            for (Long wareId : wareIds) {
               int count= this.baseMapper.lockStock(wareId,skuHasStock.getSkuId(),skuHasStock.getNum());
               if(count==1) {
                   //成功,那么就返回结束这次的循环，因为只需要锁定一个库存可以了
                   skuLocks = true;

                   //1.保存工作项订单
                   WareOrderTaskDetailEntity detailEntity = new WareOrderTaskDetailEntity(
                           null,
                           skuHasStock.getSkuId(),
                           "",
                           skuHasStock.getNum(),
                           taskEntity.getId(),
                           wareId,
                           1);
                   wareOrderTaskDetailService.save(detailEntity);
                   //2.创建工作单项
                   StockLockedTo stockLockedTo = new StockLockedTo();
                   stockLockedTo.setSkuId(skuHasStock.getSkuId());
                   StockDetailTo stockDetailTo = new StockDetailTo();
                   BeanUtils.copyProperties(detailEntity,stockDetailTo);
                   stockLockedTo.setDetailTo(stockDetailTo);

                   //3.发送到消息队列
                   rabbitTemplate.convertAndSend("stock-event-exchange","stock.locked",stockLockedTo);
                   break;
               }
            }
            //如果商品没有锁定成功那么就抛出异常。只要有一个出问题都不行
            if(!skuLocks){
                throw  new NoStockException(skuHasStock.getSkuId());
            }
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

  @RabbitHandler
    public void handle(StockLockedTo stockLockedTo, Message message, Channel channel) throws IOException {

        System.out.println("收到库存信息");
        StockDetailTo detailTo = stockLockedTo.getDetailTo();
        //1.查询detail是否存在
        Long detailId = detailTo.getId();
        WareOrderTaskDetailEntity detailEntity = wareOrderTaskDetailService.getById(detailId);
        if(detailEntity!=null){
            //如果存在那么就去解锁。
            //1.获取工作单，再获取订单
            Long id = detailTo.getId();
            WareOrderTaskEntity taskEntity = wareOrderTaskService.getById(id);
            String orderSn = taskEntity.getOrderSn();
            //2.查询订单是否存在，远程调用
            R r = orderFeignService.getOrder(orderSn);
            if(r.getCode()==0){
                //获取订单成功
                OrderVo data = r.getData(new TypeReference<OrderVo>() {
                });
                if(data==null||data.getStatus()==4){
                    //解锁信息
                    unLocked(detailTo.getSkuId(),detailTo.getWareId(),detailTo.getSkuNum(),detailId);
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                }
            }else{
                    //如果解锁失败需要重新把消息发送到消息队列中
                   channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
            }
        }else{
            //如果不存在那么就不做任何处理
           channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }
    }

    private void unLocked(Long skuId, Long wareId, Integer skuNum, Long detailId) {
        this.baseMapper.unLocked(skuId,wareId,skuNum);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

LoginInterceptor

  //如果是访问订单状态那么就放行。
        String requestURI = request.getRequestURI();
        boolean match = new AntPathMatcher().match("order/order/status/**", requestURI);
        if(match){
            return true;
        }
1
2
3
4
5
6

StockReleaseListener

@Service
@RabbitListener(queues = {"stock.release.stock.queue"})
public class StockReleaseListener {

    @Autowired
    WareSkuService wareSkuService;

    @RabbitHandler
    public void listenReleaseStock(StockLockedTo stockLockedTo, Channel channel, Message message) throws IOException {
        try {
            wareSkuService.unLockedStock(stockLockedTo);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {
            e.printStackTrace();
            //如果出现异常那么就把消息进行返回再次解锁
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
        }
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

12.定时关单

思路

①在创建订单之后发送消息给order延时队列，然后创建一个orderListener，如果超时之后发送到死信关单队列中，监听到这个消息，然后就进行关单操作

②关单操作就是修改订单的status为4

③然后就是库存队列过期之后就会查询订单是否存在，不存在那么就会解锁库存。

拓展与坑

问题：这种思路的问题就是如果订单的消息发送延时，导致解锁库存队列过期更快，然后查询订单发现没有取消，那么就不会解锁库存了。

解决方案：订单关单之后可以发送一个消息到解锁队列中提醒一次解锁。这样就能够避免这样的问题。

这种方案的思路：订单关单，然后发送消息到解锁库存队列，库存服务监听并且查询对应的工作单项，如果发现没有解锁，那么就进行解锁。主要是依靠工作单项的status来确定到底有没有解锁

orderLis

@RabbitListener(queues = {"order.release.order.queue"})
@Service
public class OrderListener {


    @Autowired
    OrderService orderService;

    @Autowired
    RabbitTemplate rabbitTemplate;


    @RabbitHandler
    public void listener(OrderEntity orderEntity, Channel channel, Message message) throws IOException {
        System.out.println("准备关闭订单号:"+orderEntity.getOrderSn());
        try {
            orderService.closeOrder(orderEntity);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            //订单释放后发送消息给库存解锁队列
            OrderTo orderTo = new OrderTo();
            BeanUtils.copyProperties(orderEntity,orderTo);
            rabbitTemplate.convertAndSend("order-event-exchange","order.release.other",orderTo);
        } catch (Exception e) {
            e.printStackTrace();
            //如果没有关闭成功那么就把消息返回去直到成功为止
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
        }

    }


}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

orderSer

//关单操作
 @Override
    public void closeOrder(OrderEntity orderEntity) {
        Long id = orderEntity.getId();
        //1.查询订单状态
        OrderEntity orderNow = this.getById(id);
        //如果状态还是待支付，那么就关单
        if(orderNow.getStatus()==OrderStatusEnum.CREATE_NEW.getCode()){
            OrderEntity update = new OrderEntity();
            update.setId(id);
            update.setStatus(OrderStatusEnum.CANCLED.getCode());
            this.updateById(update);
        }

    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

StockReleaseListener

@Service
@RabbitListener(queues = {"stock.release.stock.queue"})
public class StockReleaseListener {

    @Autowired
    WareSkuService wareSkuService;

    //库存消息过期之后查询订单是否存在解锁库存
    @RabbitHandler
    public void listenReleaseStock(StockLockedTo stockLockedTo, Channel channel, Message message) throws IOException {
        try {
            System.out.println("库存消息过期,查询订单是否消失，准备解锁库存");
            wareSkuService.unLockedStock(stockLockedTo);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {
            e.printStackTrace();
            //如果出现异常那么就把消息进行返回再次解锁
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
        }
    }

    //关单之后解锁库存的操作
    @RabbitHandler
    public void listenReleaseStock(OrderTo orderTo, Channel channel, Message message) throws IOException {
        try {
            System.out.println("订单释放，立刻解锁库存");
            wareSkuService.unLockedStockWhenReleaseOrder(orderTo);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {
            e.printStackTrace();
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
        }
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

//关单之后进行的一次解锁库存操作
@Transactional
    @Override
    public void unLockedStockWhenReleaseOrder(OrderTo orderTo) {
        //1.获取订单号
        String orderSn = orderTo.getOrderSn();
        //2.查询工作项
        WareOrderTaskEntity task = wareOrderTaskService.getOne(new QueryWrapper<WareOrderTaskEntity>().eq("order_sn", orderSn));
        Long id = task.getId();
        //3.根据工作项id查询所有的工作detail项，并且检查状态和解锁
        List<WareOrderTaskDetailEntity> details = wareOrderTaskDetailService.list(new QueryWrapper<WareOrderTaskDetailEntity>().eq("task_id", id));
        for (WareOrderTaskDetailEntity detail : details) {
            if(detail.getLockStatus()==2){
                continue;
            }else{
                unLocked(detail.getSkuId(),detail.getWareId(),detail.getSkuNum(),detail.getId());
            }
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

14.消息丢失、重复、积压

消息丢失

①网络原因，解决方案可以是在发送之前先把消息存入数据库，存入日志，然后定时发送

②消息去到brocker的时候，突然宕机，导致消息没有到达队列。解决方案就是可以通过生产者回调机制，p->b的confirm，然后就是b->q的报错，然后修改数据库中消息的报错状态，再次重新发送。

③消费者收到消息没有处理就宕机，解决方案就是手动提交

总结：解决消息丢失的关键

1.设置好确认回调机制

2.做好发送前消息的持久化处理，写入日志

消息重复

①处理业务成功在手动确认消息的时候宕机

②业务失败

解决方案;接口幂等性，接口只会进行一次修改处理。并且下次发生修改，数据不会发生变化。可以使用状态等判断。第二种方法就是防重表

消息积压

①消费者宕机，消费者处理能力不足

②生产速度太快