Gevent：2つのコンシューマと2つのコンシューマを同時に使用しないでキューを使用する

Question

コンシューマによって生成された値をディスクに書き込む必要があるという問題があります。私は毎回書くためにファイルの新しいインスタンスを開きたくないので、2番目のキューと他のコンシューマを使って1枚のGreenletからディスクに書き込むことを考えました。私のコードの問題は、2番目のキューが最初のキューから非同期で消費されないということです。最初のキューが最初に終了し、次に2番目のキューが消費されます。 同時にディスクに値を書き込んで、他の値が生成されるようにします。 助けてくれてありがとう！

#!/usr/bin/python 
#- * -coding: utf-8 - * - 
import gevent #pip install gevent 
from gevent.queue import * 
import gevent.monkey 
from timeit import default_timer as timer 
from time import sleep 
import cPickle as pickle 

gevent.monkey.patch_all() 

def save_lineCount(count): 
    with open("count.p", "wb") as f: 
     pickle.dump(count, f) 

def loader(): 
    for i in range(0,3): 
     q.put(i) 

def writer(): 
    while True: 
     task = q_w.get() 
     print "writing",task 
     save_lineCount(task) 

def worker(): 
    while not q.empty(): 
     task = q.get() 
     if task%2: 
      q_w.put(task) 
      print "put",task 
      sleep(10) 

def asynchronous(): 
    threads = [] 
    threads.append(gevent.spawn(writer)) 
    for i in range(0, 1): 
     threads.append(gevent.spawn(worker)) 
    start = timer() 
    gevent.joinall(threads,raise_error=True) 
    end = timer() 
    #pbar.close() 
    print "\n\nTime passed: " + str(end - start)[:6] 


q = gevent.queue.Queue() 
q_w = gevent.queue.Queue() 
gevent.spawn(loader).join() 
asynchronous() 

Answer 1

一般的には、そのアプローチはうまくいくはずです。いくつかの問題は、しかし、この特定のコードであります

• がtime.sleepを呼び出すと、すべてのgreenletsがブロックされます。あなたはgevent.sleepに電話をかけるか、グリーンレットブロックを1つだけ持たせるためにプロセスをモンキーパッチする必要があります（gevent.monkeyがインポートされていますが、patch_allは呼び出されません）。私はそれがここでの大きな問題だと思う。

• ファイルへの書き込みも同期しており、すべてのグリーンレットがブロックされます。大きなボトルネックの場合は、FileObjectThreadを使用できます。