分布式环境下定时任务扫描时间段模板创建可预订时间段

🎯 本文详细介绍了场馆预定系统中时间段生成的实现方案。通过设计场馆表、时间段模板表和时间段表，系统能够根据场馆的提前预定天数生成未来可预定的时间段。为了确保任务执行的唯一性和高效性，系统采用分布式锁机制和定时任务，避免重复生成时间段。通过流式查询优化大数据处理，减少内存占用和网络延迟。同时，使用唯一复合索引保证时间段生成的幂等性，避免重复插入。为提高系统性能，引入二级缓存和Redis管道技术，加速数据查询和缓存预热，确保用户在预定时间段时获得快速响应。整体方案兼顾了系统的稳定性、高效性和可扩展性。

简介

在场馆管理员创建了时间段模板之后，需要使用定时任务，每天定时生成未来可接受预定的时间段

数据表设计

【场馆表】

该表存储了 提前可预定天数（advance_booking_day），为什么要存储这个，是因为生成时间段的时候，需要参考该字段生成多少天内的时间段

DROP TABLE IF EXISTS `venue`;
CREATE TABLE `venue`(
    `id` bigint NOT NULL COMMENT 'ID',
    `create_time` datetime,
    `update_time` datetime,
    `is_deleted` tinyint default 0 COMMENT '逻辑删除 0：没删除 1：已删除',
    `organization_id` bigint NOT NULL COMMENT '所属机构ID',
    `name` varchar(30) NOT NULL COMMENT '场馆名称',
    `type` char(4) NOT NULL COMMENT '场馆类型 1:篮球馆（场） 2:足球场 3：羽毛球馆（场） 4:排球馆（场）100：体育馆 1000:其他',
    `address` varchar(255) NOT NULL COMMENT '场馆地址',
    `description` varchar(255) DEFAULT '' COMMENT '场馆描述，也可以说是否提供器材等等',
    `open_time` varchar(2000) NOT NULL COMMENT '场馆营业时间',
    `phone_number` varchar(11) NULL DEFAULT '' COMMENT '联系电话',
    `status` tinyint NOT NULL COMMENT '场馆状态 0：关闭 1：开放 2：维护中',
    `is_open` tinyint NOT NULL COMMENT '是否对外开放 0：否 1：是 如果不对外开放，需要相同机构的用户才可以预定',
    `advance_booking_day` int NOT NULL COMMENT '提前可预定天数，例如设置为1，即今天可预订明天的场',
    `start_booking_time` time NOT NULL COMMENT '开放预订时间',
     PRIMARY KEY (`id`) USING BTREE
)

【时间段模板表】

在这里需要使用 已生成到的日期（last_generated_date）来记录已经生成到的日期，避免重复生成时间段。比如说advance_booking_day=7，在 1月1 的时候，其实就已经生成了 [1月2, 1月8] 的时间段数据，那 1月2 的时候，其实只需要生成 1月9 的时间段即可

DROP TABLE IF EXISTS `time_period_model`;
CREATE TABLE `time_period_model`( 
    `id` bigint NOT NULL COMMENT 'ID',
    `create_time` datetime,
    `update_time` datetime,
    `is_deleted` tinyint default 0 COMMENT '逻辑删除 0：没删除 1：已删除',
    `price` decimal(10,2) NOT NULL COMMENT '该时间段预订使用价格（元）',
    `partition_id` bigint NOT NULL COMMENT '场区id',
    `begin_time` time NOT NULL COMMENT '时间段开始时间HH:mm（不用填日期）',
    `end_time` time NOT NULL COMMENT '时间段结束时间HH:mm（不用填日期）', 
    `effective_start_date` date NOT NULL COMMENT '生效开始日期', 
    `effective_end_date` date NOT NULL COMMENT '生效结束日期', 
    `last_generated_date` date COMMENT '已生成到的日期',
    `status` tinyint default 0 COMMENT '0：启用；1：停用', 
     PRIMARY KEY (`id`) USING BTREE,
     INDEX `idx_partition_id` (`partition_id`)
);

