一个生产者消费者问题

[sununiq]

目前有这样一个场景，每次会以文件流的形式推送一批数据（大概 500W 条）给服务器解析入库，该数据会先保存在磁盘上。因为读取数据到内存解析比较快，入库是 IO 操作比较慢，所以目前想到的思路是采用单生产者多消费者模型，其中多消费者采用线程池管理，Java 中使用 BlockingQueue 来做中间管道。

目前有这么一个需求，如果其中一个消费者入库出现异常，需要通知其他消费者停止消费，同时还要告知生产者放弃解析这批数据，直到下一批数据过来在此解析，请问下这种有什么好的方式来实现？

[zhx1991]

不同数据本身有特征(比如 id)可以分开吗?

同类的数据(比如相同的 id, 需要严格保持先后顺序)可以走同一个管道, 然后这样某个管道挂了别的管道也依然可以正常处理别的数据

[qk3z]

生产者和消费者共用一个标识符，生产者在每次解析的时候判断，消费者在每次入库的时候判断，但是这样感觉性能跟不上啊 -_-！

[sununiq]

@zhx1991 需求就是出现问题，整个这批数据都不处理

[sununiq]

@qk3z 是啊，没有想到啥好办法

[crossoverJie]

实时性要求高嘛？可以借助 MQ 失败之后发个消息，所有线程收到这个消息后停止工作。

[sununiq]

@crossoverJie 实时性要求不高，但是增加组件的话会增加整个系统的复杂度，有没有其他不增加其他组件的方法？