Java批量导入查重案例怎么开发

wen java案例 2026-06-28 233

本文目录导读：

Java批量导入查重案例怎么开发

业务场景
完整实现案例
注意事项

我来介绍一个完整的Java批量导入查重案例，包含数据准备、查重逻辑和实现代码。

业务场景

假设我们需要批量导入用户数据,要求：

支持Excel/CSV文件导入
对导入数据进行查重（数据库已存在 + 导入文件内部重复）
返回重复数据明细和导入结果

完整实现案例

1 基础实体类

// 用户实体
@Data
public class User {
    private Long id;
    private String username;  // 用户名（唯一）
    private String email;     // 邮箱（唯一）
    private String phone;     // 手机号（唯一）
    private String name;
    private Integer status;
    private Date createTime;
}
// 导入结果
@Data
public class ImportResult<T> {
    private boolean success;
    private int totalCount;           // 总记录数
    private int successCount;         // 成功导入数
    private int failCount;            // 失败数
    private List<T> successData;      // 成功数据
    private List<DuplicateRecord<T>> duplicateRecords;  // 重复记录
    @Data
    public static class DuplicateRecord<T> {
        private int rowIndex;         // Excel行号
        private T data;               // 原始数据
        private String duplicateType; // 重复类型：DB_EXISTS / FILE_DUPLICATE
        private String duplicateField; // 重复字段
        private String duplicateValue; // 重复值
        private String message;       // 提示信息
    }
}

2 查重服务层

