基于RabbitMQ，Redis，Redisson，RocketMQ四种技术实现订单延时关闭功能及其相关优缺点介绍（以12306为主题）

news2024/11/25 12:20:53

目录

1. 延迟关闭订单

1.1 订单延时关闭功能技术选型

1.1.1 定时任务

1.1.2 RabbitMQ

1.1.3 Redis 过期监听

1.1.4 Redisson

1.1.5 RocketMQ

1.2 RocketMQ订单延时关闭发送方实现

1.3 RocketMQ订单延时关闭的消费方实现

1. 延迟关闭订单

用户发起订单后，如果长时间未支付，需要将订单关闭，也就是大家常说的延时关闭订单。12306的PC端的延迟时间是10分钟:

1.1 订单延时关闭功能技术选型

1.1.1 定时任务

通过定时任务是一种常见的订单延迟关闭解决方案。

可以通过调度平台来实现定时任务的执行，具体任务是根据订单创建时间扫描所有到期的订单，并执行关闭订单的操作。

大家常用的定时任务调度平台有以下这些：

  • https://github.com/xuxueli/xxl-job

  • https://github.com/PowerJob/PowerJob

  • https://github.com/apache/shardingsphere-elasticjob

这种方案的优点在于简单易实现，但是，该方案也存在一些问题：

  1. 延迟时间不精确：使用定时任务执行订单关闭逻辑，无法保证订单在十分钟后准确地关闭。如果任务执行器在关闭订单的具体时间点出现问题，可能导致订单关闭的时间延后。

  2. 不适合高并发场景：定时任务执行的频率通常是固定的，无法根据实际订单的情况来灵活调整。在高并发场景下，可能导致大量的定时任务同时执行，造成系统负载过大。

  3. 分库分表问题：拿 12306 来说，订单表按照用户标识和订单号进行了分库分表，那这样的话，和上面说的根据订单创建时间去扫描一批订单进行关闭，自然就行不通。因为根据创建时间查询没有携带分片键，存在读扩散问题。

通常最不推荐的方式是使用定时任务来实现订单关闭。

1.1.2 RabbitMQ

RabbitMQ 是一个功能强大的消息中间件，通过使用 RabbitMQ 的延时消息特性，我们可以轻松实现订单十分钟延时关闭功能。首先，我们需要在 RabbitMQ 服务器上启用延时特性，通常通过安装 rabbitmq_delayed_message_exchange 插件来支持延时消息功能。

接下来，我们创建两个队列：订单队列和死信队列。订单队列用于存储需要延时关闭的订单消息，而死信队列则用于存储延时时间到达后的订单消息。在创建订单队列时，我们要为队列配置延时特性，指定订单消息的延时时间，比如十分钟。这样，当有新的订单需要延时关闭时，我们只需要将订单消息发送到订单队列，并设置消息的延时时间。

在订单队列中设置死信交换机和死信队列，当订单消息的延时时间到达后，消息会自动转发到死信队列，从而触发关闭订单的操作。在死信队列中，我们可以监听消息，并执行关闭订单的逻辑。为了确保消息的可靠性，可以在关闭订单操作前添加适当的幂等性措施，这样即使消息重复处理，也不会对系统产生影响。

通过以上步骤，我们就成功实现了订单的十分钟延时关闭功能。当有新的订单需要延时关闭时，将订单消息发送到订单队列，并设置延时时间。在延时时间到达后，订单消息会自动进入死信队列，从而触发关闭订单的操作。这种方式既简单又可靠，保证了系统的稳定性和可用性。

从整体来说 RabbitMQ 实现延时关闭订单功能是比较合适的，但也存在几个问题：

  1. 延时精度：RabbitMQ 的延时消息特性是基于消息的 TTL（Time-To-Live）来实现的，因此消息的延时时间并不是完全准确的，可能会有一定的误差。在处理订单十分钟延时关闭时，可能会有一些订单的关闭时间略晚于预期时间。

  2. 高并发问题：如果系统中有大量的订单需要延时关闭，而订单关闭操作非常复杂耗时，可能会导致消息队列中的消息堆积。这样就可能导致延时关闭操作无法及时处理，影响订单的实际关闭时间。

  3. 重复消息问题：由于网络原因或其他不可预知的因素，可能会导致消息重复发送到订单队列。如果没有处理好消息的幂等性，可能会导致订单重复关闭的问题，从而造成数据不一致或其他异常情况。

  4. 可靠性问题：RabbitMQ 是一个消息中间件，它是一个独立的系统。如果 RabbitMQ 本身出现故障或宕机，可能会导致订单延时关闭功能失效。因此，在使用 RabbitMQ 实现延时关闭功能时，需要考虑如何保证 RabbitMQ 的高可用性和稳定性。

延时精度和高并发属于一类问题，取决于客户端的消费能力。重复消费问题是所有消息中间件都需要解决，需要通过消息表等幂等解决方案解决。比较难搞定的是可用性问题，RabbitMQ 在可用性方面较弱，部分场景下会存在单点故障问题。

1.1.3 Redis 过期监听

可以借助 Redis 的过期消息监听机制实现延时关闭功能。

首先，在订单创建时，将订单信息存储到 Redis，并设置过期时间为十分钟。同时，在 Redis 中存储一个过期消息监听的键值对，键为订单号，值为待处理订单的标识。

其次，编写一个消息监听器，持续监听 Redis 的过期事件。监听器使用 Redis 的 PSUBSCRIBE 命令订阅过期事件，并在监听到过期事件时触发相应的处理逻辑。

当订单过期时间到达时，Redis 会自动触发过期事件，消息监听器捕获到该事件，并获取到过期的订单号。接着，监听器执行订单关闭的逻辑，如更新订单状态为关闭状态，释放相关资源等，实现订单的十分钟延时关闭功能。

