如何实现最多同时处理10个文件的事件驱动消费者？

针对你的需求，无需引入缓存组件，直接利用消息队列（Kafka/GCP Pub/Sub）+ 消费者内存状态 + DB原子操作就能实现高效的事件驱动文件处理，完全替代轮询调度器，且满足并发控制和内存优先处理的要求。

核心设计思路 #

1. 生产者侧：现有服务在文件元数据写入DB后，立即将fileId发送到消息队列，依赖队列的持久化特性确保消息至少送达一次。
2. 消费者侧：
   • 用内存维护两个核心线程安全状态：activeProcessingCount（当前正在处理的文件数）、pendingFileIds（待处理的fileId队列）。
   • 仅当activeProcessingCount < 10时，优先从内存队列取任务；内存队列空时，再从消息队列拉取消息补充。
   • 处理完成后，优先从内存队列启动新任务，避免频繁拉取队列。
3. 幂等保障：所有文件处理前先查询DB状态，若已处理则直接跳过，避免重复消费导致的重复操作。

Kafka 具体实现 #

配置与消费控制 #

• 使用Spring Kafka的@KafkaListener，开启手动提交偏移量，确保只有消息被成功处理后才提交偏移量，避免消息丢失。
• 设置max.poll.records为合理值（比如20），控制每次拉取的消息数量，防止内存队列过载。
• 用AtomicInteger维护activeProcessingCount，ConcurrentLinkedQueue维护pendingFileIds，保证线程安全。

执行流程 #

1. 消费者启动后，主动拉取一批消息存入pendingFileIds（暂不提交偏移量）。
2. 当activeProcessingCount < 10时：
   • 若pendingFileIds非空，取出队首fileId启动处理，activeProcessingCount原子加1。
   • 若pendingFileIds为空，触发拉取新消息补充队列。
3. 文件处理完成后：
   • activeProcessingCount原子减1。
   • 提交该消息的偏移量，确认消息已处理。
   • 立即检查pendingFileIds，若有剩余任务则启动下一个处理流程。
   • 当pendingFileIds数量低于阈值（比如5），主动拉取新消息补充队列。
4. 异常处理：处理失败时，可将消息重新放入pendingFileIds重试，或发送到死信队列归档。

GCP Pub/Sub 具体实现 #

配置与消费控制 #

• 使用Spring Cloud GCP Pub/Sub的@PubSubListener，开启手动消息确认，避免未处理的消息被自动移除。
• 同样用AtomicInteger和ConcurrentLinkedQueue维护内存状态，保证线程安全。

执行流程 #

1. 消费者订阅主题后，收到消息时先将fileId和对应的messageId存入pendingFileIds（暂不确认消息）。
2. 当activeProcessingCount < 10时：
   • 取出pendingFileIds中的fileId和messageId，启动处理任务，activeProcessingCount原子加1。
3. 处理成功时：
   • activeProcessingCount原子减1。
   • 调用ack()确认消息，让Pub/Sub移除该消息。
   • 优先从pendingFileIds启动新任务，若队列不足则等待新消息或主动拉取（Pub/Sub支持拉取模式）。
4. 异常处理：处理失败时，可重试或发送到死信主题，避免消息丢失。
5. 拉取模式优化：若采用拉取模式而非推送，可根据activeProcessingCount和pendingFileIds的数量动态调整拉取数量，避免积压。

多Pod场景的全局并发控制 #

如果消费者服务需要部署多个Pod，但全局并发数必须严格控制在10以内，无需缓存组件，直接用DB实现分布式计数：

1. 在DB中创建processing_control表，仅需一个字段current_active_count（初始值0）。
2. 启动新任务前，执行原子更新：

   UPDATE processing_control SET current_active_count = current_active_count + 1 WHERE current_active_count < 10;
   

3. 若更新影响行数为1，说明获取到并发名额，启动处理任务；否则等待直到有任务完成。
4. 处理完成后，执行原子更新current_active_count = current_active_count -1，释放名额。

为什么优于调度器方案 #

• 低延迟：文件上传后立即触发处理，无需等待15秒轮询周期。
• 资源高效：消费者仅在有任务时占用CPU/内存，轮询会定期消耗DB和计算资源。
• 扩展性：后续调整并发上限只需修改配置，无需修改调度逻辑；多Pod场景下通过DB计数轻松实现全局控制。

内容的提问来源于stack exchange，提问作者Rahul