制作完成后通知消费者

我从LDAP中读取了大量数据,需要将这些数据与数据库中的相应记录进行比较。为了最小化sql查询的数量,我想将多个ldap记录批处理到一个查询中。

所有这些都非常简单:一个线程生成ldap结果,一个线程使用这些结果并运行sql查询。

ldap_results = Queue.Queue(10)
def producer():
  for result in ldap_results():
    ldap_results.put(result)

def consumer():
  buffer = []
  buffer_size = 5
  while True:
    record = ldap_results.get()
    buffer.append(record)
    if len(buffer) >= buffer_size:
      do_sql(buffer)
      buffer = []

问题是:如果ldap只返回3个结果 buffer_size 是5,它会永远阻塞。我意识到我可以把一些特殊的标记放入缓冲区,比如 None ,或 "EOF" ,但这似乎是一个糟糕的设计:“迭代直到完成,哦,除非你看到这个特殊的值,这意味着你也完成了”。

我想出了两个备选方案。首先是分享 eof 变量,但我不知道如何正确地同步它。

def producer():
  while data:
    buffer.put()
  eof = True

def consumer():
  while not eof:
    buffer.get()

第二是要有一个 ProduceChunks(chunk_size) 方法,它将处理结果的批处理,但我不喜欢这样,因为它假定生产者知道如何最好地缓冲结果,而实际上,我认为这是消费者的责任。

有人有什么指导吗?