Java 千万级数据去重120s 实现0错误（查询+去重+插入数据库）

话不多说上代码

Controller类

  @Operation(summary = "导入手机号")
    @PostMapping("/import")
    @PreAuthorize("@ss.hasPerm('sys:user:import')")
    @Log(value = "导入手机号", module = LogModuleEnum.PHONE)
    public Result<ExcelResult> importUsers(MultipartFile file) throws IOException {
        String username = SecurityUtils.getUsername();
        String key = StrUtil.format(RedisConstants.Repeat.REPEAT, username);
        // 分段处理数据
        if(redisTemplate.hasKey(key)){
            return Result.failed("数据已提交20秒之内之内点击1次！");
        }
        redisTemplate.opsForValue().set(key,username, 20, TimeUnit.SECONDS);
        String fileName = file.getOriginalFilename();
        int index = fileName.indexOf(".");
        String name = fileName.substring(0,index);
        CompletableFuture.runAsync(() -> {
            //(1)布隆
//            uploadExcel.upLoad(file,name);
            // (2)
            uploadListSet.upLoad(file,name);
        });
        ExcelResult r =new ExcelResult();
        r.setCode("00000");
        return Result.success(r);
    }

工具类 UploadListSet

package com.youlai.boot.system.utils;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import com.youlai.boot.common.util.ExcelUtilsA;
import com.youlai.boot.system.mapper.PhoneMapper;
import com.youlai.boot.system.mapper.RepeatphoneMapper;
import com.youlai.boot.system.model.dto.PhoneImportDTO;
import com.youlai.boot.system.model.entity.Phone;

import jakarta.annotation.Resource;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;

import java.nio.charset.Charset;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@Component
public class UploadListSet {
    @Resource
    private PhoneMapper phoneMapper;
    @Resource
    private JdbcTemplate jdbcTemplate;
    @Resource
    private RepeatphoneMapper repeatphoneMapper;

    // 配置参数
    private static final int BATCH_SIZE = 30000;
    private static final int THREAD_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;
    // 使用线程安全的计数器
    private static  AtomicInteger totalProcessed = new AtomicInteger(0);
    private static  AtomicInteger duplicateCount = new AtomicInteger(0);
    private static  AtomicInteger  duplicateCountlIST = new AtomicInteger(0);

    public void upLoad(MultipartFile file, String name) {
//        duplicateCountlIST =new AtomicInteger(0);
//        duplicateCount = new AtomicInteger(0);
//        totalProcessed = new AtomicInteger(0);
        long startTime = System.currentTimeMillis();
        try {
            importPhoneNumbers(file,name);
        } catch (Exception e) {
            e.printStackTrace();
        }
        long endTime = System.currentTimeMillis();
        System.out.println("导入完成，耗时: " + (endTime - startTime) / 1000 + "秒");
//        System.out.println("处理总数: " + totalProcessed.get() + ", 数据重复数: " + duplicateCount.get() + " SSSSS "+ duplicateCountlIST);
    }

    public void importPhoneNumbers(MultipartFile file,String name) throws Exception {

        // 创建线程池
        ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
        // 使用CompletionService管理任务结果
        CompletionService<List<String>> completionService =
                new ExecutorCompletionService<>(executor);
        List<PhoneImportDTO> phoneNumbers =  ExcelUtilsA.read(file,PhoneImportDTO.class);
        if(!phoneNumbers.isEmpty()){
            try {
                List<List<String>>  listAll = geng(0);
                List<String> listUpload =phoneNumbers.stream().map(PhoneImportDTO::getId).toList();
                int totalRows = listUpload.size();
                System.out.println("导入数据条数：" + totalRows);
                BloomFilter<String> yuanBloomFilter = BloomFilter.create(
                        Funnels.stringFunnel(Charset.defaultCharset()),
                        700_000_0,
                        0.0000001);
                // 分片处理Excel，每个任务处理一部分行
                for (int start = 0; start < totalRows; start += BATCH_SIZE) {
                    int end = Math.min(start + BATCH_SIZE, totalRows);
                    completionService.submit(new UploadExcel.ExcelProcessor(listUpload, start, end));
                }
                for (int i = 0; i < (totalRows + BATCH_SIZE - 1) / BATCH_SIZE; i++) {
                    List<Phone> listCf =new ArrayList<>();
                    List<String> batch = completionService.take().get();
                    // 格式化当前cell对应数据为String
                    List<Phone> uniqueNumbers = new ArrayList<>();
                    //(3) 集合List<set>
                    findDuplicates(batch, listAll,name,uniqueNumbers,listCf,yuanBloomFilter);
                    // 批量插入数据库
                    if (!uniqueNumbers.isEmpty()) {
                        asynchronous(uniqueNumbers);
                        System.out.println("第次插入新新新： "+ uniqueNumbers.size());
                    }
                    if(!listCf.isEmpty()){
                        asynchronousRepeat(listCf);
                        System.out.println("第次插入重复： "+ listCf.size());
                    }
                }
            } finally {
                executor.shutdown();
                executor.awaitTermination(1, TimeUnit.HOURS);
            }
        }
        // 关闭线程池
        executor.shutdown();
    }




