ItemProcessor 详解
本章定位：深入掌握 ItemProcessor——批处理中数据转换和过滤的核心接口，理解转换、过滤、校验和链式处理的全部模式。
定义与作用
ItemProcessor<I, O> 是 Chunk Step 中的可选处理组件，位于 Reader 和 Writer 之间，负责将输入类型 I 转换为输出类型 O。
@FunctionalInterface
public interface ItemProcessor<I, O> {
@Nullable
O process(I item) throws Exception;
}
飞翔科技架构师白歌的比喻：
"Reader 是原料进货员，Writer 是成品入库员，Processor 是加工车间。原料进去，成品出来。如果原料不合格（return null），直接丢弃不进库。如果加工时出问题（抛异常），这条就报废了。"
核心原理
Processor 的三种返回模式
常见使用模式
| 模式 | 返回 | 说明 |
|---|---|---|
| 数据转换 | 新对象 | 输入 DTO → 输出 Entity |
| 数据增强 | 增强后的输入 | 补充/计算字段后返回同类型 |
| 数据过滤 | null | 不符合条件的数据丢弃 |
| 数据校验 | 校验失败抛异常 | 由 Skip/Retry 策略处理 |
| 复合处理 | 链式 | CompositeItemProcessor 串联多个 Processor |
完整示例
场景一：飞翔科技——学生数据转换与 GPA 计算
@Component
public class StudentProcessor implements ItemProcessor<StudentDTO, StudentEntity> {
@Override
public StudentEntity process(StudentDTO dto) {
// 过滤：年龄无效的学生
if (dto.getAge() == null || dto.getAge() < 10 || dto.getAge() > 100) {
System.out.println("过滤无效年龄: " + dto.getStudentId() + " age=" + dto.getAge());
return null;
}
// 转换：DTO → Entity
StudentEntity entity = new StudentEntity();
entity.setStudentId(dto.getStudentId());
entity.setName(dto.getName().trim()); // 去除空格
entity.setEmail(dto.getEmail().toLowerCase()); // 邮箱小写
entity.setAge(dto.getAge());
entity.setImportTime(LocalDateTime.now()); // 补充导入时间
// 校验：邮箱格式
if (!entity.getEmail().contains("@")) {
throw new ValidationException("Invalid email: " + entity.getEmail());
}
return entity;
}
}
场景二：链式处理——CompositeItemProcessor
@Bean
public CompositeItemProcessor<OrderDTO, OrderEntity> orderProcessorChain() {
CompositeItemProcessor<OrderDTO, OrderEntity> processor = new CompositeItemProcessor<>();
processor.setDelegates(List.of(
// Step 1: 过滤（取消的订单）
(ItemProcessor<OrderDTO, OrderDTO>) item -> {
if ("CANCELLED".equals(item.getStatus())) return null;
return item;
},
// Step 2: 转换 + 计算
(ItemProcessor<OrderDTO, OrderEntity>) item -> {
OrderEntity entity = new OrderEntity();
entity.setOrderId(item.getOrderId());
entity.setAmount(item.getAmount());
entity.setTaxAmount(item.getAmount().multiply(new BigDecimal("0.13")));
return entity;
},
// Step 3: 校验
(ItemProcessor<OrderEntity, OrderEntity>) entity -> {
if (entity.getAmount().compareTo(BigDecimal.ZERO) <= 0) {
throw new ValidationException("Negative amount: " + entity.getOrderId());
}
return entity;
}
));
return processor;
}
操作前后对比：
| 维度 | 操作前（手动 if-else） | 操作后（Processor 链） |
|---|---|---|
| 责任分离 | 过滤+转换+校验混在一个方法 | 三个独立 Processor |
| 可测试性 | 难以单独测试过滤逻辑 | 每个 Processor 可独立单元测试 |
| 可复用性 | 无法单独复用过滤逻辑 | 过滤 Processor 可用于其他 Job |
| 异常管理 | try-catch 嵌套 | 异常向上传播，由 Step 容错策略处理 |
易错场景与避坑
反例一：在 Processor 中执行耗时 I/O
// ❌ 每条数据调一次外部 API → 10 万条 = 10 万次 HTTP 请求
@Override
public StudentEntity process(StudentDTO dto) {
String result = restTemplate.getForObject(
"http://external-api/validate/" + dto.getStudentId(), String.class);
// ...
}
正确做法：要么将外部 API 调用移到 Reader 中一次性批量获取，要么使用 ItemProcessor + 缓存（如 LoadingCache）减少重复调用。
反例二：return null 会被误当成过滤
// ❌ 意外返回 null → 数据被静默丢弃！
@Override
public StudentEntity process(StudentDTO dto) {
if (dto == null) {
return null; // 看起来无害，但数据可能真的被丢了
}
return transform(dto);
}
面试高频考点
Q1：Processor 返回 null 和抛出异常的区别？
返回 null：正常过滤，该条不计入 writeCount 和 skipCount，Writer 不会收到。抛出异常：异常情况，进入容错处理（Skip/Retry），计入 skipCount，达到 skipLimit 后 Step 失败。
Q2：ItemProcessor 为什么不建议做 IO 操作？
每条数据执行一次 → 大量网络/磁盘 IO 会严重降低吞吐量。Chunk Size 50 时，Reader 一次读 50 条只需一次 IO，Writer 一次写 50 条只需一次 IO，但 Processor 处理每一条都需要一次 IO → 成为瓶颈。