【时间段表】

之所以要创建唯一复合索引，是因为怕重复生成时间段，为什么会出现重复生成时间段的情况，看到后面就明白了

DROP TABLE IF EXISTS `time_period`;
CREATE TABLE `time_period`( 
    `id` bigint NOT NULL COMMENT 'ID',
    `create_time` datetime,
    `update_time` datetime,
    `is_deleted` tinyint default 0 COMMENT '逻辑删除 0：没删除 1：已删除',
    `partition_id` bigint NOT NULL COMMENT '场区id',
    `price` decimal(10,2) NOT NULL COMMENT '该时间段预订使用价格（元）',
    `stock` int NOT NULL COMMENT '库存',
    `booked_slots` bigint unsigned NOT NULL DEFAULT 0 COMMENT '已预订的场地（位图表示）',
    `period_date` date NOT NULL COMMENT '预定日期', 
    `begin_time` time NOT NULL COMMENT '时间段开始时间HH:mm（不用填日期）',
    `end_time` time NOT NULL COMMENT '时间段结束时间HH:mm（不用填日期）', 
     PRIMARY KEY (`id`) USING BTREE,
     INDEX `idx_partition_id` (`partition_id`),
     UNIQUE INDEX `idx_unique_partition_period_time` (`partition_id`, `period_date`, `begin_time`, `end_time`)
);

定时任务

由于项目是微服务架构，在开发定时任务的时候需要考虑任务执行的唯一性，否则集群的所有机器都会执行同一个任务，浪费计算机算力，可以通过直接加一个分布式锁，只让集群中的一台机器执行整个定时任务，但是如果这样的话，其他机器处于空闲状态。为了提高机器的利用率和定时任务的执行效率，这里将不同表的时间段模板扫描工作交与不同机器实现

【实现思路】

• 第一次定时任务：机器通过对表加锁，如果可以加锁成功，则扫描时间段模板进行时间段生成。注意，加锁时设置过期时间为2小时，如果任务执行完成，将锁状态设置1，同时过期时间设置更长
• 第二次定时任务：用来兜底，避免第一次定时任务执行时有机器宕机，其负责的任务并没有完成。第二次扫描就是找出没有执行完成的任务，重新执行一遍。在第二次定时任务时，要么任务已经执行完成，其任务状态被设置为1，要么没有执行，其锁状态已经过期

package com.vrs.config.scheduled;

