当前位置：首页 > news >正文

Redis打包事务,分批提交

news 2025/9/12 1:17:00

一、需求背景

接手一个老项目，在项目启动的时候，需要将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 {@AutowiredprivateAreaMapper areaMapper;private static boolean isStart = false;/*** 项目启动后，初始化字典到缓存*/@Overridepublic 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打包事务+分批提交的方式来提高启动速度,具体实现如下:

三、实现方法

实现方法

   @Autowiredpublic 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>() {@Overridepublic 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;}
}