使用协程与线程时的吞吐量差异

Question

几天前，我问了一个关于帮助我设计一个构造多个HTTP请求的范例的问题

这是场景。我想要一个多生产，多消费的系统。我的制作人会抓取一些网站并将找到的链接添加到队列中。因为我将爬网多个站点，所以我希望有多个生产者/爬网者。

使用者/工作者从这个队列中获取信息，向这些链接发出TCP/UDP请求，并将结果保存到我的Django DB。我还希望有多个worker，因为每个队列项彼此完全独立。

人们建议对此使用协同程序库，即Gevent或Eventlet。由于从未使用过协程，我读到，尽管编程范例类似于线程范例，但只有一个线程在积极执行，但当阻塞调用（如I/O调用）发生时，堆栈在内存中切换，另一个绿色线程接管，直到遇到某种阻塞I/O调用。希望我是对的？这是我的一篇SO文章中的代码：

import gevent
from gevent.queue import *
import time
import random

q = JoinableQueue()
workers = []
producers = []


def do_work(wid, value):
    gevent.sleep(random.randint(0,2))
    print 'Task', value, 'done', wid


def worker(wid):
    while True:
        item = q.get()
        try:
            print "Got item %s" % item
            do_work(wid, item)
        finally:
            print "No more items"
            q.task_done()


def producer():
    while True:
        item = random.randint(1, 11)
        if item == 10:
            print "Signal Received"
            return
        else:
            print "Added item %s" % item
            q.put(item)


for i in range(4):
    workers.append(gevent.spawn(worker, random.randint(1, 100000)))

# This doesn't work.
for j in range(2):
    producers.append(gevent.spawn(producer))

# Uncommenting this makes this script work.
# producer()

q.join()

这很好，因为sleep调用正在阻塞调用，当发生sleep事件时，另一个绿色线程将接管。这比顺序执行快得多。 如您所见，我的程序中没有任何代码故意将一个线程的执行结果生成到另一个线程。我不明白这是如何适应上述场景的，因为我希望所有线程同时执行。

所有的工作都很好，但是我觉得我使用Gevent/Eventlets实现的吞吐量比原来的顺序运行的程序要高，但是大大低于使用真正的线程可以实现的吞吐量。

如果我要使用线程机制重新实现我的程序，那么我的每个生产者和消费者都可以同时工作，而无需像协程那样交换堆栈。

这应该使用线程重新实现吗？我的设计错了吗？我没有看到使用协同旅行的真正好处。

也许我的观念有点混乱，但这是我所吸收的。对我的范例和概念的任何帮助或澄清都是很好的。

谢谢

Answer 1

在这种情况下，问题不在于程序速度（即选择gevent或线程），而在于网络IO吞吐量。这应该是决定程序运行速度的瓶颈。

Gevent是确保是瓶颈而不是程序架构的一种好方法。

这是您想要的过程：

import gevent
from gevent.queue import Queue, JoinableQueue
from gevent.monkey import patch_all


patch_all()  # Patch urllib2, etc


def worker(work_queue, output_queue):
    for work_unit in work_queue:
        finished = do_work(work_unit)
        output_queue.put(finished)
        work_queue.task_done()


def producer(input_queue, work_queue):
    for url in input_queue:
        url_list = crawl(url)
        for work in url_list:
            work_queue.put(work)
        input_queue.task_done()


def do_work(work):
    gevent.sleep(0)  # Actually proces link here
    return work


def crawl(url):
    gevent.sleep(0)
    return list(url)  # Actually process url here

input = JoinableQueue()
work = JoinableQueue()
output = Queue()

workers = [gevent.spawn(worker, work, output) for i in range(0, 10)]
producers = [gevent.spawn(producer, input, work) for i in range(0, 10)]


list_of_urls = ['foo', 'bar']

for url in list_of_urls:
    input.put(url)

# Wait for input to finish processing
input.join()
print 'finished producing'
# Wait for workers to finish processing work
work.join()
print 'finished working'

# We now have output!
print 'output:'
for message in output:
    print message
# Or if you'd like, you could use the output as it comes!

您不需要等待输入和工作队列完成，我在这里演示了这一点。

Answer 2

当你有很多（绿色）线程时，gevent很好。我测试了几千次，效果很好。你必须确保所有的库你都使用刮和保存到数据库得到绿色。如果他们使用python的套接字，gevent注入应该可以工作。但是，用C语言编写的扩展（例如mysqldb）会阻塞，您需要使用绿色等价物。

如果您使用gevent，您可以基本上消除队列，为每个任务生成新的（绿色）线程，该线程的代码简单到db.save(web.get(address))。当数据库或web块中的某些库时，gevent将负责抢占。只要你的任务符合记忆，它就会起作用。

Answer 3

As you can see, I don't have any code in my program that purposely yields the execution of one thread to another thread. I fail to see how this fits into scenario above as I would like to have all the threads executing simultaneously.

有一个操作系统线程，但有几个greenlet。在您的情况下，gevent.sleep()允许工人并发执行。如果使用urllib2patched来处理gevent（通过调用gevent.monkey.patch_*()），阻塞IO调用（如urllib2.urlopen(url).read()）也可以这样做。

另请参见A Curious Course on Coroutines and Concurrency以了解代码如何在单线程环境中并发工作。

要比较gevent、线程、多处理之间的吞吐量差异，可以编写与所有aproach兼容的代码：

#!/usr/bin/env python
concurrency_impl = 'gevent' # single process, single thread
##concurrency_impl = 'threading' # single process, multiple threads
##concurrency_impl = 'multiprocessing' # multiple processes

if concurrency_impl == 'gevent':
    import gevent.monkey; gevent.monkey.patch_all()

import logging
import time
import random
from itertools import count, islice

info = logging.info

if concurrency_impl in ['gevent', 'threading']:
    from Queue import Queue as JoinableQueue
    from threading import Thread
if concurrency_impl == 'multiprocessing':
    from multiprocessing import Process as Thread, JoinableQueue

脚本的其余部分对于所有并发实现都是相同的：

def do_work(wid, value):
    time.sleep(random.randint(0,2))
    info("%d Task %s done" % (wid, value))

def worker(wid, q):
    while True:
        item = q.get()
        try:
            info("%d Got item %s" % (wid, item))
            do_work(wid, item)
        finally:
            q.task_done()
            info("%d Done item %s" % (wid, item))

def producer(pid, q):
    for item in iter(lambda: random.randint(1, 11), 10):
        time.sleep(.1) # simulate a green blocking call that yields control
        info("%d Added item %s" % (pid, item))
        q.put(item)
    info("%d Signal Received" % (pid,))

不要在模块级执行代码，将其放入main()：

def main():
    logging.basicConfig(level=logging.INFO,
                        format="%(asctime)s %(process)d %(message)s")

    q = JoinableQueue()
    it = count(1)
    producers = [Thread(target=producer, args=(i, q)) for i in islice(it, 2)]
    workers = [Thread(target=worker, args=(i, q)) for i in islice(it, 4)]
    for t in producers+workers:
        t.daemon = True
        t.start()

    for t in producers: t.join() # put items in the queue
    q.join() # wait while it is empty
    # exit main thread (daemon workers die at this point)

if __name__=="__main__":    
   main()