需要注意的是，消息监听器应该是一个长期运行的任务，确保持续监听 Redis 的过期事件。为了保证系统的稳定性和可靠性，可以在实现订单关闭逻辑时添加容错机制，以应对 Redis 可能发生故障或重启的情况。

Redis 过期消息也存在几个问题：

  1. 不够精确：Redis 的过期时间是通过定时器实现的，可能存在一定的误差，导致订单的关闭时间不是精确的十分钟。这对于某些对时间要求较高的场景可能不适用。

  2. Redis 宕机：如果 Redis 宕机或重启，那些已经设置了过期时间但还未过期的订单信息将会丢失，导致这部分订单无法正确关闭。需要考虑如何处理这种异常情况。

  3. 可靠性：依赖 Redis 的过期时间来实现订单关闭功能，需要确保 Redis 的高可用性和稳定性。如果 Redis 发生故障或网络问题，可能导致订单关闭功能失效。

  4. 版本问题：Redis 5.0 之前是不保证延迟消息持久化的，如果客户端消费过程中宕机或者重启，这个消息不会重复投递。5.0 之后推出了 Stream 功能，有了持久化等比较完善的延迟消息功能。

1.1.4 Redisson

通过 Redisson 的 RDelayedQueue 功能可以实现订单十分钟延时关闭的功能。

首先，我们需要创建一个 RDelayedQueue 对象，用于存放需要延时关闭的订单信息。当用户创建订单时，我们将订单信息添加到 RDelayedQueue 中，并设置订单的延时时间为十分钟。

Redisson 提供了监听功能，可以实现对 RDelayedQueue 中订单信息的监听。一旦订单到达设定的延时时间，Redisson 会触发监听事件。在监听到订单的延时事件后，我们可以编写相应的处理逻辑，即关闭对应的订单。

在处理订单关闭时，我们可以根据订单号或订单创建时间等信息，来找到对应的订单进行关闭操作。

不过这种方式也不推荐使用，基本上 Redis 过期监听消息存在的问题，RDelayedQueue 也都会有，因为 RDelayedQueue 本质上也是依赖 Redis 实现。

1.1.5 RocketMQ

在订单生成时，我们将订单关闭消息发送到 RocketMQ，并设置消息的延迟时间为十分钟。RocketMQ 支持设置消息的延迟时间，可以通过设置消息的 delayLevel 来指定延迟级别，每个级别对应一种延迟时间。这样，订单关闭消息将在十分钟后自动被消费者接收到。

需要注意，RocketMQ 5.0 之后已经支持了自定义时间的延迟，而不仅是延迟级别范围内的时间。

为了处理订单关闭消息，我们需要在消费者端创建一个消息监听器。当消息监听器接收到订单关闭消息时，触发订单关闭操作，将订单状态设置为关闭状态。

需要注意的是，RocketMQ 的消息传递机制保证了消息的可靠性传递，因此消息可能会进行多次重试。为了确保订单关闭操作的幂等性，即多次执行不会产生副作用，我们需要在订单关闭逻辑中进行幂等性的处理。

铁路 12306 系统中，使用 RocketMQ 作为延时关闭订单的技术实现，从整体考虑，处理能力以及稳定性相对来说较为合适。

1.2 RocketMQ订单延时关闭发送方实现

在12306中,当订单添加完成之后, 在OrderService方,会延迟10分钟,发送一个关闭订单的消息给到MQ, 而关闭订单的消费方订阅这个消息,进行未及时支付的订单进行取消操作.

1) 我们创建一个延迟关闭订单的事件实体类 DelayCloseOrderEvent

​


package com.fs.fs12306.biz.orderservice.mq.event;
​
import com.fs.fs12306.biz.orderservice.dto.req.TicketOrderItemCreateReqDTO;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
​
import java.util.List;
​
/**
 * 延迟关闭订单事件
 */
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class DelayCloseOrderEvent {
​
    /**
     * 车次 ID
     */
    private String trainId;
​
    /**
     * 出发站点
     */
    private String departure;
​
    /**
     * 到达站点
     */
    private String arrival;
​
    /**
     * 订单号
     */
    private String orderSn;
​
    /**
     * 乘车人购票信息
     */
    private List<TicketOrderItemCreateReqDTO> trainPurchaseTicketResults;
}
​

2) 消息发送事件基础扩充属性实体 BaseSendExtendDTO

​


package com.fs.fs12306.biz.orderservice.mq.produce;
​
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
​
/**
 * 消息发送事件基础扩充属性实体
 *
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public final class BaseSendExtendDTO {
​
    /**
     * 事件名称
     */
    private String eventName;
​
    /**
     * 主题
     */
    private String topic;
​
    /**
     * 标签
     */
    private String tag;
​
    /**
     * 业务标识
     */
    private String keys;
​
    /**
     * 发送消息超时时间
     */
    private Long sentTimeout;
​
    /**
     * 延迟消息
     */
    private Integer delayLevel;
}
​

3) 把发送消息封装成一个模版类AbstractCommonSendProduceTemplate,目的是为了提供一个抽象的公共类，方便在项目中发送RocketMQ消息。

​


package com.fs.fs12306.biz.orderservice.mq.produce;
​
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
​
import java.util.Optional;
​
/**
 * RocketMQ 抽象公共发送消息组件
 *
 */
