python - 如何在 python 中一次发送一个异步 http 请求？

Question

我们有一个作业队列，worker 一次处理一个作业。每个作业都需要我们格式化一些数据并发出 HTTP POST 请求，并将数据作为请求负载。

我们如何让每个工作人员以单线程、非阻塞的方式异步发出这些 HTTP POST 请求？我们不关心请求的响应——我们只希望请求尽快执行，然后工作人员立即转移到下一个工作。

我们探索了使用 gevent 和 grequests 库(参见 Why does gevent.spawn not execute the parameterized function until a call to Greenlet.join? )。我们的 worker 代码看起来像这样:

def execute_task(worker, job):

    print "About to spawn request"
    greenlet = gevent.spawn(requests.post, url, params=params)

    print "Request spawned, about to call sleep"
    gevent.sleep()

    print "Greenlet status: ", greenlet.ready()

第一个 print 语句执行，但第二个和第三个 print 语句永远不会打印，也永远不会命中 url。

我们怎样才能让这些异步请求得到执行？

Answer 1

1) 创建一个 Queue.Queue 对象

2) 创建尽可能多的“工作”线程，循环并从 Queue.Queue 中读取

3) 将作业提供给 Queue.Queue

工作线程将按照放置顺序读取 Queue.Queue

从文件中读取行并将它们放入 Queue.Queue 的示例

import sys
import urllib2
import urllib
from Queue import Queue
import threading
import re

THEEND = "TERMINATION-NOW-THE-END"


#read from file into Queue.Queue asynchronously
class QueueFile(threading.Thread):
    def run(self):
        if not(isinstance(self.myq, Queue)):
            print "Queue not set to a Queue"
            sys.exit(1)
        h = open(self.f, 'r')
        for l in h:
            self.myq.put(l.strip())  # this will block if the queue is full
        self.myq.put(THEEND)

    def set_queue(self, q):
        self.myq = q

    def set_file(self, f):
        self.f = f

关于工作线程可能是什么样子的想法(仅示例)

class myWorker(threading.Thread):
    def run(self):
        while(running):           
            try:
                data = self.q.get()  # read from fifo

                req = urllib2.Request("http://192.168.1.10/url/path")
                req.add_data(urllib.urlencode(data))
                h1 = urllib2.urlopen(req, timeout=10)
                res = h1.read()
                assert(len(res) > 80)

            except urllib2.HTTPError, e:
                print e

            except urllib2.URLError, e:
                print "done %d reqs " % n
                print e
                sys.exit()

要使对象基于 threading.Thread go，创建对象然后在实例上调用“start”