Различия в пропускной способности при использовании сопрограмм и потоков

Как видите, в моей программе нет кода, который намеренно передает выполнение одного потока другому потоку. Я не понимаю, как это вписывается в приведенный выше сценарий, поскольку я хотел бы, чтобы все потоки выполнялись одновременно.

Существует один поток ОС, но несколько гринлетов. В вашем случае gevent.sleep() позволяет рабочим выполняться одновременно. Блокирующие вызовы ввода-вывода, такие как urllib2.urlopen(url).read(), делают то же самое, если вы используете urllib2 patched для работы с gevent (путем вызова gevent.monkey.patch_*()).

См. также A Curious Course on Coroutines and Concurrency, чтобы понять, как код может работать одновременно в однопоточная среда.

Чтобы сравнить различия в пропускной способности между gevent, threading, multiprocessing, вы можете написать код, совместимый со всеми подходами:

#!/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()

gevent отлично подходит, когда у вас очень много (зеленых) тем. Я протестировал его с тысячами, и он работал очень хорошо. вы должны убедиться, что все библиотеки, которые вы используете как для очистки, так и для сохранения в БД, становятся зелеными. afaik, если они используют сокет python, инъекция gevent должна работать. Однако расширения, написанные на C (например, mysqldb), будут блокироваться, и вместо этого вам нужно будет использовать зеленые эквиваленты.

если вы используете gevent, вы можете в основном покончить с очередями, создать новый (зеленый) поток для каждой задачи, код для потока будет таким же простым, как db.save(web.get(address)). gevent позаботится о вытеснении, когда какая-либо библиотека в db или веб-блоках. он будет работать до тех пор, пока ваши задачи помещаются в память.

В этом случае ваша проблема связана не со скоростью программы (т.е. с выбором gevent или threading), а с пропускной способностью сетевого ввода-вывода. Это (должно быть) узкое место, определяющее скорость работы программы.

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!

Вам не нужно ждать завершения очереди ввода и обработки, я только что продемонстрировал это здесь.