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

目录

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;
    }

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

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

校园表白墙源码修复版

校园表白墙源码修复版

此校园表白墙源码基于thinkphp&#xff0c;因为时代久远有不少bug&#xff0c;经本人修复已去除大部分bug&#xff0c;添加了美化元素。 https://pan.quark.cn/s/1f9b3564c84b https://pan.baidu.com/s/1bb9vu9VV2jJoo9-GF6W3xw?pwd7293 https://caiyun.139.com/m/i?2hoTc…
阅读更多...
Etcd 可观测最佳实践

Etcd 可观测最佳实践

简介 Etcd 是一个高可用的分布式键值存储系统&#xff0c;它提供了一个可靠的、强一致性的存储服务&#xff0c;用于配置管理和服务发现。它最初由 CoreOS 开发&#xff0c;现在由 Cloud Native Computing Foundation (CNCF) 维护。Etcd 使用 Raft 算法来实现数据的一致性&…
阅读更多...
AI运动抽取转3D动画

AI运动抽取转3D动画

用的Wonder Studio&#xff0c;不免费哈&#xff0c;也不能试了&#xff0c;都要钱。只能看别人的经验。 网址https://wonderdynamics.com/ 先别激动&#xff0c;参考别人的评论 登录后&#xff08;google登录&#xff09;界面如下 收费情况 视频教程 https://www.youtube.co…
阅读更多...
【记录】VSCode｜自用设置项

【记录】VSCode｜自用设置项

文章目录 1 基础配置1.1 自动保存1.2 编辑区自动换行1.3 选项卡换行1.4 空格代替制表符1.5 开启滚轮缩放 2 进阶设置2.1 选项卡不自我覆盖2.2 选项卡限制宽度2.3 选项卡组限制高度2.4 字体设置2.5 字体加粗2.6 侧边栏2.7 沉浸式代码模式 Zen Mode2.8 设置 Zen 模式的选项卡组 3…
阅读更多...
深入了解 kotlinx-datetime：配置与使用指南

深入了解 kotlinx-datetime：配置与使用指南

深入了解 kotlinx-datetime&#xff1a;配置与使用指南 在Kotlin多平台开发中&#xff0c;处理日期和时间是常见的需求。kotlinx-datetime库提供了强大且简洁的API来帮助开发者应对这一挑战。本文将详细介绍如何配置kotlinx-datetime库&#xff0c;并通过生动的示例演示其核心…
阅读更多...
远程：HTTP基本身份验证失败。提供的密码或令牌不正确，或者您的账户启用了两步验证，您必须使用个人访问令牌而不是密码。

远程：HTTP基本身份验证失败。提供的密码或令牌不正确，或者您的账户启用了两步验证，您必须使用个人访问令牌而不是密码。

问题描述&#xff1a; remote: HTTP Basic: Access denied. The provided password or token is incorrect or your account has 2FA enabled and you must use a personal access token insteadof a password. See http://gitlab.cnovit.com/help/topics/git/troubleshooting…
阅读更多...
从 Hadoop 迁移到数据 Lakehouse 的架构师指南

从 Hadoop 迁移到数据 Lakehouse 的架构师指南

从 Hadoop 到数据湖仓一体架构的演变代表了数据基础架构的重大飞跃。虽然 Hadoop 曾经以其强大的批处理能力统治着大数据领域&#xff0c;但如今的组织正在寻求更敏捷、更具成本效益和现代化的解决方案。尤其是当他们越来越多地开始实施 AI 计划时。根本没有办法让 Hadoop 为 A…
阅读更多...
C++ 类与对象（中） 默认成员函数

C++ 类与对象（中） 默认成员函数

我们知道在类中&#xff0c;有成员变量和成员函数&#xff0c;我们可以通过创造不同的成员函数来实现这个类不同的功能&#xff0c;如果我们创造一个类&#xff0c;却不实现它的成员函数会如何呢&#xff1f;这个就涉及到类中的默认成员函数的概念了。但在本文我们主要介绍以下…
阅读更多...
【Dash】feffery_antd_components 按钮组件的应用

【Dash】feffery_antd_components 按钮组件的应用

一、feffery_antd_componenet 中的 AntdFloatButton 和 AntdFloatButtonGroup AntdFloatButton 和 AntdFloatButtonGroup 是两个用于创建悬浮按钮和悬浮按钮组的组件。 AntdFloatButton 是单个悬浮按钮组件&#xff0c;它提供了多种属性来定义按钮的外观及行为。AntdFloatBut…
阅读更多...
WebView渲染异常导致闪退解决方案

WebView渲染异常导致闪退解决方案

背景&#xff1a; App主页面使用了大量WebView容器(10个以上)显示图表信息&#xff0c;最新发现bugly上面出现一些关于浏览器Native Crash&#xff0c;如下&#xff1a; 经排查&#xff0c;是WebView渲染失败导致Crash&#xff0c;可以通过webView.loadUrl("chrome://cra…
阅读更多...
Pycharm搭配miniConda(Anaconda的轻量版本)建立虚拟环境管理不同项目不同依赖