@Slf4j
@RequiredArgsConstructor
public abstract class AbstractCommonSendProduceTemplate<T> {
​
    private final RocketMQTemplate rocketMQTemplate;
​
    /**
     * 构建消息发送事件基础扩充属性实体
     * @param messageSendEvent 消息发送事件
     * @return 扩充属性实体
     */
    protected abstract BaseSendExtendDTO buildBaseSendExtendParam(T messageSendEvent);
​
    /**
     * 构建消息基本参数，请求头、Keys...
     *
     * @param messageSendEvent 消息发送事件
     * @param requestParam     扩充属性实体
     * @return 消息基本参数
     */
    protected abstract Message<?> buildMessage(T messageSendEvent, BaseSendExtendDTO requestParam);
​
    /**
     * 消息事件通用发送
     *
     * @param messageSendEvent 消息发送事件
     * @return 消息发送返回结果
     */
    public SendResult sendMessage(T messageSendEvent) {
        //构建基础扩充属性实体
        BaseSendExtendDTO baseSendExtendDTO = buildBaseSendExtendParam(messageSendEvent);
        SendResult sendResult;
        try {
            //构建消息的topic和tag,格式: topic:tag
            StringBuilder destinationBuilder = StrUtil.builder().append(baseSendExtendDTO.getTopic());
            if (StrUtil.isNotBlank(baseSendExtendDTO.getTag())) {
                destinationBuilder.append(":").append(baseSendExtendDTO.getTag());
            }
            //同步发送消息
            sendResult = rocketMQTemplate.syncSend(
                    destinationBuilder.toString(),
                    buildMessage(messageSendEvent, baseSendExtendDTO),
                    baseSendExtendDTO.getSentTimeout(),
                    Optional.ofNullable(baseSendExtendDTO.getDelayLevel()).orElse(0)
            );
            log.info("[{}] 消息发送结果：{}，消息ID：{}，消息Keys：{}", baseSendExtendDTO.getEventName(), sendResult.getSendStatus(), sendResult.getMsgId(), baseSendExtendDTO.getKeys());
        } catch (Throwable ex) {
            log.error("[{}] 消息发送失败，消息体：{}", baseSendExtendDTO.getEventName(), JSON.toJSONString(messageSendEvent), ex);
            throw ex;
        }
        return sendResult;
    }
}
​

AbstractCommonSendProduceTemplate定义了一个抽象的RocketMQ消息发送模板，具体的实现需要子类根据具体需求进行。通过这种方式，可以在不同的场景下复用消息发送的逻辑，提高代码的可维护性和可扩展性。

4) 编写延迟关闭订单生产者 DelayCloseOrderSendProduce

让延迟关闭订单生产者继承AbstractCommonSendProduceTemplate类,复用消息发送的逻辑

package com.fs.fs12306.biz.orderservice.mq.produce;
​
import cn.hutool.core.util.StrUtil;
import com.fs.fs12306.biz.orderservice.common.constant.OrderRocketMQConstant;
import com.fs.fs12306.biz.orderservice.mq.domain.MessageWrapper;
import com.fs.fs12306.biz.orderservice.mq.event.DelayCloseOrderEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
​
import java.util.UUID;
​
/**
 * 延迟关闭订单生产者
 *
 */
@Slf4j
@Component
public class DelayCloseOrderSendProduce extends AbstractCommonSendProduceTemplate<DelayCloseOrderEvent> {
​
    private final ConfigurableEnvironment environment;
​
    public DelayCloseOrderSendProduce(@Autowired RocketMQTemplate rocketMQTemplate, @Autowired ConfigurableEnvironment environment) {
        super(rocketMQTemplate);
        this.environment = environment;
    }
​
    @Override
    protected BaseSendExtendDTO buildBaseSendExtendParam(DelayCloseOrderEvent messageSendEvent) {
        return BaseSendExtendDTO.builder()
                .eventName("延迟关闭订单")
                .keys(messageSendEvent.getOrderSn())
                .topic(environment.resolvePlaceholders(OrderRocketMQConstant.ORDER_DELAY_CLOSE_TOPIC_KEY))
                .tag(environment.resolvePlaceholders(OrderRocketMQConstant.ORDER_DELAY_CLOSE_TAG_KEY))
                .sentTimeout(2000L)
                // RocketMQ 延迟消息级别 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
                .delayLevel(14)
                .build();
    }
​
    @Override
    protected Message<?> buildMessage(DelayCloseOrderEvent messageSendEvent, BaseSendExtendDTO requestParam) {
        String keys = StrUtil.isEmpty(requestParam.getKeys()) ? UUID.randomUUID().toString() : requestParam.getKeys();
        return MessageBuilder
                .withPayload(new MessageWrapper(requestParam.getKeys(), messageSendEvent))
                .setHeader(MessageConst.PROPERTY_KEYS, keys)
                .setHeader(MessageConst.PROPERTY_TAGS, requestParam.getTag())
                .build();
    }
}

对于消息的负载信息,我们进行再次封装: MessageWrapper

​


package com.fs.fs12306.biz.orderservice.mq.domain;
​
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
​
import java.io.Serializable;
import java.util.UUID;
​
/**
 * 消息体包装器
 *
 */
@Data
@Builder
@NoArgsConstructor(force = true)
@AllArgsConstructor
@RequiredArgsConstructor
public final class MessageWrapper<T> implements Serializable {
​
    private static final long serialVersionUID = 1L;
​
    /**
     * 消息发送 Keys
     */
    @NonNull
    private String keys;
​
    /**
     * 消息体
     */
    @NonNull
    private T message;
​
    /**
     * 唯一标识，用于客户端幂等验证
     */
    private String uuid = UUID.randomUUID().toString();
​
    /**
     * 消息发送时间
     */
    private Long timestamp = System.currentTimeMillis();
}
​

在往数据库插入订单相关的数据之后,我们就使用延时关闭订单的生产者延迟发送一个消息,从而实现延时关闭订单的功能

