如何让龙卷风执行并发代码？

Question

我试图写一个龙卷风服务器的简单工作负载生成，这里是它的简化版本：

class EventsLoader(object): 

    generate_num_requests = 1000 
    generate_concurrency = 32 
    server_port = 8001 

    def __init__(self, conf_file): 
     self.parse_config(conf_file) 
     self.client = AsyncHTTPClient() 

    def generate(self): 
     IOLoop.current().run_sync(self.generate_work) 

    @gen.coroutine 
    def generate_work(self): 
     self.queue = queues.Queue() 
     IOLoop.current().spawn_callback(self.fetch_requests) 
     for i in range(self.generate_concurrency): 
      yield self.generate_requests(i) 
     print 'before join queue size: %s' % self.queue.qsize() 
     yield self.queue.join() 

    @gen.coroutine 
    def generate_requests(self, i): 
     load = self.generate_num_requests/self.generate_concurrency 
     for j in range(load): 
      request = self.generate_request(i * 1000 + j) 
      self.queue.put(request) 

    @gen.coroutine 
    def fetch_requests(self): 
     while True: 
      try: 
       request = yield self.queue.get() 
       yield self.client.fetch(request) 
      except Exception as e: 
       print 'failed fetching: %s: %s' % (request.body, e) 
      finally: 
       print 'fetched: %s' % json.loads(request.body)['seq'] 
       self.queue.task_done() 

    def generate_request(self, seq): 
     event = { 
      'seq': seq, 
      # ... more fields here ... 
     } 
     return HTTPRequest(
      'http://localhost:%s/events' % self.server_port, 
      method='POST', 
      body=json.dumps(event), 
     ) 

我看到发生的是，所有的消息fetched: xxxx出现的顺序，这是绝对不可能的，如果发电机的确在同时工作。

如何让它同时运行？在我了解I/O循环是什么以及什么是@gen.coroutine时，肯定会有一些巨大的缺失。即不管我的generate_concurrency设置如何，性能不变。

Answer 1

无论您如何生成请求，您都只能抓取任务。这是取，不发电，你需要并行：

for i in range(self.fetch_concurrency): 
    IOLoop.current().spawn_callback(self.fetch_requests) 

这会给你多fetch_requests工人可以从共享队列中提取工作。

另外，这段代码的生成部分也没有并行运行。 /：与其

for i in range(self.generate_concurrency): 
     yield self.generate_requests(i) 

它等待一个generate_requests呼叫开始下一个之前完成，则可以并行地

yield [self.generate_requests(i) for i in range(self.generate_concurrency)]