    // 异步插入
    private void asynchronous(List<Phone> uniqueNumbers){
        CompletableFuture.runAsync(() -> {
            batchInsertToDatabase(uniqueNumbers);
        });
    }
    // 异步插入重复数据
    private void asynchronousRepeat(List<Phone> uniqueNumbers){
//        CompletableFuture.runAsync(() -> {
            batchInsertToDatabaseRepeat(uniqueNumbers);
//        });
    }
    /**
     * JDBC 批量插入或JPA批量处理
     * @param phoneNumbers
     */
    private void batchInsertToDatabase(List<Phone> phoneNumbers) {
        // 使用JDBC批量插入或JPA批量处理
        // 这里简化为伪代码
        if(!phoneNumbers.isEmpty()){
            phoneMapper.saveAll(phoneNumbers);
        }
    }
    /**
     * JDBC 批量插入重复手机号或JPA批量处理
     * @param phoneNumbers
     */
    @Transactional(rollbackFor = Exception.class)
    private void batchInsertToDatabaseRepeat(List<Phone> phoneNumbers) {
        // 使用JDBC批量插入或JPA批量处理
        // 这里简化为伪代码
        if(!phoneNumbers.isEmpty()){
            repeatphoneMapper.saveAll(phoneNumbers);
        }
    }

    private static boolean b = false;

    public static void findDuplicates(List<String> listUpload, List<List<String>> listAll,String name,List<Phone> uniqueNumbers,List<Phone> listCf,BloomFilter<String> yuanBloomFilter) {
        LocalDateTime time = LocalDateTime.now();
       if(!listUpload.isEmpty()){
           for (String upPhone:listUpload){
               Phone s = new Phone();
               s.setCreateTime(time);
               s.setId(upPhone);
               s.setName(name);
               if(!yuanBloomFilter.mightContain(upPhone)){
                   yuanBloomFilter.put(upPhone);
                   //判断重复
                   if(!listAll.isEmpty()){
                      b = false;
                       listAll.stream().map(m->{
                           if(!b){
                               for(String ll : m){
                                   if(yuanBloomFilter.mightContain(ll)){
                                       listCf.add(s);
                                       b = true;
                                       break;
                                   }
                               }
                           } return m;
                       }).toList();
                       if(!b) {
                           // 有效数据
                           uniqueNumbers.add(s);
                       }
                   }else {
                       uniqueNumbers.add(s);
                   }
               }else {
                   listCf.add(s);
               }
           }
       }
    }


    // 更新策略
    private List<List<String>> geng(long yuanBloomFilterCount){

        long startTime = System.currentTimeMillis();
        List<List<String>> fenduan = new ArrayList<>();
        long countAll = phoneMapper.countAll();
        if(countAll==0){
            return fenduan;
        }
        if(yuanBloomFilterCount<countAll){
            int countLimit = 1000;
            if(checkOddEven((int)countAll)){
                countLimit = (int)countAll/20+20;
            }else {
                countLimit =(int)countAll/20+20;
            }
            if (countAll<500000){
                fenduan.add(phoneMapper.phoneIdList());
            }else {
                final int[] offset = {0};
                int limit = countLimit;

                List<Integer> ints = new ArrayList<>();
                for(int i =0; i<21; i++){
                    ints.add(i);
                }
                List<CompletableFuture<Void>> futures = ints.stream()
                        .map(data -> CompletableFuture.runAsync(() -> {
                            List<String> list = jdbcTemplate.queryForList("SELECT id FROM phone LIMIT " + limit + " OFFSET " + offset[0], String.class);
                            offset[0] = offset[0] +limit;
                            fenduan.add(list);
                        }))
                        .collect(Collectors.toList());
                CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); // 等待所有任务完成
            }
        }
        long entTime = System.currentTimeMillis();
        System.out.println("读取原数据耗时: " + (entTime - startTime) + " 毫秒 " + countAll+ " 条");
        return fenduan;
    }

    // 取模5数判断
    public boolean checkOddEven(int number) {
        if (number % 2 == 0) {
            return true;
        }
        return false;
    }

    // 造数据
    public void zao(){
        LocalDateTime l =  LocalDateTime.now();
        Long yua =  18645550314L;
        for(int i=0; i<100; i++){
            List<Phone> phoneNumbers = new ArrayList<>();
            for(int j = 0; j<50000; j++){
                yua++;
                Phone p = new Phone();
                p.setId(yua.toString());
                p.setName("测试");
                p.setCreateTime(l);
                phoneNumbers.add(p);
            }
//            batchInsertToDatabase(phoneNumbers);
        }
    }


    static class ExcelProcessor implements Callable<List<String>> {
        private final List<String> phoneNumbers;
        private final int startRow;
        private final int endRow;
        public ExcelProcessor(List<String> phoneNumbers, int startRow, int endRow) {
            this.phoneNumbers = phoneNumbers;
            this.startRow = startRow;
            this.endRow = endRow;
        }
        @Override
        public List<String> call() {
            List<String> numbers = new ArrayList<>();
            for (int i = startRow; i < endRow; i++) {
                String p = phoneNumbers.get(i);
                if(Objects.nonNull(p)){
                    numbers.add(p);
                }
            }
            return numbers;
        }
    }

}

数据库表

CREATE TABLE `phone` (
  `id` varchar(24) COLLATE utf8mb4_unicode_ci NOT NULL,
  `name` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL,
  `create_time` datetime DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;