定时器循环分页处理大数据

2023-03-20
在定时器触发下，读取数据库大数据，处理之后，再写回数据库。这种，一次性取全部的数据下来处理不太现实，一般需要分页处理，一批一批的处理。
此文讲述基本的处理方法结构
处理类code

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.github.pagehelper.PageHelper;
import com.yqxc.aiface.biz.orderWorks.model.entity.MergeEmojPictureEntity;
import com.yqxc.aiface.biz.orderWorks.service.MergeEmojPictureService;
import com.yqxc.aiface.common.oss.helper.OssHelper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;

/**
 * @author 曳戈泰尔
 * @version 1.0
 * @description 定时任务 执行内容定义
 * @date 2023/3/20 11:32 AM
 */
@Component
@Slf4j
public class ScheduleTask {

    /** 注入 合成表情包 service */
    private MergeEmojPictureService mergeEmojPictureService;

    @Autowired
    public void setMergeEmojPictureService(MergeEmojPictureService mergeEmojPictureService) {
        this.mergeEmojPictureService = mergeEmojPictureService;
    }

    /** 注入对象存储OSS工具类 */
    private OssHelper ossHelper;

    @Autowired
    public void setOssHelper(OssHelper ossHelper) {
        this.ossHelper = ossHelper;
    }

    /** 注入 线程池 */
    private Executor threadPoolExecutor;

    @Autowired
    public void setThreadPoolExecutor(Executor threadPoolExecutor) {
        this.threadPoolExecutor = threadPoolExecutor;
    }

    /** 定义分页 开启多线程处理 设置合适的pageSize值 */
    public static Integer pageNum = 1;

    public static Integer pageSize = 10;

    public void execute() {

        // 处理合成表情包删除
        executeDeleteUserEmojMergeImage();
    }

    /** 处理合成表情包删除 */
    public void executeDeleteUserEmojMergeImage() {

        // 总数据
        LambdaQueryWrapper<MergeEmojPictureEntity> lqwMerge = new LambdaQueryWrapper<>();
        lqwMerge.eq(MergeEmojPictureEntity::getDel, 0);
        lqwMerge.eq(MergeEmojPictureEntity::getDeleteUrl, 0);
        Integer total = mergeEmojPictureService.count(lqwMerge);

        // 执行次数
        int count = total / pageSize + 1;

        for (int i = 0; i < count; i++) {

            log.info("定时处理用户合成作品图片，正在处理第 {} 页", i);

            deleteUserEmojMergeImageBatchDealData(pageNum, pageSize);

            // 处理后的数据影响下一次查询数据，如处理之后的数据会排除在下一次查询数据之外，页码一直1即可，一直获取第一页数据 处理即可；注释pageNum += 1;
            // 处理后的数据不影响下一次查询数据，页码需要递增跟着走，才能查询出所有的数据 进行处理；不注释pageNum += 1;
            // pageNum += 1;
        }
    }

    public void deleteUserEmojMergeImageDealData(Integer pageNum, Integer pageSize) {

        // 分页查询所有数据
        LambdaQueryWrapper<MergeEmojPictureEntity> lqwMerge = new LambdaQueryWrapper<>();
        lqwMerge.select(MergeEmojPictureEntity::getId, MergeEmojPictureEntity::getUrl);
        lqwMerge.eq(MergeEmojPictureEntity::getDel, 0);
        lqwMerge.eq(MergeEmojPictureEntity::getDeleteUrl, 0);

        PageHelper.startPage(pageNum, pageSize);
        List<MergeEmojPictureEntity> mergeEmojPictureEntityList =
                mergeEmojPictureService.list(lqwMerge);

        if (mergeEmojPictureEntityList.isEmpty()) {
            return;
        }       
      
        // 线程池处理数据
        CountDownLatch countDownLatch = new CountDownLatch(mergeEmojPictureEntityList.size());
        // 记录更新数据
        mergeEmojPictureEntityList.forEach(
                item -> {
                    threadPoolExecutor.execute(
                            new Runnable() {
                                @Override
                                public void run() {

                                    Boolean deleteResult = ossHelper.deleteFile(item.getUrl());
                                    if (deleteResult) {
                                        item.setDeleteUrl(1);
                                        item.setDel(1);
                                        item.setUpdateTime(LocalDateTime.now());
                                        if (!mergeEmojPictureService.updateById(item)) {
                                            log.error(
                                                    "已删除OSS文件，更新数据库失败，merge_emoj_picture ID：{}",
                                                    item.getId());
                                        }
                                    } else {
                                        log.error(
                                                "删除用户合成表情包的OSS文件失败，merge_emoj_picture ID：{}，OSS链接：{}",
                                                item.getId(),
                                                item.getUrl());
                                    }

                                    countDownLatch.countDown();
                                }
                            });
                });

        try {
            countDownLatch.await();
        } catch (Exception ex) {
            log.error("定时处理用户合成作品图片，处理第 {} 页失败，原因：{}", pageNum, ex.getMessage());
        }
    }

    public void deleteUserEmojMergeImageBatchDealData(Integer pageNum, Integer pageSize) {

        // 分页查询所有数据
        LambdaQueryWrapper<MergeEmojPictureEntity> lqwMerge = new LambdaQueryWrapper<>();
        lqwMerge.select(MergeEmojPictureEntity::getId, MergeEmojPictureEntity::getUrl);
        lqwMerge.eq(MergeEmojPictureEntity::getDel, 0);
        lqwMerge.eq(MergeEmojPictureEntity::getDeleteUrl, 0);

        PageHelper.startPage(pageNum, pageSize);
        List<MergeEmojPictureEntity> mergeEmojPictureEntityList =
                mergeEmojPictureService.list(lqwMerge);

        if (mergeEmojPictureEntityList.isEmpty()) {
            return;
        }     
      
        // 线程池处理数据
        CountDownLatch countDownLatch = new CountDownLatch(mergeEmojPictureEntityList.size());
        mergeEmojPictureEntityList.forEach(
                item -> {
                    threadPoolExecutor.execute(
                            new Runnable() {
                                @Override
                                public void run() {

                                    Boolean deleteResult = ossHelper.deleteFile(item.getUrl());
                                    if (deleteResult) {
                                        item.setDeleteUrl(1);
                                        item.setDel(1);
                                        item.setUpdateTime(LocalDateTime.now());
                                    } else {
                                        log.error(
                                                "删除用户合成表情包的OSS文件失败，merge_emoj_picture ID：{}，OSS链接：{}",
                                                item.getId(),
                                                item.getUrl());
                                    }

                                    countDownLatch.countDown();
                                }
                            });
                });

        try {
            countDownLatch.await();
        } catch (Exception ex) {
            log.error("定时处理用户合成作品图片，处理第 {} 页失败，原因：{}", pageNum, ex.getMessage());
        }

        if (!mergeEmojPictureService.updateBatchById(mergeEmojPictureEntityList)) {
            log.error(
                    "已删除OSS文件，批量更新数据库失败，merge_emoj_picture ID-1：{}",
                    mergeEmojPictureEntityList.get(0).getId());
        }
    }
}