try {
            // 发送 RocketMQ 延时消息，指定时间后取消订单
            DelayCloseOrderEvent delayCloseOrderEvent = DelayCloseOrderEvent.builder()
                    .trainId(String.valueOf(requestParam.getTrainId()))
                    .departure(requestParam.getDeparture())
                    .arrival(requestParam.getArrival())
                    .orderSn(orderSn)
                    .trainPurchaseTicketResults(requestParam.getTicketOrderItems())
                    .build();
            SendResult sendResult = delayCloseOrderSendProduce.sendMessage(delayCloseOrderEvent);
            if (!Objects.equals(sendResult.getSendStatus(), SendStatus.SEND_OK)) {
                throw new ServiceException("投递延迟关闭订单消息队列失败");
            }
        } catch (Throwable ex) {
            log.error("延迟关闭订单消息队列发送错误，请求参数：{}", JSON.toJSONString(requestParam), ex);
            throw ex;
        }

1.3 RocketMQ订单延时关闭的消费方实现

12306中的关闭订单的消费方是在TicketService实现的.可能会有一个疑问，为什么延迟关闭订单发送端在订单服务，消费延迟关闭订单消息却是在购票服务？

如果消息的消费在订单服务中进行，那么订单服务就需要调用购票服务来进行座位的释放或订单状态的更新。这样就形成了一个远程循环依赖：订单服务依赖于购票服务，同时购票服务又依赖于订单服务。这对于微服务涉及中是需要尽可能避免的。所以，最终将消费延迟关闭消息放在了购票服务。

在ticketService服务中,也定义一个延时关闭订单的事件实体类, 与发送方的延时关闭订单事件的实体类是一样,

​


package com.fs.fs12306.biz.ticketservice.mq.event;
​
import com.fs.fs12306.biz.ticketservice.service.handler.ticket.dto.TrainPurchaseTicketRespDTO;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
​
import java.util.List;
​
/**
 * 延迟关闭订单事件
 *
 */
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class DelayCloseOrderEvent {
​
    /**
     * 车次 ID
     */
    private String trainId;
​
    /**
     * 出发站点
     */
    private String departure;
​
    /**
     * 到达站点
     */
    private String arrival;
​
    /**
     * 订单号
     */
    private String orderSn;
​
    /**
     * 乘车人购票信息
     */
    private List<TrainPurchaseTicketRespDTO> trainPurchaseTicketResults;
}
​

封装消息体包装器MessageWrapper

​


package com.fs.fs12306.biz.ticketservice.mq.domain;
​
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
​
import java.io.Serializable;
import java.util.UUID;
​
/**
 * 消息体包装器
 *
 */
@Data
@Builder
@NoArgsConstructor(force = true)
@AllArgsConstructor
@RequiredArgsConstructor
public final class MessageWrapper<T> implements Serializable {
​
    private static final long serialVersionUID = 1L;
​
    /**
     * 消息发送 Keys
     */
    @NonNull
    private String keys;
​
    /**
     * 消息体
     */
    @NonNull
    private T message;
​
    /**
     * 唯一标识，用于客户端幂等验证
     */
    private String uuid = UUID.randomUUID().toString();
​
    /**
     * 消息发送时间
     */
    private Long timestamp = System.currentTimeMillis();
}
​

这里有一个可以优化的地方,我们发现DelayCloseOrderEvent,MessageWrapper在OrderService和ticketService重复编写了, 可以把DelayCloseOrderEvent类抽取出来,放到一个公共模块中.

编写延时关闭订单的消费者DelayCloseOrderConsumer

​


package com.fs.fs12306.biz.ticketservice.mq.consumer;
​
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.fs.fs12306.biz.ticketservice.common.constant.TicketRocketMQConstant;
import com.fs.fs12306.biz.ticketservice.dto.domain.RouteDTO;
import com.fs.fs12306.biz.ticketservice.dto.req.CancelTicketOrderReqDTO;
import com.fs.fs12306.biz.ticketservice.mq.domain.MessageWrapper;
import com.fs.fs12306.biz.ticketservice.mq.event.DelayCloseOrderEvent;
import com.fs.fs12306.biz.ticketservice.remote.TicketOrderRemoteService;
import com.fs.fs12306.biz.ticketservice.remote.dto.TicketOrderDetailRespDTO;
import com.fs.fs12306.biz.ticketservice.remote.dto.TicketOrderPassengerDetailRespDTO;
import com.fs.fs12306.biz.ticketservice.service.SeatService;
import com.fs.fs12306.biz.ticketservice.service.TrainStationService;
import com.fs.fs12306.biz.ticketservice.service.handler.ticket.dto.TrainPurchaseTicketRespDTO;
import com.fs.fs12306.biz.ticketservice.service.handler.ticket.tokenbucket.TicketAvailabilityTokenBucket;
import com.fs.fs12306.frameworks.starter.cache.DistributedCache;
import com.fs.fs12306.frameworks.starter.common.util.BeanUtil;
import com.fs.fs12306.frameworks.starter.convention.result.Result;
import lombok.RequiredArgsConstructor;
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.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
​
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
​
import static com.fs.fs12306.biz.ticketservice.common.constant.RedisKeyConstant.TRAIN_STATION_REMAINING_TICKET;
​
/**
 * 延迟关闭订单消费者
 *
 */
