在 Python 中同时运行 POST

Question

我正在尝试将 100,000 个数据点上传到 Web 服务后端。如果我一次运行一个，大约需要 12 个小时。它们同时支持 20 个 API 调用。如何同时运行此 POST 以加快导入速度？

def AddPushTokens():

 import requests
 import csv
 import json

 count=0
 tokenList=[]

 apikey="12345"
 restkey="12345"
 URL="https://api.web.com/1/install/"
 headers={'content-type': 'application/json','Application-Id': apikey,'REST-API-Key':restkey}

 with open('/Users/name/Desktop/push-new.csv','rU') as csvfile:
      deviceTokens=csv.reader(csvfile, delimiter=',')

      for token in deviceTokens:

       deviceToken=token[0].replace("/","")
       deviceType="ios"
       pushToken="pushtoken_"+deviceToken
       payload={"deviceType": deviceType,"deviceToken":deviceToken,"channels":["",pushToken]}
       r = requests.post(URL, data=json.dumps(payload), headers=headers)

       count=count+1
       print "Count: " + str(count)
       print r.content

编辑：我正在尝试使用 concurrent.futures。我感到困惑的是如何设置它以便从 CSV 中提取令牌并将其传递给 load_url？另外，我想确保它通过前 20 次运行请求，然后在 21 处接收并运行下一组 20 次。

import concurrent.futures
import requests

URLS = ['https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/',
     'https://api.web.com/1/installations/']


apikey="12345"
restkey="12345"
URL="https://api.web.com/1/installations/"
headers={'content-type': 'application/json','X-web-Application-Id': apikey,'X-web-REST-API-Key':restkey}


     with open('/Users/name/Desktop/push-new.csv','rU') as csvfile:
     deviceTokens=csv.reader(csvfile, delimiter=',')

     for token in deviceTokens:

          deviceToken=token[0].replace("/","")
          deviceType="ios"
          pushToken="pushtoken_"+deviceToken
          payload={"deviceType": deviceType,"deviceToken":deviceToken,"channels":["",pushToken]}
          r = requests.post(URL, data=json.dumps(payload), headers=headers)


# Retrieve a single page and report the url and contents
def load_url(token):

     URL='https://api.web.com/1/installations/'

     deviceToken=token[0].replace("/","")
     deviceType="ios"
     pushToken="pushtoken_"+deviceToken
     payload={"deviceType": deviceType,"deviceToken":deviceToken,"channels":["",pushToken]}
     r = requests.post(URL, data=json.dumps(payload), headers=headers)

     count=count+1
     print "Count: " + str(count)
     print r.content

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
    # Start the load operations and mark each future with its URL
    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
        except Exception as exc:
            print('%r generated an exception: %s' % (url, exc))
        else:
            print('%r page is %d bytes' % (url, len(data)))

编辑：根据下面的评论更新

import concurrent.futures
import requests
import csv
import json

apikey="ldy0eSCqPz9PsyOLAt35M2b0XrfDZT1NBW69Z7Bw"
restkey="587XASjEYdQwH2UHruA1yeZfT0oX7uAUJ8kWTmE3"
URL="https://api.parse.com/1/installations/"
headers={'content-type': 'application/json','X-Parse-Application-Id': apikey,'X-Parse-REST-API-Key':restkey}

with open('/Users/jgurwin/Desktop/push/push-new.csv','rU') as csvfile:
     deviceTokens=csv.reader(csvfile, delimiter=',')

     for device in deviceTokens:

        token=device[0].replace("/","")

        # Retrieve a single page and report the url and contents

        def load_url(token):

          count=0
          deviceType="ios"
          pushToken="pushtoken_"+token
          payload={"deviceType": deviceType,"deviceToken":token,"channels":["",pushToken]}
          r = requests.post(URL, data=json.dumps(payload), headers=headers)

          count=count+1
          print "Count: " + str(count)
          print r.content


        # We can use a with statement to ensure threads are cleaned up promptly
          with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
              # Start the load operations and mark each future with its URL
              future_to_token = {executor.submit(load_url, token, 60): token for token in deviceTokens}
              for future in concurrent.futures.as_completed(future_to_url):
                  url = future_to_url[future]
                  try:
                      data = future.result()
                  except Exception as exc:
                      print('%r generated an exception: %s' % (url, exc))
                  else:
                      print('%r page is %d bytes' % (url, len(data)))

Answer 1

执行此操作的简单方法是使用线程。几乎一样简单的方法是使用gevent 或类似的库（grequests 甚至将gevent 和requests 联系在一起，因此您不必弄清楚如何去做）。困难的方法是构建一个事件循环（或者，更好的是，使用 Twisted 或 Tulip 之类的东西）并自己多路复用请求。

让我们用简单的方法来做吧。

您不想一次运行 100000 个线程。除了它会占用数百 GB 的堆栈空间，而且您的 CPU 将花费更多时间进行上下文切换而不是运行实际代码之外，该服务一次仅支持 20 个连接。所以，你需要 20 个线程。

那么，如何在 20 个线程上运行 100000 个任务？使用线程池执行器（或裸线程池）。

concurrent.futures 文档有一个 example，这与您想要做的几乎相同，除了使用 GET 代替 POST 并使用 urllib 代替 requests。只需将 load_url 函数更改为如下所示：

def load_url(token):
    deviceToken=token[0].replace("/","")
    # … your original code here …
    r = requests.post(URL, data=json.dumps(payload), headers=headers)
    return r.content

...示例将按原样运行。

由于您使用的是 Python 2.x，因此 stdlib 中没有 concurrent.futures 模块；你需要后端端口，futures。

---

在 Python（至少是 CPython）中，一次只有一个线程可以做任何 CPU 工作。如果您的任务花费更多时间通过网络下载（I/O 工作）而不是构建请求和解析响应（CPU 工作），那不是问题。但如果不是这样，您将希望使用进程而不是线程。只需将示例中的ThreadPoolExecutor 替换为ProcessPoolExecutor。

---

如果您想完全在 2.7 标准库中执行此操作，那么使用 multiprocessing 中内置的线程和进程池几乎一样简单。请参阅Using a pool of workers 和Process Pools API，如果您想使用线程而不是进程，请参阅multiprocessing.dummy。

Answer 2

可能有点矫枉过正，但您可能想看看Celery。

Tutorial

tasks.py 可能是：

from celery import Celery
import requests

app = Celery('tasks', broker='amqp://guest@localhost//')

apikey="12345"
restkey="12345"

URL="https://api.web.com/1/install/"
headers={'content-type': 'application/json','Application-Id': apikey,'REST-API-Key':restkey}

f = open('upload_data.log', 'a+')
@app.task
def upload_data(data, count):
    r = requests.post(URL, data=data, headers=headers)
    f.write("Count: %d\n%s\n\n" % (count, r.content)

开始 celery 任务：

$ celery -A tasks worker --loglevel=info -c 20

然后在另一个脚本中：

import tasks
def AddPushTokens():

    import csv
    import json

    count=0
    tokenList=[]

    with open('/Users/name/Desktop/push-new.csv','rU') as csvfile:
        deviceTokens=csv.reader(csvfile, delimiter=',')

        for token in deviceTokens:
            deviceToken=token[0].replace("/","")
            deviceType="ios"
            pushToken="pushtoken_"+deviceToken
            payload={"deviceType": deviceType,"deviceToken":deviceToken,"channels":["",pushToken]}
   r = tasks.upload_data.delay(json.dumps(payload), count)

   count=count+1

注意：以上代码为示例。您可能需要根据需要对其进行修改。