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

🎯 本文详细介绍了场馆预定系统中时间段生成的实现方案。通过设计场馆表、时间段模板表和时间段表，系统能够根据场馆的提前预定天数生成未来可预定的时间段。为了确保任务执行的唯一性和高效性，系统采用分布式锁机制和定时任务，避免重复生成时间段。通过流式查询优化大数据处理，减少内存占用和网络延迟。同时，使用唯一复合索引保证时间段生成的幂等性，避免重复插入。为提高系统性能，引入二级缓存和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@SneakyThrowspublic 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，否则返回falsewhile (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();}
}