@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = TicketRocketMQConstant.ORDER_DELAY_CLOSE_TOPIC_KEY,
        selectorExpression = TicketRocketMQConstant.ORDER_DELAY_CLOSE_TAG_KEY,
        consumerGroup = TicketRocketMQConstant.TICKET_DELAY_CLOSE_CG_KEY
)
public final class DelayCloseOrderConsumer implements RocketMQListener<MessageWrapper<DelayCloseOrderEvent>> {
​
    private final SeatService seatService;
    private final TicketOrderRemoteService ticketOrderRemoteService;
    private final TrainStationService trainStationService;
    private final DistributedCache distributedCache;
    private final TicketAvailabilityTokenBucket ticketAvailabilityTokenBucket;
​
    @Value("${ticket.availability.cache-update.type:}")
    private String ticketAvailabilityCacheUpdateType;
​
    @Override
    public void onMessage(MessageWrapper<DelayCloseOrderEvent> delayCloseOrderEventMessageWrapper) {
        log.info("[延迟关闭订单] 开始消费：{}", JSON.toJSONString(delayCloseOrderEventMessageWrapper));
        DelayCloseOrderEvent delayCloseOrderEvent = delayCloseOrderEventMessageWrapper.getMessage();
        String orderSn = delayCloseOrderEvent.getOrderSn();
        Result<Boolean> closedTickOrder;
        try {
            //远程调用OrderService的关闭订单的接口
            closedTickOrder = ticketOrderRemoteService.closeTickOrder(new CancelTicketOrderReqDTO(orderSn));
        } catch (Throwable ex) {
            log.error("[延迟关闭订单] 订单号：{} 远程调用订单服务失败", orderSn, ex);
            throw ex;
        }
        if (closedTickOrder.isSuccess() && !StrUtil.equals(ticketAvailabilityCacheUpdateType, "binlog")) {
            if (!closedTickOrder.getData()) {
                log.info("[延迟关闭订单] 订单号：{} 用户已支付订单", orderSn);
                return;
            }
            String trainId = delayCloseOrderEvent.getTrainId();
            String departure = delayCloseOrderEvent.getDeparture();
            String arrival = delayCloseOrderEvent.getArrival();
            List<TrainPurchaseTicketRespDTO> trainPurchaseTicketResults = delayCloseOrderEvent.getTrainPurchaseTicketResults();
            try {
                //释放锁定的座位
                seatService.unlock(trainId, departure, arrival, trainPurchaseTicketResults);
            } catch (Throwable ex) {
                log.error("[延迟关闭订单] 订单号：{} 回滚列车DB座位状态失败", orderSn, ex);
                throw ex;
            }
            try {
                StringRedisTemplate stringRedisTemplate = (StringRedisTemplate) distributedCache.getInstance();
                Map<Integer, List<TrainPurchaseTicketRespDTO>> seatTypeMap = trainPurchaseTicketResults.stream()
                        .collect(Collectors.groupingBy(TrainPurchaseTicketRespDTO::getSeatType));
                List<RouteDTO> routeDTOList = trainStationService.listTakeoutTrainStationRoute(trainId, departure, arrival);
                //缓存中的余票加
                routeDTOList.forEach(each -> {
                    String keySuffix = StrUtil.join("_", trainId, each.getStartStation(), each.getEndStation());
                    seatTypeMap.forEach((seatType, trainPurchaseTicketRespDTOList) -> {
                        stringRedisTemplate.opsForHash()
                                .increment(TRAIN_STATION_REMAINING_TICKET + keySuffix, String.valueOf(seatType), trainPurchaseTicketRespDTOList.size());
                    });
                });
                TicketOrderDetailRespDTO ticketOrderDetail = BeanUtil.convert(delayCloseOrderEvent, TicketOrderDetailRespDTO.class);
                ticketOrderDetail.setPassengerDetails(BeanUtil.convert(delayCloseOrderEvent.getTrainPurchaseTicketResults(), TicketOrderPassengerDetailRespDTO.class));
                ticketAvailabilityTokenBucket.rollbackInBucket(ticketOrderDetail);
            } catch (Throwable ex) {
                log.error("[延迟关闭订单] 订单号：{} 回滚列车Cache余票失败", orderSn, ex);
                throw ex;
            }
        }
    }
}
​

OrderService的feign client接口

​


package com.fs.fs12306.biz.ticketservice.remote;
​
import com.fs.fs12306.biz.ticketservice.dto.req.CancelTicketOrderReqDTO;
import com.fs.fs12306.biz.ticketservice.dto.req.TicketOrderItemQueryReqDTO;
import com.fs.fs12306.biz.ticketservice.remote.dto.TicketOrderCreateRemoteReqDTO;
import com.fs.fs12306.biz.ticketservice.remote.dto.TicketOrderDetailRespDTO;
import com.fs.fs12306.biz.ticketservice.remote.dto.TicketOrderPassengerDetailRespDTO;
import com.fs.fs12306.frameworks.starter.convention.result.Result;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.cloud.openfeign.SpringQueryMap;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import java.util.List;
​
/**
 * 车票订单远程服务调用
 *
 */
@FeignClient(value = "fs12306-order-service", url = "${aggregation.remote-url:}")
public interface TicketOrderRemoteService {
    /**
     * 创建车票订单
     *
     * @param requestParam 创建车票订单请求参数
     * @return 订单号
     */
    @PostMapping("/api/order-service/order/ticket/create")
    Result<String> createTicketOrder(@RequestBody TicketOrderCreateRemoteReqDTO requestParam);
​
    /**
     * 车票订单关闭
     *
     * @param requestParam 车票订单关闭入参
     * @return 关闭订单返回结果
     */
    @PostMapping("/api/order-service/order/ticket/close")
    Result<Boolean> closeTickOrder(@RequestBody CancelTicketOrderReqDTO requestParam);
​
    /**
     * 车票订单取消
     *
     * @param requestParam 车票订单取消入参
     * @return 订单取消返回结果
     */
    @PostMapping("/api/order-service/order/ticket/cancel")
    Result<Void> cancelTicketOrder(@RequestBody CancelTicketOrderReqDTO requestParam);
}
​

