Redis打包事务,分批提交

一、需求背景

接手一个老项目，在项目启动的时候，需要将xxx省整个省的所有区域数据数据、以及系统字典配置逐条保存在Redis缓存里面，这样查询的时候会更快;
区域数据+字典数据一共大概20000多条,，前同事直接使用 list.forEach()逐条写入Redis,如下:

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
/**
 * @author xxx
 * @version 1.0
 * @date 2022/7/21 15:29
 * @Description: 项目启动成功后初始化区域数据到redis
 */
@Component
@Slf4j
public class AreasInitialComponent implements ApplicationRunner {

    @Autowired
    privateAreaMapper areaMapper;
    private static boolean isStart = false;
    
    /**
     * 项目启动后，初始化字典到缓存
     */
    @Override
    public void run(ApplicationArguments args) throws Exception {
        if (isStart) {
            return;
        }
        try {
            log.info("Start*******************项目启动后，初始化字典到缓存*******************");
            QueryWrapper<Area> wrapper = new QueryWrapper<>();
            wrapper.eq("del", "0");
            List<Area> areas = areaMapper.selectList(wrapper);
            if (!CollectionUtils.isEmpty(areas )) {
                RedisCache redisCache = SpringUtils.getBean(RedisCache.class);
                //先将区域集合整体做个缓存
                log.info("*******************先将区域集合整体做个缓存*******************");
                AreaUtil.setAreaListCache(redisCache, areas);
                //再将每一条区域进行缓存
                areas.stream().forEach(a -> {
                    AreaUtil.setAreaCache(redisCache, a.getId(), a);
                });
            }
            isStart = true;
            log.info("End*******************项目启动后，初始化字典到缓存*******************");
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }
}

导致项目启动速度巨慢,再加上需要使用代理软件才能连接公司的数据库,每次启动项目都需要10几分钟,当真是苦不堪言;由于受不了这样的启动速度,因此决定自己动手优化。

二、解决思路

联想到MySQL的事务打包方式，于是自己动手尝试通过Redis打包事务+分批提交的方式来提高启动速度,具体实现如下:

三、实现方法

实现方法

   @Autowired
    public RedisTemplate redisTemplate;  
 /**
     * 逐条设置区域缓存
     *
     * @param areas
     * @throws InterruptedException
     */
    public void setAreaCacheItemByItem(List<Area> areas) throws InterruptedException {
        MoreThreadCallBack<Area> callBack = new MoreThreadCallBack<>();
        callBack.setThreadCount(10);
        callBack.setLimitCount(50);
        callBack.setTitle("设置区域缓存批量任务");
        callBack.setAllList(areas);
        callBack.call((list, threadNum) -> {
            //使用自定义线程回调工具分摊任务
            redisTemplate.execute(new SessionCallback<Object>() {
                @Override
                public Object execute(RedisOperations operations) throws DataAccessException {
                    //开启redis事务
                    operations.multi();
                    list.forEach(item -> {
                        operations.opsForValue().set(item.getId(), item);
                    });
                    // 提交事务
                    operations.exec();
                    return null;
                }
            });
        });
    }

线程回调工具MoreThreadCallBack()

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.compress.utils.Lists;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

@Data
@Slf4j
public class MoreThreadCallBack<P> {

    public int limitCount = 1000;
    private int threadCount = 10;
    private List<P> allList;
    private AtomicInteger errorCheck;
    private String title;

    public interface CallBack<P> {
        void call(List<P> list, Integer threadNum);
    }

    public boolean call(CallBack<P> callBack) throws InterruptedException, RuntimeException {
        if (allList.isEmpty()) {
            return false;
        }

        // 线程池
        ExecutorService exec = Executors.newCachedThreadPool();

        // 根据大小判断线程数量
        if (allList.size() <= limitCount) {
            threadCount = 1;
        }
        // 等待结果类
        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);

        // 分摊多份
        List<List<P>> llist = Lists.newArrayList();
        for (int i = 0; i < threadCount; i++) {
            llist.add(Lists.newArrayList());
        }
        int index = 0;
        for (P p : allList) {
            llist.get(index).add(p);
            index = index == (threadCount - 1) ? 0 : index + 1;
        }

        // 异常记录
        errorCheck = new AtomicInteger(0);

        // 执行
        for (int i = 0; i < llist.size(); i++) {
            List<P> list = llist.get(i);
            final Integer threadNum = i;
            exec.execute(() -> {
                long startTime = System.currentTimeMillis();
                //抛出异常 自身不处理
                log.info("标题:{}-{}号线程开始回调执行 数量:{}", this.getTitle(), threadNum, list.size());
                callBack.call(list, threadNum);
                long endTime = System.currentTimeMillis();
                log.info("标题:{}-{}号线程回调执行完毕 耗时:{}", this.getTitle(), threadNum, +(endTime - startTime));
                countDownLatch.countDown();
            });
        }
        // 等待处理完毕
        countDownLatch.await();
        // 关闭线程池
        exec.shutdown();
        return errorCheck.get() <= 0;
    }
    public boolean next() {
        // 检测是否有线程提前结束
        if (errorCheck.get() > 0) {
            return false;
        }
        return true;
    }
    public void error() {
        errorCheck.incrementAndGet();
    }

    public String getTitle() {
        return title == null ? "" : title;
    }
}