@Service
@Slf4j
public class UserImportService {
    @Autowired
    private UserMapper userMapper;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    /**
     * 批量导入用户（带查重）
     * @param userList 待导入的用户列表
     * @param batchId 批次号
     * @return 导入结果
     */
    @Transactional(rollbackFor = Exception.class)
    public ImportResult<User> batchImport(List<User> userList, String batchId) {
        ImportResult<User> result = new ImportResult<>();
        result.setTotalCount(userList.size());
        if (CollectionUtils.isEmpty(userList)) {
            result.setSuccess(true);
            return result;
        }
        List<User> validUsers = new ArrayList<>();
        List<ImportResult.DuplicateRecord<User>> duplicateRecords = new ArrayList<>();
        try {
            // 第一步：文件内部查重
            Map<String, List<Integer>> fileDuplicateMap = checkFileDuplicate(userList);
            // 第二步：数据库查重
            Map<String, User> dbUserMap = checkDatabaseDuplicate(userList);
            // 第三步：处理每一条数据
            for (int i = 0; i < userList.size(); i++) {
                User user = userList.get(i);
                int rowIndex = i + 2; // Excel行号（从第2行开始，第1行是表头）
                // 检查文件内部重复
                boolean isFileDuplicate = false;
                if (fileDuplicateMap.containsKey(user.getUsername())) {
                    List<Integer> duplicateIndexes = fileDuplicateMap.get(user.getUsername());
                    if (duplicateIndexes.size() > 1 && duplicateIndexes.get(0) == i) {
                        // 第一条记录保留，其余标记为重复
                        isFileDuplicate = i != duplicateIndexes.get(0);
                    }
                }
                if (isFileDuplicate) {
                    ImportResult.DuplicateRecord<User> record = new ImportResult.DuplicateRecord<>();
                    record.setRowIndex(rowIndex);
                    record.setData(user);
                    record.setDuplicateType("FILE_DUPLICATE");
                    record.setDuplicateField("username");
                    record.setDuplicateValue(user.getUsername());
                    record.setMessage("文件内存在重复的用户名");
                    duplicateRecords.add(record);
                    continue;
                }
                // 检查数据库重复
                String duplicateField = null;
                Object duplicateValue = null;
                if (dbUserMap.containsKey(user.getUsername())) {
                    duplicateField = "username";
                    duplicateValue = user.getUsername();
                } else if (dbUserMap.containsKey(user.getEmail())) {
                    duplicateField = "email";
                    duplicateValue = user.getEmail();
                } else if (dbUserMap.containsKey(user.getPhone())) {
                    duplicateField = "phone";
                    duplicateValue = user.getPhone();
                }
                if (duplicateField != null) {
                    ImportResult.DuplicateRecord<User> record = new ImportResult.DuplicateRecord<>();
                    record.setRowIndex(rowIndex);
                    record.setData(user);
                    record.setDuplicateType("DB_EXISTS");
                    record.setDuplicateField(duplicateField);
                    record.setDuplicateValue(String.valueOf(duplicateValue));
                    record.setMessage("数据库中已存在相同的" + duplicateField);
                    duplicateRecords.add(record);
                    continue;
                }
                // 验证通过，添加到有效列表
                user.setCreateTime(new Date());
                user.setStatus(1);
                validUsers.add(user);
            }
            // 第四步：批量插入有效数据
            if (!CollectionUtils.isEmpty(validUsers)) {
                // 分批插入，每批500条
                List<List<User>> batches = Lists.partition(validUsers, 500);
                for (List<User> batch : batches) {
                    userMapper.batchInsert(batch);
                }
                // 更新缓存
                updateUserCache(validUsers);
            }
            // 设置结果
            result.setSuccess(true);
            result.setSuccessCount(validUsers.size());
            result.setFailCount(duplicateRecords.size());
            result.setSuccessData(validUsers);
            result.setDuplicateRecords(duplicateRecords);
            log.info("批量导入完成: batchId={}, total={}, success={}, fail={}", 
                    batchId, userList.size(), validUsers.size(), duplicateRecords.size());
        } catch (Exception e) {
            log.error("批量导入失败: batchId={}", batchId, e);
            result.setSuccess(false);
            result.setFailCount(userList.size());
            throw new RuntimeException("导入失败：" + e.getMessage());
        }
        return result;
    }
    /**
     * 文件内部查重
     */
    private Map<String, List<Integer>> checkFileDuplicate(List<User> userList) {
        Map<String, List<Integer>> duplicateMap = new HashMap<>();
        for (int i = 0; i < userList.size(); i++) {
            User user = userList.get(i);
            if (StringUtils.isNotBlank(user.getUsername())) {
                duplicateMap.computeIfAbsent(user.getUsername(), k -> new ArrayList<>()).add(i);
            }
        }
        // 只保留有重复的项
        duplicateMap.entrySet().removeIf(entry -> entry.getValue().size() <= 1);
        return duplicateMap;
    }
    /**
     * 数据库查重（使用缓存优化）
     */
    private Map<String, User> checkDatabaseDuplicate(List<User> userList) {
        Map<String, User> duplicateMap = new HashMap<>();
        // 收集所有需要检查的字段
        Set<String> usernames = userList.stream()
                .map(User::getUsername)
                .filter(StringUtils::isNotBlank)
                .collect(Collectors.toSet());
        Set<String> emails = userList.stream()
                .map(User::getEmail)
                .filter(StringUtils::isNotBlank)
                .collect(Collectors.toSet());
        Set<String> phones = userList.stream()
                .map(User::getPhone)
                .filter(StringUtils::isNotBlank)
                .collect(Collectors.toSet());
        // 先从缓存查询
        Map<String, User> cacheResults = queryFromCache(usernames, emails, phones);
        duplicateMap.putAll(cacheResults);
        // 缓存未命中的从数据库查询
        Set<String> notCachedUsernames = new HashSet<>(usernames);
        Set<String> notCachedEmails = new HashSet<>(emails);
        Set<String> notCachedPhones = new HashSet<>(phones);
        // 从缓存结果中移除已存在的
        cacheResults.values().forEach(user -> {
            notCachedUsernames.remove(user.getUsername());
            notCachedEmails.remove(user.getEmail());
            notCachedPhones.remove(user.getPhone());
        });
        // 批量查询数据库
        if (!notCachedUsernames.isEmpty() || !notCachedEmails.isEmpty() || !notCachedPhones.isEmpty()) {
            List<User> dbUsers = userMapper.selectByFields(
                    new ArrayList<>(notCachedUsernames),
                    new ArrayList<>(notCachedEmails),
                    new ArrayList<>(notCachedPhones)
            );
            dbUsers.forEach(user -> {
                duplicateMap.put(user.getUsername(), user);
                if (StringUtils.isNotBlank(user.getEmail())) {
                    duplicateMap.put(user.getEmail(), user);
                }
                if (StringUtils.isNotBlank(user.getPhone())) {
                    duplicateMap.put(user.getPhone(), user);
                }
            });
            // 更新缓存
            updateUserCache(dbUsers);
        }
        return duplicateMap;
    }
    /**
     * 从缓存查询用户
     */
    private Map<String, User> queryFromCache(Set<String> usernames, Set<String> emails, Set<String> phones) {
        Map<String, User> result = new HashMap<>();
        // 批量查询缓存
        Set<String> cacheKeys = new HashSet<>();
        usernames.forEach(u -> cacheKeys.add("user:username:" + u));
        emails.forEach(e -> cacheKeys.add("user:email:" + e));
        phones.forEach(p -> cacheKeys.add("user:phone:" + p));
        List<Object> cacheValues = redisTemplate.opsForValue().multiGet(cacheKeys);
        if (cacheValues != null) {
            for (int i = 0; i < cacheValues.size(); i++) {
                if (cacheValues.get(i) != null) {
                    User user = (User) cacheValues.get(i);
                    result.put(user.getUsername(), user);
                }
            }
        }
        return result;
    }
    /**
     * 更新用户缓存
     */
    private void updateUserCache(List<User> users) {
        users.forEach(user -> {
            // 缓存用户名映射
            redisTemplate.opsForValue().set(
                    "user:username:" + user.getUsername(),
                    user,
                    1, TimeUnit.DAYS
            );
            // 缓存邮箱映射
            if (StringUtils.isNotBlank(user.getEmail())) {
                redisTemplate.opsForValue().set(
                        "user:email:" + user.getEmail(),
                        user,
                        1, TimeUnit.DAYS
                );
            }
            // 缓存手机号映射
            if (StringUtils.isNotBlank(user.getPhone())) {
                redisTemplate.opsForValue().set(
                        "user:phone:" + user.getPhone(),
                        user,
                        1, TimeUnit.DAYS
                );
            }
        });
    }
}