接下来需要在OrderService服务中,实现对订单关闭的业务实现

​


package com.fs.fs12306.biz.orderservice.controller;
​
import cn.crane4j.annotation.AutoOperate;
import com.fs.fs12306.biz.orderservice.dto.req.*;
import com.fs.fs12306.biz.orderservice.dto.resp.TicketOrderDetailRespDTO;
import com.fs.fs12306.biz.orderservice.dto.resp.TicketOrderDetailSelfRespDTO;
import com.fs.fs12306.biz.orderservice.dto.resp.TicketOrderPassengerDetailRespDTO;
import com.fs.fs12306.biz.orderservice.service.OrderItemService;
import com.fs.fs12306.biz.orderservice.service.OrderService;
import com.fs.fs12306.frameworks.starter.convention.page.PageResponse;
import com.fs.fs12306.frameworks.starter.convention.result.Result;
import com.fs.fs12306.frameworks.starter.web.Results;
import lombok.RequiredArgsConstructor;
​
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
​
import java.util.List;
​
/**
 * 车票订单接口控制层
 *
 */
@RestController
@RequiredArgsConstructor
public class TicketOrderController {
​
    private final OrderService orderService;
    private final OrderItemService orderItemService;
​
    /**
     * 根据订单号查询车票订单
     */
    @GetMapping("/api/order-service/order/ticket/query")
    public Result<TicketOrderDetailRespDTO> queryTicketOrderByOrderSn(@RequestParam(value = "orderSn") String orderSn) {
        return Results.success(orderService.queryTicketOrderByOrderSn(orderSn));
    }
​
    /**
     * 根据子订单记录id查询车票子订单详情
     */
    @GetMapping("/api/order-service/order/item/ticket/query")
    public Result<List<TicketOrderPassengerDetailRespDTO>> queryTicketItemOrderById(TicketOrderItemQueryReqDTO requestParam) {
        return Results.success(orderItemService.queryTicketItemOrderById(requestParam));
    }
​
    /**
     * 分页查询车票订单
     */
    @AutoOperate(type = TicketOrderDetailRespDTO.class, on = "data.records")
    @GetMapping("/api/order-service/order/ticket/page")
    public Result<PageResponse<TicketOrderDetailRespDTO>> pageTicketOrder(TicketOrderPageQueryReqDTO requestParam) {
        return Results.success(orderService.pageTicketOrder(requestParam));
    }
​
    /**
     * 分页查询本人车票订单
     */
    @GetMapping("/api/order-service/order/ticket/self/page")
    public Result<PageResponse<TicketOrderDetailSelfRespDTO>> pageSelfTicketOrder(TicketOrderSelfPageQueryReqDTO requestParam) {
        return Results.success(orderService.pageSelfTicketOrder(requestParam));
    }
​
    /**
     * 车票订单创建
     */
    @PostMapping("/api/order-service/order/ticket/create")
    public Result<String> createTicketOrder(@RequestBody TicketOrderCreateReqDTO requestParam) {
        return Results.success(orderService.createTicketOrder(requestParam));
    }
​
    /**
     * 车票订单关闭
     */
    @PostMapping("/api/order-service/order/ticket/close")
    public Result<Boolean> closeTickOrder(@RequestBody CancelTicketOrderReqDTO requestParam) {
        return Results.success(orderService.closeTickOrder(requestParam));
    }
​
    /**
     * 车票订单取消
     */
    @PostMapping("/api/order-service/order/ticket/cancel")
    public Result<Boolean> cancelTickOrder(@RequestBody CancelTicketOrderReqDTO requestParam) {
        return Results.success(orderService.cancelTickOrder(requestParam));
    }
}
​

关闭订单的业务实现:

OrderService接口定义关闭订单的方法:closeTickOrder

/**
     * 关闭火车票订单
     *
     * @param requestParam 关闭火车票订单入参
     */
boolean closeTickOrder(CancelTicketOrderReqDTO requestParam);
OrderServiceImpl类的具体closeTickOrder方法的实现如下:

@Override
    public boolean closeTickOrder(CancelTicketOrderReqDTO requestParam) {
        String orderSn = requestParam.getOrderSn();
        LambdaQueryWrapper<OrderDO> queryWrapper = Wrappers.lambdaQuery(OrderDO.class)
                .eq(OrderDO::getOrderSn, orderSn)
                .select(OrderDO::getStatus);
        OrderDO orderDO = orderMapper.selectOne(queryWrapper);
        if (Objects.isNull(orderDO) || orderDO.getStatus() != OrderStatusEnum.PENDING_PAYMENT.getStatus()) {
            return false;
        }
        // 原则上订单关闭和订单取消这两个方法可以复用，为了区分未来考虑到的场景，这里对方法进行拆分但复用逻辑
        return cancelTickOrder(requestParam);
    }
