本文目录导读：
我来介绍一个完整的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;
}
}
这个案例提供了完整的批量导入查重解决方案，包括文件内部查重、数据库查重、结果返回等核心功能,并考虑了性能优化和异常处理。