3 控制器层

@RestController
@RequestMapping("/api/import")
@Slf4j
public class UserImportController {
    @Autowired
    private UserImportService userImportService;
    @PostMapping("/users")
    public ResponseEntity<ImportResult<User>> importUsers(
            @RequestParam("file") MultipartFile file) {
        try {
            // 1. 解析文件
            List<User> userList = parseExcelFile(file);
            // 2. 生成批次号
            String batchId = UUID.randomUUID().toString().replace("-", "");
            // 3. 执行导入
            ImportResult<User> result = userImportService.batchImport(userList, batchId);
            // 4. 返回结果
            return ResponseEntity.ok(result);
        } catch (Exception e) {
            log.error("文件导入失败", e);
            ImportResult<User> errorResult = new ImportResult<>();
            errorResult.setSuccess(false);
            errorResult.setFailCount(0);
            errorResult.setMessage("导入失败：" + e.getMessage());
            return ResponseEntity.badRequest().body(errorResult);
        }
    }
    /**
     * 解析Excel文件
     */
    private List<User> parseExcelFile(MultipartFile file) throws IOException {
        List<User> userList = new ArrayList<>();
        // 使用EasyExcel解析
        EasyExcel.read(file.getInputStream())
                .head(User.class)
                .sheet()
                .doReadSync();
        // 实际项目中可以使用EasyExcel的监听器逐行读取
        // 这里简化处理
        return userList;
    }
}

4 Mapper层

@Mapper
public interface UserMapper {
    // 批量插入
    @InsertProvider(type = UserSqlProvider.class, method = "batchInsert")
    int batchInsert(@Param("list") List<User> list);
    // 批量查询用户
    @SelectProvider(type = UserSqlProvider.class, method = "selectByFields")
    List<User> selectByFields(
            @Param("usernames") List<String> usernames,
            @Param("emails") List<String> emails,
            @Param("phones") List<String> phones
    );
    // 根据用户名查询
    @Select("SELECT * FROM user WHERE username = #{username} LIMIT 1")
    User findByUsername(@Param("username") String username);
}
// SQL提供类
public class UserSqlProvider {
    public String batchInsert(Map<String, Object> params) {
        List<User> list = (List<User>) params.get("list");
        StringBuilder sql = new StringBuilder();
        sql.append("INSERT INTO user (username, email, phone, name, status, create_time) VALUES ");
        for (int i = 0; i < list.size(); i++) {
            sql.append(String.format(
                    "(#{list[%d].username}, #{list[%d].email}, #{list[%d].phone}, " +
                    "#{list[%d].name}, #{list[%d].status}, #{list[%d].createTime})",
                    i, i, i, i, i, i
            ));
            if (i < list.size() - 1) {
                sql.append(",");
            }
        }
        return sql.toString();
    }
    public String selectByFields(Map<String, Object> params) {
        List<String> usernames = (List<String>) params.get("usernames");
        List<String> emails = (List<String>) params.get("emails");
        List<String> phones = (List<String>) params.get("phones");
        StringBuilder sql = new StringBuilder();
        sql.append("SELECT * FROM user WHERE ");
        List<String> conditions = new ArrayList<>();
        if (!CollectionUtils.isEmpty(usernames)) {
            conditions.add("username IN (" + 
                    usernames.stream().map(u -> "'" + u + "'").collect(Collectors.joining(",")) + 
                    ")");
        }
        if (!CollectionUtils.isEmpty(emails)) {
            conditions.add("email IN (" + 
                    emails.stream().map(e -> "'" + e + "'").collect(Collectors.joining(",")) + 
                    ")");
        }
        if (!CollectionUtils.isEmpty(phones)) {
            conditions.add("phone IN (" + 
                    phones.stream().map(p -> "'" + p + "'").collect(Collectors.joining(",")) + 
                    ")");
        }
        sql.append(String.join(" OR ", conditions));
        return sql.toString();
    }
}