​
@Override
    public boolean cancelTickOrder(CancelTicketOrderReqDTO requestParam) {
        String orderSn = requestParam.getOrderSn();
        LambdaQueryWrapper<OrderDO> queryWrapper = Wrappers.lambdaQuery(OrderDO.class)
                .eq(OrderDO::getOrderSn, orderSn);
        OrderDO orderDO = orderMapper.selectOne(queryWrapper);
        if (orderDO == null) {
            throw new ServiceException(OrderCanalErrorCodeEnum.ORDER_CANAL_UNKNOWN_ERROR);
        } else if (orderDO.getStatus() != OrderStatusEnum.PENDING_PAYMENT.getStatus()) {
            throw new ServiceException(OrderCanalErrorCodeEnum.ORDER_CANAL_STATUS_ERROR);
        }
        RLock lock = redissonClient.getLock(StrBuilder.create("order:canal:order_sn_").append(orderSn).toString());
        if (!lock.tryLock()) {
            throw new ClientException(OrderCanalErrorCodeEnum.ORDER_CANAL_REPETITION_ERROR);
        }
        try {
            OrderDO updateOrderDO = new OrderDO();
            updateOrderDO.setStatus(OrderStatusEnum.CLOSED.getStatus());
            updateOrderDO.setOrderSn(orderSn);
            LambdaUpdateWrapper<OrderDO> updateWrapper = Wrappers.lambdaUpdate(OrderDO.class)
                    .eq(OrderDO::getOrderSn, orderSn);
            int updateResult = orderMapper.update(updateOrderDO, updateWrapper);
            if (updateResult <= 0) {
                throw new ServiceException(OrderCanalErrorCodeEnum.ORDER_CANAL_ERROR);
            }
            OrderItemDO updateOrderItemDO = new OrderItemDO();
            updateOrderItemDO.setStatus(OrderItemStatusEnum.CLOSED.getStatus());
            updateOrderItemDO.setOrderSn(orderSn);
            LambdaUpdateWrapper<OrderItemDO> updateItemWrapper = Wrappers.lambdaUpdate(OrderItemDO.class)
                    .eq(OrderItemDO::getOrderSn, orderSn);
            int updateItemResult = orderItemMapper.update(updateOrderItemDO, updateItemWrapper);
            if (updateItemResult <= 0) {
                throw new ServiceException(OrderCanalErrorCodeEnum.ORDER_CANAL_ERROR);
            }
        } finally {
            lock.unlock();
        }
        return true;
    }

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2223423.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

基于SSM的BBS社区论坛系统源码

基于SSM的BBS社区论坛系统源码

运行环境&#xff1a;ideamysql5.7jdk8maven 使用技术&#xff1a;ssmmysqlshirolayui 功能模块&#xff1a;用户管理、模板管理、帖子管理、公告管理、权限管理等
阅读更多...
使用 v-html 指令渲染的标签, 标签内绑定的 click 事件不生效

使用 v-html 指令渲染的标签, 标签内绑定的 click 事件不生效

背景 在项目开发中&#xff0c;实现用户友好的输入交互是提升用户体验的关键之一。例如&#xff0c;在客服对话框中&#xff0c;其中有包含多个快捷选项用于快速问答&#xff0c;每个快捷选项都是一个可点击的按钮&#xff0c;并需要绑定点击事件来执行相应操作。然而&#xf…
阅读更多...
Android Junit 单元测试 | 依赖配置和编译报错解决

Android Junit 单元测试 | 依赖配置和编译报错解决

问题 为什么在依赖中添加了testImplement在build APK的时候还是会报错&#xff1f;是因为没有识别到test文件夹是test源代码路径吗&#xff1f; 最常见的配置有: implementation - 所有源代码集(包括test源代码集)中都有该依赖库.testImplementation - 依赖关系仅在test源代码…
阅读更多...
【CSS in Depth 2 精译_054】8.2 CSS 层叠图层（cascade layer）的推荐组织方案

【CSS in Depth 2 精译_054】8.2 CSS 层叠图层（cascade layer）的推荐组织方案

当前内容所在位置&#xff08;可进入专栏查看其他译好的章节内容&#xff09; 【第三部分 现代 CSS 代码组织】 ✔️【第八章 层叠图层及其嵌套】 ✔️ 8.1 用 layer 图层来操控层叠规则&#xff08;上篇&#xff09; 8.1.1 图层的定义&#xff08;上篇&#xff09;8.1.2 图层的…
阅读更多...
华为云实战杂记

华为云实战杂记

配置nginx服务器 首先我们拿到一台服务器时&#xff0c;并不知道系统是否存在Nginx我们可以在Linux命令行执行如下命令查看 find / -name nginx* find / -name nginx* 查找所有名字以nginx开头的文件或者目录&#xff0c;我们看看系统里面都有哪些文件先&#xff0c;这样可以快…
阅读更多...
Linux系统安装Redis详细操作步骤（二进制发布包安装方式）

Linux系统安装Redis详细操作步骤（二进制发布包安装方式）

安装方式介绍 在Linux系统中&#xff0c;安装软件的方式主要有四种&#xff0c;这四种安装方式的特点如下&#xff1a; 安装方式特点二进制发布包安装软件已经针对具体平台编译打包发布&#xff0c;只要解压&#xff0c;修改配置即可rpm安装软件已经按照redhat的包管理规范进…
阅读更多...
雷池社区版有多个防护站点监听在同一个端口上，匹配顺序是怎么样的

雷池社区版有多个防护站点监听在同一个端口上，匹配顺序是怎么样的

如果域名处填写的分别为 IP 与域名&#xff0c;那么当使用进行 IP 请求时&#xff0c;则将会命中第一个配置的站点 以上图为例&#xff0c;如果用户使用 IP 访问&#xff0c;命中 example.com。 如果域名处填写的分别为域名与泛域名&#xff0c;除非准确命中域名&#xff0c;否…
阅读更多...
NGINX 保护 Web 应用安全之基于 IP 地址的访问

NGINX 保护 Web 应用安全之基于 IP 地址的访问

根据客户端的 IP 地址控制访问 使用 HTTP 或 stream 访问模块控制对受保护资源的访问&#xff1a; location /admin/ { deny 10.0.0.1; allow 10.0.0.0/20; allow 2001:0db8::/32; deny all; } } 给定的 location 代码块允许来自 10.0.0.0/20 中的任何 IPv4 地址访问&#xf…
阅读更多...
UE4_Niagara基础实例—9、使用条带渲染器来制作闪电