import com.vrs.constant.RedisCacheConstant;
import com.vrs.service.TimePeriodModelService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * @Author dam
 * @create 2024/11/17 16:44
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class TimePeriodScheduledTasks {

    private final TimePeriodModelService timePeriodModelService;
    private final StringRedisTemplate stringRedisTemplate;
    private int tableNum = 16;

    /**
     * 在每天凌晨1点执行
     * 扫描数据库的时间段模板，生成可预定的时间段
     */
    @Scheduled(cron = "0 0 1 * * ?")
    public void timePeriodGenerator1() {
        for (int i = 0; i < tableNum; i++) {
            timePeriodGenerate(i);
        }
    }

    /**
     * 兜底一次，保证时间段都生成成功了
     * 扫描数据库的时间段模板，生成可预定的时间段
     */
    @Scheduled(cron = "0 0 4 * * ?")
    public void timePeriodGenerator2() {
        for (int i = 0; i < tableNum; i++) {
            // 获取当前表的状态
            String status = stringRedisTemplate.opsForValue().get(String.format(RedisCacheConstant.VENUE_TIME_PERIOD_GENERATE_KEY, i));

            // 如果状态为 "1"，说明任务已经完成，跳过
            if ("1".equals(status)) {
                continue;
            }

            // 如果还没有完成，尝试获取锁并执行任务
            timePeriodGenerate(i);
        }
    }

    /**
     * 时间段生成
     * @param tableIndex
     */
    private void timePeriodGenerate(int tableIndex) {
        // 0状态设置2小时就过期，方便没有执行完成的任务在兜底时可以重新执行
        boolean isSuccess = stringRedisTemplate.opsForValue().setIfAbsent(
                String.format(RedisCacheConstant.VENUE_TIME_PERIOD_GENERATE_KEY, tableIndex),
                "0", 2, TimeUnit.HOURS
        ).booleanValue();

        if (isSuccess) {
            try {
                // --if-- 设置键成功，说明集群中的其他机器还没有扫描当前表，由当前机器执行
                // 执行时间段生成
                timePeriodModelService.generateTimePeriodByModel(tableIndex);

                // 时间段生成成功，设置状态为1，过期时间段也设置长一点，方便兜底时检测
                stringRedisTemplate.opsForValue().set(
                        String.format(RedisCacheConstant.VENUE_TIME_PERIOD_GENERATE_KEY, tableIndex),
                        "1", 6, TimeUnit.HOURS
                );
            } catch (Exception e) {
                // 如果任务执行失败，删除锁，以便其他机器可以重试
                stringRedisTemplate.delete(String.format(RedisCacheConstant.VENUE_TIME_PERIOD_GENERATE_KEY, tableIndex));
                log.error("时间段生成失败，表编号：{}，错误信息：{}", tableIndex, e.getMessage(), e);
            }
        } else {
            // 如果锁已被占用，检查任务是否已完成
            String currentStatus = stringRedisTemplate.opsForValue().get(String.format(RedisCacheConstant.VENUE_TIME_PERIOD_GENERATE_KEY, tableIndex));
            if (!"1".equals(currentStatus)) {
                log.warn("表编号：{} 的任务未完成，但锁已被占用，可能由其他机器处理中", tableIndex);
            }
        }
    }
}

根据时间段模板创建时间段

MySQL流式查询

首先要解决的事情是：如何高效扫描时间段模板表？

很容易想到的是，分页查询时间段模板，一批一批进行时间段生成，还可以套一个多线程，并行执行不同批次的任务。但分页查询有一个最大的缺点，在处理大数据时，每次分页都需要执行完整的查询并跳过前面的记录，尤其是深分页时，查询效率非常低下。

为了优化这个问题，本文使用流式查询来处理。

流式查询是一种处理和传输查询结果的方式，它允许客户端逐行接收来自数据库的数据，而不是一次性获取整个结果集。这种方式可以显著减少内存占用和网络延迟，优化资源使用，并且能够更快地响应用户请求，特别适合处理大规模数据集或实时性要求高的应用场景。通过流式查询，即使面对海量数据，应用程序也能保持高效和流畅的用户体验。相较于分页查询，它效率高的一个原因是，它每次查询都是从上一条数据开始，而不是每次从头来过

时间段生成如何保证幂等性

由于在扫描时间段模板生成时间段时，服务器可能发生宕机，在兜底时，可能该任务再次被集群中的其他机器执行。但是由于部分时间段模板其实已经被第一台机器扫描过，相应的时间段也已经创建完成，在兜底时，如何保证已创建的时间段不会再被重复生成，最简单的一种实现方式：给partition_id, period_date, begin_time, end_time生成唯一复合索引，这样重复创建时间段时，插入数据库就会失败。

还有一个问题，为了提高数据插入效率，使用了批量插入策略，即累积一定的数据量才进行插入。但 Mybatis Plus 提供的saveBatch函数是原子性的，要么全部插入成功，要么全部插入失败。由于部分时间段可能已经被第一台机器创建完成，兜底时插入数据库可能出现唯一索引异常，这样会导致其他没有创建过的时间段也插入失败。因此我们需要自己实现一段SQL，让批量插入时，及时部分数据插入异常，也不影响其他数据

<insert id="insertBatchIgnore">
    INSERT IGNORE INTO time_period (
    id,
    create_time,
    update_time,
    is_deleted,
    partition_id,
    price,
    stock,
    booked_slots,
    period_date,
    begin_time,
    end_time
    ) VALUES
    <foreach collection="timePeriodDOList" item="item" separator=",">
        (
        #{item.id},
        NOW(),
        NOW(),
        0,
        #{item.partitionId},
        #{item.price},
        #{item.stock},
        #{item.bookedSlots},
        #{item.periodDate},
        #{item.beginTime},
        #{item.endTime}
        )
    </foreach>
