5

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

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

我们已经探索了使用gevent和grequests库（请参阅为什么 gevent.spawn 在调用 Greenlet.join 之前不执行参数化函数？）。我们的工作代码如下所示：

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 永远不会被命中。

我们如何才能让这些异步请求执行？

4

4 回答 4

1

您可能希望使用该join方法而不是sleep然后检查状态。如果你想一次执行一个，那将解决问题。稍微修改您的代码以测试它似乎工作正常。

import gevent
import requests

def execute_task(worker, job):

    print "About to spawn request"
    greenlet = gevent.spawn(requests.get, 'http://example.com', params={})

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

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

execute_task(None, None)

给出结果：

About to spawn request
Request spawned, about to call sleep
Greenlet status:  True
<Response [200]>

这个 Python 进程中是否还有更多可能阻止 Gevent 运行这个 greenlet？

于 2013-04-21T15:51:46.227 回答
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 的对象运行，请创建对象然后在实例上调用“start”

于 2013-04-03T07:41:51.610 回答
1

您必须在不同的线程中运行它或使用内置的 asyncore 库。大多数库都会在你不知道的情况下使用线程，或者它将依赖于 Python 的标准部分 asyncore。

这是线程和异步的组合：

#!/usr/bin/python
# -*- coding: iso-8859-15 -*-
import asyncore, socket
from threading import *
from time import sleep
from os import _exit
from logger import *  # <- Non-standard library containing a log function
from config import *  # <- Non-standard library containing settings such as "server"

class logDispatcher(Thread, asyncore.dispatcher):
    def __init__(self, config=None):
        self.inbuffer = ''
        self.buffer = ''
        self.lockedbuffer = False
        self.is_writable = False

        self.is_connected = False

        self.exit = False
        self.initated = False

        asyncore.dispatcher.__init__(self)
        Thread.__init__(self)

        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            self.connect((server, server_port))
        except:
            log('Could not connect to ' + server, 'LOG_SOCK')
            return None

        self.start()

    def handle_connect_event(self):
        self.is_connected = True

    def handle_connect(self):
        self.is_connected = True
        log('Connected to ' + str(server), 'LOG_SOCK')

    def handle_close(self):
        self.is_connected = False
        self.close()

    def handle_read(self):
        data = self.recv(8192)
        while self.lockedbuffer:
            sleep(0.01)

        self.inbuffer += data


    def handle_write(self):
        while self.is_writable:
            sent = self.send(self.buffer)
            sleep(1)

            self.buffer = self.buffer[sent:]
            if len(self.buffer) <= 0:
                self.is_writable = False
            sleep(0.01)

    def _send(self, what):
        self.buffer += what + '\r\n'
        self.is_writable = True

    def run(self):
        self._send('GET / HTTP/1.1\r\n')

while 1:
    logDispatcher() # <- Initate one for each request.
    asyncore.loop(0.1)
    log('All threads are done, next loop in 10', 'CORE')
    sleep(10)

或者你可以简单地做一个线程来完成这项工作然后死掉。

from threading import *
class worker(Thread):
    def __init__(self, host, postdata)
        Thread.__init__(self)
        self.host = host
        self.postdata = postdata
        self.start()
    def run(self):
        sock.send(self.postdata) #Pseudo, create the socket!

for data in postDataObjects:
    worker('example.com', data)

如果您需要限制线程数（如果您发送的帖子超过 5k 条左右，可能会对系统造成负担），只需执行 awhile len(enumerate()) > 1000: sleep(0.1)并让 looper 对象等待几个线程消失。

于 2013-04-03T07:49:02.223 回答
0

将你的 url 和 params 包装在一个列表中，然后每次将一对弹出到任务池（这里的任务池要么有一个任务，要么是空的），创建线程，从任务池中读取任务，当一个线程获得任务并发送请求，然后从您的列表中弹出另一个（即这实际上是一个队列列表）

于 2013-04-03T07:49:42.010 回答