Pycharm搭配miniConda(Anaconda的轻量版本)建立虚拟环境管理不同项目不同依赖

需求 pip安装的版本会在全局环境下生效,导致不用的项目使用的是同一套环境,容易出现别人的项目跑的好好的,到了自己电脑就不能跑的问题需求 1.不同的项目可以设置不同的环境,同时支持自动切换环境2.便捷,轻量化管理当前的环境 如果可以,最好有一个可视化的平台,anaconda倒是支…
阅读更多...
图像分割-DeepLab

图像分割-DeepLab

DeepLabV3源码链接&#xff1a;https://github.com/bubbliiiing/deeplabv3-plus-pytorch&#xff08;打不开私信我获取&#xff09; 一、简介 一般的模型如Unet一般用于医学领域&#xff0c;小目标&#xff0c;如细胞分割。 为了增大感受野&#xff08;从而更好的获得全局特征…
阅读更多...
docker 镜像详解

docker 镜像详解

Docker镜像是一种轻量级、可移植的软件打包格式&#xff0c;它包含了运行应用程序所需的一切&#xff0c;是构建和分发应用程序的基础。以下是对Docker镜像的详细解释&#xff1a; 一、镜像的定义 镜像本质上是一个只读文件&#xff0c;包含了文件系统、源码、库文件、依赖、…
阅读更多...
wordcloud 字体报错

wordcloud 字体报错

wordcloud 字体报错 词云库报错&#xff1a;Only supported for TrueType fonts字体文件问题pillow版本的问题wordcloud版本问题&#xff08;我的最终解决方案&#xff09; 词云库报错&#xff1a;Only supported for TrueType fonts 字体文件问题 解决方法 写绝对路径 &…
阅读更多...
【CSS3】css开篇基础（4）

【CSS3】css开篇基础（4）

1.❤️❤️前言~&#x1f973;&#x1f389;&#x1f389;&#x1f389; Hello, Hello~ 亲爱的朋友们&#x1f44b;&#x1f44b;&#xff0c;这里是E绵绵呀✍️✍️。 如果你喜欢这篇文章&#xff0c;请别吝啬你的点赞❤️❤️和收藏&#x1f4d6;&#x1f4d6;。如果你对我的…
阅读更多...
二百七十一、Kettle——ClickHouse增量导入数据清洗记录表

二百七十一、Kettle——ClickHouse增量导入数据清洗记录表

一、目的 在完成错误数据表任务后&#xff0c;需要对每条错误数据的错误字段及其字段值进行分析 Hive中原有SQL语句和ClickHouse现有SQL语句很大不同 二、Hive中原有代码 2.1 表结构 --31、静态排队数据清洗记录表 create table if not exists hurys_db.dwd_data_clean_…
阅读更多...
洞察前沿趋势！2024深圳国际金融科技大赛——西丽湖金融科技大学生挑战赛技术公开课指南

洞察前沿趋势！2024深圳国际金融科技大赛——西丽湖金融科技大学生挑战赛技术公开课指南

在当前信息技术与“互联网”深度融合的背景下&#xff0c;金融行业的转型升级是热门话题&#xff0c;创新与发展成为金融科技主旋律。随着区块链技术、人工智能技术、5G通信技术、大数据技术等前沿科技的飞速发展&#xff0c;它们与金融领域的深度融合&#xff0c;正引领着新型…
阅读更多...
uniapp：sqlite最详细教程，小白可直接粘贴复制

uniapp：sqlite最详细教程，小白可直接粘贴复制

新建uniapp项目，需要4个页面， loading 启动页：打开数据库，判断数据表是否存在，表内是否有数据，创建数据表的逻辑。 register 注册页：数据表已存在，但是没有数据，需要进入该页面注册第一条数据 index 首页：展示数据列表内的数据，可修改默认，添加新数据 edit 编辑：编…
阅读更多...
linux面试题复习

linux面试题复习

前言 现在只是初版&#xff0c;很多格式我还没有改好&#xff0c;会慢慢修改订正。 可能用到的网址&#xff1a;在线 EXCEL 到 MARKDOWN 转换器。 参考了很多网上的面试题和外网上的面试题&#xff1a; 参考文档&#xff1a; 程序员的50大Linux面试问题及答案 Top 60 Linux …
阅读更多...
机床发那科转profinet IO项目案例

机床发那科转profinet IO项目案例

目录 1 案例说明 1 2 VFBOX网关工作原理 1 3 准备工作 2 4 网关采集发那科机床数据 2 5 用PROFINET IO协议转发数据 5 6 案例总结 7 1 案例说明 设置网关采集发那科机床数据把采集的数据转成profinet IO协议转发给其他系统。 2 VFBOX网关工作原理 VFBOX网关是协议转换网关&a…
阅读更多...
最新文章

Copyright @ 2022~2023