</insert>

二级缓存

在生成未来时间段时，需要查询advance_booking_day才知道要生成未来多少天的时间段，由于很多时间段模板可能都来源于同一个分区、同一个场馆，为了加速这个查询，可以使用缓存来提高效率，即第一次查询之后将分区ID对应的场馆信息存储起来，第二次获取就很快了。

但如果每次从Redis缓存中加载数据，需要多次网络I/O，为了进一步的效率提升，可以使用本地缓存 HashMap 来进一步优化，即每次先从本地缓存中查询，查询不到再去 Redis 缓存中加载。

二级缓存策略：

• 首先使用本地缓存（如 HashMap）存储分区ID对应的场馆信息，以实现极快的查询速度；
• 当本地缓存未命中时，再从Redis缓存中加载数据。Redis作为第二级缓存，提供了比数据库更快的数据访问速度，并且能够跨多个实例共享缓存数据，确保了数据的一致性和高可用性。

Redis管道

在创建时间段时，还需要做的一件事是缓存预热，即将相应的时间段库存、时间段信息添加到 Redis 缓存中，保证用户在预定时间段时有较快的响应速度，而不是预定时再去数据库中查询放到缓存中。

由于需要添加大量时间段的缓存，如果每个数据都单独提交给 Redis，会导致大量的网络 I/O 操作，从而降低效率。因此，是否有一种类似于数据库批量插入的方式来优化这一过程？

Redis 管道（Pipeline）是一种用于优化客户端与 Redis 服务器之间通信的技术，它允许客户端一次性发送多个命令给服务器，并在所有命令执行完毕后一次性接收所有的回复。这种方式减少了客户端与服务器之间的往返时间，尤其是在需要执行大量命令时，能够显著提高性能。

/**
 * 使用管道来批量将数据存储到Redis中
 *
 * @param timePeriodDOList
 */
@Override
public void batchPublishTimePeriod(List<TimePeriodDO> timePeriodDOList) {
    if (timePeriodDOList == null || timePeriodDOList.size() == 0) {
        return;
    }
    /// 将时间段存放到数据库中
//        this.saveBatch(timePeriodDOList);
    baseMapper.insertBatchIgnore(timePeriodDOList);

    /// 将时间段信息放到缓存中
    // 创建一个管道回调
    RedisCallback<Void> pipelineCallback = connection -> {
        // 开始管道
        connection.openPipeline();
        for (TimePeriodDO timePeriodDO : timePeriodDOList) {
            // 时间段开始时间
            long timePeriodStartMill = DateUtil.combineLocalDateAndLocalTimeToDateTimeMill(timePeriodDO.getPeriodDate(), timePeriodDO.getBeginTime());
            // 计算从现在到时间段开始还有多少毫秒 + 余量（86400000表示一天）
            //todo 待确认 cacheTimeSecond 是否一定为正数
            long cacheTimeSecond = (timePeriodStartMill - System.currentTimeMillis() + 86400000) / 1000;
            // 时间段信息
            connection.setEx(
                    String.format(RedisCacheConstant.VENUE_TIME_PERIOD_KEY, timePeriodDO.getId()).getBytes(),
                    cacheTimeSecond,
                    JSON.toJSONString(timePeriodDO).getBytes()
            );
            // 库存
            connection.setEx(
                    String.format(RedisCacheConstant.VENUE_TIME_PERIOD_STOCK_KEY, timePeriodDO.getId()).getBytes(),
                    cacheTimeSecond,
                    JSON.toJSONString(timePeriodDO.getStock()).getBytes()
            );
        }
        // 执行管道中的所有命令
        connection.closePipeline();
        return null;
    };
    // 使用StringRedisTemplate执行管道回调
    stringRedisTemplate.execute(pipelineCallback);
}

代码实现

