@SupportsBatching 是 NiFi 处理器上的标记注解，启用后会在 UI 调度页显示“Run Duration”滑块，允许框架将多次 onTrigger() 调用合并为一个时间窗口内的批量处理，从而在延迟与吞吐量之间进行权衡。
`@supportsbatching` 是 nifi 处理器上的标记注解，启用后会在 ui 调度页显示“run duration”滑块，允许框架将多次 `ontrigger()` 调用合并为一个时间窗口内的批量处理，从而在延迟与吞吐量之间进行权衡。
@SupportsBatching 并非功能增强型注解，而是一个调度行为声明——它向 NiFi 框架明确告知：“本处理器支持以时间窗口为单位的批量触发”，进而解锁两项关键机制：
- UI 可配置的 Run Duration 控制项：启用该注解后，处理器配置界面的 Scheduling 选项卡中将出现「Run Duration」滑块（默认值通常为 0 sec），用户可设置如 100 ms、2 sec 或 30 sec 等时间阈值；
- 底层执行模型优化：NiFi 调度器将尽可能在指定时间内多次调用该处理器的 onTrigger(Context, Session) 方法，并复用同一 ProcessSession 实例（而非每次新建），同时延迟提交（session.commit()）直至批次结束或超时。
批处理的实际效果示例
假设你开发了一个自定义处理器 FetchFromKafkaProcessor 并添加了该注解：
@SupportsBatching
@Processor(name = "Fetch From Kafka", ...)
public class FetchFromKafkaProcessor extends AbstractProcessor {
@Override
public void onTrigger(final ProcessorContext context, final ProcessSessionFactory sessionFactory) throws ProcessException {
final ProcessSession session = sessionFactory.createSession(); // 可能复用同一 session
try {
// 在 Run Duration 时间内，此方法可能被连续调用多次
// 每次都向同一 session 添加 FlowFile（例如批量拉取 50 条 Kafka 消息）
List<consumerrecord byte>> records = pollKafka();
for (ConsumerRecord<byte byte> r : records) {
FlowFile flowFile = session.create();
flowFile = session.write(flowFile, out -> out.write(r.value()));
session.transfer(flowFile, SUCCESS);
}
// 注意：此处不立即 commit！由框架在批次结束时统一提交
} catch (Exception e) {
session.rollback(true);
throw new ProcessException(e);
}
}
}</byte></consumerrecord>
✅ 正确实践：不主动调用 session.commit() —— NiFi 会在 Run Duration 到期或会话被强制刷新时自动提交，确保元数据与内容仓库（Content Repository / FlowFile Repository）的一致性批量更新。
关键注意事项（务必牢记）
- ❌ 不可用于需强持久化语义的场景：若处理器逻辑依赖 session.commit() 的返回来确认数据已落盘（例如：先写入 NiFi，再从数据库删除源记录），则绝对禁止使用 @SupportsBatching。因为 commit() 调用不保证立即刷盘，仅表示“批次逻辑完成”。
- ⚠️ Run Duration = 0 表示禁用批处理，等效于逐个 FlowFile 触发（最低延迟，但高开销）；
- ⚙️ 非零值（如 5 sec）意味着：NiFi 将尽力让该处理器在 5 秒内持续 onTrigger，聚合处理更多 FlowFile，显著降低线程上下文切换和存储 I/O 频次，提升吞吐量；
- ? 典型适用场景：GetFile、ConsumeKafka、SplitJson（配合后续聚合）、PutDatabaseRecord（批量插入）等 I/O 密集型或可聚合操作。
总结：何时使用？如何选择？
| 场景 | 推荐 Run Duration | 原因 |
|---|---|---|
| 实时告警/低延迟路由（如 RouteOnAttribute） | 0 sec | 避免任何人为延迟，每个 FlowFile 立即流转 |
| 日志归档/批量同步（如 ListS3 + FetchS3Object） | 1–10 sec | 平衡响应速度与 S3 API 调用效率 |
| 数据库批量写入（如 PutDatabaseRecord） | 5–30 sec | 充分利用 JDBC 批处理能力，减少事务开销 |
简言之：@SupportsBatching 是 NiFi 提供的轻量级批处理契约，它不改变处理器逻辑，但赋予调度层智能聚合的能力——开发者只需专注单次 onTrigger 的正确性，框架负责在可控延迟下交付更高吞吐。合理启用，是构建高性能数据流的关键一环。