UE4_Niagara基础实例—9、使用条带渲染器来制作闪电

效果图&#xff1a; 一、通过模板Static Beam来熟悉条带渲染器 从Static Beam发射器新建niagara系统&#xff0c;更名为NS_StaticBeam。 打开粒子系统&#xff0c;界面如下&#xff1a; Beam Emitter Setup模块可以设置条带的开始点、结束点和切线。 我们就可以通过这个Beam E…
阅读更多...
自动化测试：等待方式

自动化测试：等待方式

在自动化测试中&#xff0c;等待是一个重要的技术&#xff0c;用于处理页面加载、元素定位、元素状态改变等延迟问题。 等待能够确保在条件满足后再进行后续操作&#xff0c;提高自动化测试的稳定性以及可靠性。 等待方式&#xff1a;显示等待、隐式等待、线程睡眠 1. 显式等…
阅读更多...
【python】OpenCV—WaterShed Algorithm（1）

【python】OpenCV—WaterShed Algorithm（1）

文章目录 1、功能描述2、代码实现3、完整代码4、效果展示5、涉及到的库函数5.1、cv2.pyrMeanShiftFiltering5.2、cv2.morphologyEx5.3、cv2.distanceTransform5.4、cv2.normalize5.5、cv2.watershed 6、参考 1、功能描述 基于分水岭算法对图片进行分割 分水岭分割算法&#x…
阅读更多...
什么是域名？什么是泛域名？

什么是域名？什么是泛域名？

域名 定义 域名是互联网上用于识别和定位网站或网络服务的名称。它是由一串用点分隔的字符组成&#xff0c;例如 “baidu.com”。就像是现实生活中建筑物的地址&#xff0c;方便用户在互联网的海量信息中找到特定的网站。 结构 域名从右到左依次为顶级域名&#xff08;TLD&…
阅读更多...
go语言中的Scan()和Scanln()输入函数

go语言中的Scan()和Scanln()输入函数

Scan()输入函数 package mainimport "fmt"func main() {var a intvar b stringfor {fmt.Println("请输入一个整数和一个字符串&#xff08;用空格分隔&#xff09;&#xff1a;")fmt.Scan(&a, &b) // 直接读取输入到变量中fmt.Println("整数…
阅读更多...
图书管理系统的简单实现

图书管理系统的简单实现

文章目录 图书系统逻辑分析各种操作功能的实现完整代码 个人主页 JavaSE专栏 图书系统逻辑分析 该程序设置有三个包&#xff0c;user 包&#xff0c;book 包&#xff0c;operation包。 book包中包含对于书的一些信息和操作;operation包中包含有对 书 的所有操作功能;user包中包…
阅读更多...
‘perl‘ 不是内部或外部命令，也不是可运行的程序 或批处理文件。

‘perl‘ 不是内部或外部命令，也不是可运行的程序 或批处理文件。

‘perl’ 不是内部或外部命令，也不是可运行的程序 或批处理文件。 明明已经根据教程安装了perl环境，但是在cmd中依赖报该错误，本章教程提供解决办法。 一、激活perl环境 state shell ActiveState-Perl-5.36.0此时输入perl -v 是可以直接输出perl版本号的。 二、找到perl的执…
阅读更多...
想进体制内？到底有哪些路可走？原来有这么多方法

想进体制内？到底有哪些路可走？原来有这么多方法

在如今的就业大环境下&#xff0c;体制内工作越来越受到大家的青睐。那么&#xff0c;体制内为何如此受欢迎呢&#xff1f; 一、体制内为何备受青睐 体制内工作首先给人一种强烈的稳定感和安全感。一旦进入体制&#xff0c;你不用时刻担心失业的风险&#xff0c;能够拥有一份长…
阅读更多...
安康旅游网站：SpringBoot设计与实现详解

安康旅游网站：SpringBoot设计与实现详解

目 录 目 录 I 摘 要 III Abstract IV 第一章 绪论 1 1.1 研究现状 1 1.2 设计原则 1 1.3 研究内容 2 第二章 相关技术简介 1 2.1 JSP技术 1 2.2 Java技术 2 2.3 MYSQL数据库 2 2.4 B/S结构 3 2.5 Spring Boot框架 4 第三章 系统分析 5 3.1可行性分析 5 3.1.1技术可行性 5 3.1.…
阅读更多...
阿里云项目启动OOM问题解决

阿里云项目启动OOM问题解决

问题描述 随着项目业务的增长&#xff0c;系统启动时内存紧张&#xff0c;每次第一次启动的时候就会出现oom第二次或者第n的时候&#xff0c;就启动成功了。 带着这个疑问&#xff0c;我就在阿里云上提交了工单&#xff0c;咨询为什么第一次提交失败但是后面却能提交成功尼&a…
阅读更多...
开挖 Domain - 前奏

开挖 Domain - 前奏

WPF App 主机配置 Microsot.Extension.Hosting 一键启动&#xff08;配置文件、依赖注入&#xff0c;日志&#xff09; // App.xaml.cs 中定义 IHost private readonly IHost _host Host.CreateDefaultBuilder().ConfigureAppConfiguration(c > {_ c.SetBasePath(Envi…
阅读更多...
基于NERF技术重建学习笔记

基于NERF技术重建学习笔记

NeRF&#xff08;Neural Radiance Fields&#xff09;是一种用于3D场景重建的神经网络模型&#xff0c;能够从2D图像生成逼真的3D渲染效果。它将场景表征为一个连续的5D函数&#xff0c;利用了体积渲染和神经网络的结合&#xff0c;通过学习光线穿过空间时的颜色和密度来重建场…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023