5 性能优化建议

@Configuration
public class ImportConfig {
    // 配置线程池用于异步导入
    @Bean("importExecutor")
    public ExecutorService importExecutor() {
        return new ThreadPoolExecutor(
                5,  // 核心线程数
                10, // 最大线程数
                60L, TimeUnit.SECONDS, // 空闲线程存活时间
                new LinkedBlockingQueue<>(1000), // 任务队列
                new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
        );
    }
    // 批量导入优化：使用批处理
    @Component
    public static class BatchImportOptimizer {
        @Value("${import.batch-size:500}")
        private int batchSize;
        @Value("${import.concurrent-threads:5}")
        private int concurrentThreads;
        @Autowired
        private ExecutorService importExecutor;
        public <T> ImportResult<T> parallelImport(
                List<T> dataList,
                Function<List<T>, ImportResult<T>> importFunction) {
            // 分割数据
            List<List<T>> batches = Lists.partition(dataList, batchSize);
            // 并行导入
            List<CompletableFuture<ImportResult<T>>> futures = new ArrayList<>();
            for (List<T> batch : batches) {
                CompletableFuture<ImportResult<T>> future = 
                        CompletableFuture.supplyAsync(() -> importFunction.apply(batch), importExecutor);
                futures.add(future);
            }
            // 合并结果
            ImportResult<T> finalResult = new ImportResult<>();
            finalResult.setSuccess(true);
            futures.forEach(future -> {
                try {
                    ImportResult<T> result = future.get(5, TimeUnit.MINUTES);
                    finalResult.setTotalCount(finalResult.getTotalCount() + result.getTotalCount());
                    finalResult.setSuccessCount(finalResult.getSuccessCount() + result.getSuccessCount());
                    finalResult.setFailCount(finalResult.getFailCount() + result.getFailCount());
                    if (result.getSuccessData() != null) {
                        if (finalResult.getSuccessData() == null) {
                            finalResult.setSuccessData(new ArrayList<>());
                        }
                        finalResult.getSuccessData().addAll(result.getSuccessData());
                    }
                    if (result.getDuplicateRecords() != null) {
                        if (finalResult.getDuplicateRecords() == null) {
                            finalResult.setDuplicateRecords(new ArrayList<>());
                        }
                        finalResult.getDuplicateRecords().addAll(result.getDuplicateRecords());
                    }
                } catch (Exception e) {
                    log.error("并行导入任务执行失败", e);
                    finalResult.setSuccess(false);
                }
            });
            return finalResult;
        }
    }
}

注意事项

1 性能优化策略

分批处理：500-1000条一批
缓存预热：批量查询前先加载缓存
批量SQL：使用批量插入和批量查询
并行处理：大文件支持并行导入

2 内存优化

// 使用流式处理避免OOM
public class StreamImportHandler {
    public void processLargeFile(InputStream inputStream) {
        try (BufferedReader reader = new BufferedReader(
                new InputStreamReader(inputStream, StandardCharsets.UTF_8))) {
            List<User> batch = new ArrayList<>(500);
            String line;
            int lineNumber = 0;
            while ((line = reader.readLine()) != null) {
                lineNumber++;
                if (lineNumber == 1) continue; // 跳过表头
                User user = parseLine(line);
                batch.add(user);
                if (batch.size() >= 500) {
                    processBatch(batch);
                    batch.clear();
                }
            }
            // 处理最后一批
            if (!batch.isEmpty()) {
                processBatch(batch);
            }
        } catch (IOException e) {
            log.error("流式处理文件失败", e);
        }
    }
}

3 事务管理

@Configuration
public class TransactionConfig {
    @Bean
    public PlatformTransactionManager transactionManager(DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    }
    // 设置事务超时时间
    @Bean
    public TransactionTemplate transactionTemplate(
            PlatformTransactionManager transactionManager) {
        TransactionTemplate template = new TransactionTemplate(transactionManager);
        template.setTimeout(30); // 30秒超时
        return template;
    }
}

这个案例提供了完整的批量导入查重解决方案，包括文件内部查重、数据库查重、结果返回等核心功能,并考虑了性能优化和异常处理。

本文地址： https://www.bmjup.com/post/10508.html