package com.vrs.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.vrs.constant.RedisCacheConstant;
import com.vrs.convention.exception.ClientException;
import com.vrs.convention.page.PageResponse;
import com.vrs.convention.page.PageUtil;
import com.vrs.domain.dto.req.TimePeriodModelListReqDTO;
import com.vrs.domain.entity.PartitionDO;
import com.vrs.domain.entity.TimePeriodDO;
import com.vrs.domain.entity.TimePeriodModelDO;
import com.vrs.domain.entity.VenueDO;
import com.vrs.mapper.TimePeriodModelMapper;
import com.vrs.service.PartitionService;
import com.vrs.service.TimePeriodModelService;
import com.vrs.service.TimePeriodService;
import com.vrs.service.VenueService;
import com.vrs.utils.DateUtil;
import com.vrs.utils.SnowflakeIdUtil;
import groovy.util.logging.Slf4j;
import lombok.Cleanup;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;

/**
 * @author dam
 * @description 针对表【time_period_model_0】的数据库操作Service实现
 * @createDate 2024-11-17 14:29:46
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class TimePeriodModelServiceImpl extends ServiceImpl<TimePeriodModelMapper, TimePeriodModelDO>
        implements TimePeriodModelService {

    private final DataSource dataSource;
    private final TimePeriodService timePeriodService;
    private final StringRedisTemplate stringRedisTemplate;
    private final PartitionService partitionService;
    private final VenueService venueService;

    /**
     * 流式处理
     */
    @Override
    @SneakyThrows
    public void generateTimePeriodByModel(int tableIndex) {
        // 获取 dataSource Bean 的连接
        @Cleanup Connection conn = dataSource.getConnection();
        @Cleanup Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
        stmt.setFetchSize(Integer.MIN_VALUE);

        long start = System.currentTimeMillis();
        // 查询sql，只查询关键的字段
        String sql = "SELECT id,price,partition_id,begin_time,end_time,effective_start_date,effective_end_date,last_generated_date FROM time_period_model_" + tableIndex + " where is_deleted = 0 and status = 0";
        @Cleanup ResultSet rs = stmt.executeQuery(sql);

        HashMap<Long, Integer> partitionIdAndAdvanceBookingDayMap = new HashMap<>();

        List<TimePeriodDO> timePeriodDOInsertBatch = new ArrayList<>();
        List<TimePeriodModelDO> timePeriodDOModelUpdateBatch = new ArrayList<>();
        int batchSize = 2000;
        SimpleDateFormat sdf = new SimpleDateFormat("HH:mm");

        // 每次获取一行数据进行处理，rs.next()如果有数据返回true，否则返回false
        while (rs.next()) {
            // 获取数据中的属性
            long id = rs.getLong("id");
            long partitionId = rs.getLong("partition_id");
            Date beginTime = sdf.parse(rs.getString("begin_time"));
            Date endTime = sdf.parse(rs.getString("end_time"));
            BigDecimal price = rs.getBigDecimal("price");
            Date effectiveStartDate = rs.getDate("effective_start_date");
            Date effectiveEndDate = rs.getDate("effective_end_date");
            // 上次生成到的日期
            Date lastGeneratedDate = rs.getDate("last_generated_date");
            int advanceBookingDay = getAdvanceBookingDayByPartitionId(partitionIdAndAdvanceBookingDayMap, partitionId);

            PartitionDO partitionDO = partitionService.getPartitionDOById(partitionId);
            if (partitionDO == null) {
                continue;
            }

            // 如果当前分区存在可预订分区的缓存，这里进行删除，因为生成了新的，需要重新查询数据库
            stringRedisTemplate.delete(String.format(
                    RedisCacheConstant.VENUE_TIME_PERIOD_BY_PARTITION_ID_KEY,
                    partitionId));

            // 这里其实不需要每天定时任务，都把advanceBookingDay都生成一遍，例如今天已经生成了未来七天的时间段了，那么明天其实只需要生成第八天的时间段即可
            Date generateDate = null;
            for (int i = 1; i <= advanceBookingDay; i++) {
                // 获取要生成的日期
                generateDate = new Date(System.currentTimeMillis() + i * 24 * 60 * 60 * 1000);
                if (lastGeneratedDate != null && generateDate.before(lastGeneratedDate)) {
                    // 如果对应日期的时间段已经被生成过了，直接跳过
                    continue;
                }
                // 检查明天的日期是否在这个范围内
                boolean isInDateRange = generateDate.after(effectiveStartDate) && generateDate.before(effectiveEndDate);
                if (isInDateRange) {
                    TimePeriodDO timePeriodDO = TimePeriodDO.builder()
                            .partitionId(partitionId)
                            .price(price)
                            .stock(partitionDO.getNum())
                            .bookedSlots(0L)
                            .periodDate(DateUtil.dateToLocalDate(generateDate))
                            .beginTime(DateUtil.dateToLocalTime(beginTime))
                            .endTime(DateUtil.dateToLocalTime(endTime))
                            .build();
                    timePeriodDO.setId(SnowflakeIdUtil.nextId());
                    timePeriodDOInsertBatch.add(timePeriodDO);
                    if (timePeriodDOInsertBatch.size() >= batchSize) {
                        // --if-- 数据量够了，存储数据库
                        timePeriodService.batchPublishTimePeriod(timePeriodDOInsertBatch);
                        timePeriodDOInsertBatch.clear();
                    }
                }
            }
            if (generateDate != null) {
                // 批量更新时间段模板的最新生成日期
                TimePeriodModelDO timePeriodModelDO = new TimePeriodModelDO();
                timePeriodModelDO.setId(id);
                timePeriodModelDO.setPartitionId(partitionId);
                timePeriodModelDO.setLastGeneratedDate(generateDate);
                timePeriodDOModelUpdateBatch.add(timePeriodModelDO);
                if (timePeriodDOModelUpdateBatch.size() >= batchSize) {
                    // --if-- 数据量够了，修改数据库
                    this.updateLastGeneratedDateBatch(timePeriodDOModelUpdateBatch);
                    timePeriodDOModelUpdateBatch.clear();
                }
            }
        }
        // 处理最后一波数据
        if (timePeriodDOInsertBatch.size() >= 0) {
            // 将时间段存储到数据库
            timePeriodService.batchPublishTimePeriod(timePeriodDOInsertBatch);
            timePeriodDOInsertBatch.clear();
        }
        if (timePeriodDOModelUpdateBatch.size() >= 0) {
            // --if-- 数据量够了，修改数据库
            this.updateLastGeneratedDateBatch(timePeriodDOModelUpdateBatch);
            timePeriodDOModelUpdateBatch.clear();
        }
        log.debug("流式生成时间段花费时间：" + ((System.currentTimeMillis() - start) / 1000));
    }

    private void updateLastGeneratedDateBatch(List<TimePeriodModelDO> timePeriodDOModelUpdateBatch) {
        if (timePeriodDOModelUpdateBatch == null || timePeriodDOModelUpdateBatch.size() == 0) {
            return;
        }
        baseMapper.updateLastGeneratedDateBatch(timePeriodDOModelUpdateBatch);
    }

    /**
     * 获取分区的提前预定时间
     * 使用二级缓存，本地缓存找不到，再去Redis中找，还找不到的话，去数据库中找
     *
     * @param partitionIdAndAdvanceBookingDayMap
     * @param partitionId
     * @return
     */
    private int getAdvanceBookingDayByPartitionId(HashMap<Long, Integer> partitionIdAndAdvanceBookingDayMap, long partitionId) {
        if (partitionIdAndAdvanceBookingDayMap.containsKey(partitionId)) {
            return partitionIdAndAdvanceBookingDayMap.get(partitionId);
        }
        VenueDO venueDO = venueService.getVenueDOByPartitionId(partitionId);
        partitionIdAndAdvanceBookingDayMap.put(partitionId, venueDO.getAdvanceBookingDay());
        return venueDO.getAdvanceBookingDay();
    }
}