java+sql实现mysql数据表的数据结转-CFANZ编程社区

数据表：emax_timer_request，存储的数据是mq数据消费的流水。

历史数据并没有什么用，所以，随着数据量的增大，实现历史数据的自动结转。

java+sql实现mysql数据表的数据结转_数据

程序的ORM是mybatisplus。

实现方案是基于新增数据的save方法。

TimerRequestManager.java----------->TimerRequestMapper.java---------------->TimerRequestMapper.xml

TimerRequestManager.java

这段代码解决的问题是：何时结转？

把数据记录数放到redis里，每次有新增记录时调用incr命令实现递增计数。当这个计数值超过20w时，就是说，当表里的数据达到20w条时，开始结转数据。
为了不影响插入性能，结转的程序放在异步线程里执行。
考虑到并发插入的场景，结转程序要防并发。故借助redis分布式说实现。

1 @Slf4j
 2 @Service
 3 public class TimerRequestManager extends ServiceImpl<TimerRequestMapper, TimerRequest> implements IService<TimerRequest> {
 4     @Autowired
 5     private RedisUtil redisUtil;
 6     @Autowired
 7     private DistributedLock distributedLock;
 8 
 9     private static final String COUNT_EMAX_TIMER_REQUEST = "count.emax_timer_request";
10 
11     @Override
12     public boolean save(TimerRequest entity) {
13         boolean save = super.save(entity);
14 
15         long incr = redisUtil.incr(COUNT_EMAX_TIMER_REQUEST, 1);
16         if (incr == 1) {
17             incr = baseMapper.selectCount(Wrappers.query());
18             redisUtil.set(COUNT_EMAX_TIMER_REQUEST, incr);
19         }
20         //  大于20w条自动结转数据
21         if (incr >= 200000) {
22             //重置初始的count值
23             redisUtil.set(COUNT_EMAX_TIMER_REQUEST, 5);
24             // 异步结转数据
25             boolean locked = distributedLock.lock("dataMig.emax_timer_request", 60 * 1000L);
26             if (locked) {
27                 ThreadPoolUtil.getThreadPoolExecutor().execute(() -> {
28                     try {
29                         int i = baseMapper.dataMigration(DateUtils.formatDate(new Date(), "yyyyMMddHHmm"));
30                         if (i > 0) {
31                             //重置count值
32                             Integer newCount = baseMapper.selectCount(Wrappers.query());
33                             redisUtil.set(COUNT_EMAX_TIMER_REQUEST, newCount);
34                             log.info("emax_timer_request数据结转完成");
35                         }
36                     } catch (Exception e) {
37                         log.error("emax_timer_request数据结转异常,", e);
38                     }
39                 });
40             }
41         }
42         return save;
43     }
44

TimerRequestMapper.java

TimerRequestMapper.java是一个interface。定义数据结转的操作方法。

1 public interface TimerRequestMapper extends BaseMapper<TimerRequest> {
2     int dataMigration(@Param("tableSuffix") String tableSuffix);
3

TimerRequestMapper.xml

如下xml文件里“dataMigration”这个update statement实现了数据结转的策略：根据评估业务场景，只保留最新的1w条记录，其他历史数据保存到新的结转表里。

需要注意的是，对于mybatisplus（mybatis）来说，如果要在statement里执行多条sql语句，需要在jdbc数据库连接串加上allowMultiQueries=true，如：

jdbc:mysql://192.168.40.84:3306/emax_base?characterEncoding=UTF-8&useUnicode=true&useSSL=false&serverTimezone=Asia/Shanghai&allowMultiQueries=true

<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.emax.zhenghe.rpcapi.provider.modules.mq.mapper.TimerRequestMapper">

    <update id="dataMigration">
        <![CDATA[
        SELECT MIN(id) INTO @myId FROM( SELECT id FROM emax_timer_request ORDER BY id DESC LIMIT 10000) a;
        CREATE TABLE emax_timer_request${tableSuffix} LIKE emax_timer_request;
        INSERT INTO emax_timer_request${tableSuffix} SELECT * FROM emax_timer_request WHERE id<=@myId;
        DELETE FROM emax_timer_request WHERE id<=@myId;
        ]]>
    </update>
</mapper>

以上，并发多线程测试ok。

结转20w条数据平均用时=10s。当然，这个会因实际数据和服务器计算能力而有所不同。

java+sql实现mysql数据表的数据结转_数据结转_02