multiprocessing.Queue 可以与 gevent 一起使用吗？

发布于 2024-12-06 21:43:47 字数 413 浏览 1 评论 0原文

有人知道这段代码有什么问题吗？它只是永远“加载”。无输出。 “Sites”是一个由几十个字符串组成的列表。

num_worker_threads = 30

def mwRegisterWorker():
    while True:
        try:
            print q.get()
        finally:
            pass

q = multiprocessing.JoinableQueue()
for i in range(num_worker_threads):
     gevent.spawn(mwRegisterWorker)

for site in sites:
    q.put(site)

q.join()  # block until all tasks are done

原文

Anyone know what is wrong with this code? It simply "loads" forever. No output. "Sites" is a list of a few dozen strings.

num_worker_threads = 30

def mwRegisterWorker():
    while True:
        try:
            print q.get()
        finally:
            pass

q = multiprocessing.JoinableQueue()
for i in range(num_worker_threads):
     gevent.spawn(mwRegisterWorker)

for site in sites:
    q.put(site)

q.join()  # block until all tasks are done

分享到QQ

分享到微博

如果你对这篇内容有疑问，欢迎到本站社区发帖提问参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

尸血腥色 2024-12-13 21:43:47

gevent.spawn() 创建 greenlet 而不是进程（甚至更多：所有 greenlet 都在单个操作系统线程中运行）。所以 multiprocessing.JoinableQueue 在这里不合适。

gevent 基于协作多任务处理，即，除非您调用切换到 gevent 事件循环的阻塞函数，否则其他 greenlet 将不会运行。例如，下面的 conn 使用修补过的 gevent 套接字方法，允许其他 greenlet 在等待站点回复时运行。如果没有 pool.join() 将控制权交给运行事件循环的 greenlet，则不会建立连接。

要在向多个站点发出请求时限制并发，您可以使用 gevent.pool.Pool：

#!/usr/bin/env python
from gevent.pool import Pool
from gevent import monkey; monkey.patch_socket()
import httplib # now it can be used from multiple greenlets

import logging
info = logging.getLogger().info

def process(site):
    """Make HEAD request to the `site`."""
    conn = httplib.HTTPConnection(site)
    try:
        conn.request("HEAD", "/")
        res = conn.getresponse()
    except IOError, e:
        info("error %s reason: %s" % (site, e))
    else:
        info("%s %s %s" % (site, res.status, res.reason))
    finally:
        conn.close()

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

    num_worker_threads = 2
    pool = Pool(num_worker_threads)    
    sites = ["google.com", "bing.com", "duckduckgo.com", "stackoverflow.com"]*3
    for site in sites:
        pool.apply_async(process, args=(site,))
    pool.join()

if __name__=="__main__":
   main()

gevent.spawn() creates greenlets not processes (even more: all greenlets run in a single OS thread). So multiprocessing.JoinableQueue is not appropriate here.

gevent is based on cooperative multitasking i.e, until you call a blocking function that switches to gevent's event loop other greenlets won't run. For example conn below uses patched for gevent socket methods that allow other greenlets to run while they wait for a reply from the site. And without pool.join() that gives up control to the greenlet that runs the event loop no connections will be made.

To limit concurrency while making requests to several sites you could use gevent.pool.Pool:

#!/usr/bin/env python
from gevent.pool import Pool
from gevent import monkey; monkey.patch_socket()
import httplib # now it can be used from multiple greenlets

import logging
info = logging.getLogger().info

def process(site):
    """Make HEAD request to the `site`."""
    conn = httplib.HTTPConnection(site)
    try:
        conn.request("HEAD", "/")
        res = conn.getresponse()
    except IOError, e:
        info("error %s reason: %s" % (site, e))
    else:
        info("%s %s %s" % (site, res.status, res.reason))
    finally:
        conn.close()

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

    num_worker_threads = 2
    pool = Pool(num_worker_threads)    
    sites = ["google.com", "bing.com", "duckduckgo.com", "stackoverflow.com"]*3
    for site in sites:
        pool.apply_async(process, args=(site,))
    pool.join()

if __name__=="__main__":
   main()

回复收藏 0 原文