本文內容：

• python multiprocessing.dummy Pool多線程、進程任務隊列使用
• http壓力測試簡單示例

工作中有個常用的場景，比如現在需要下載10W張圖片，我們不可能寫個for循環一張一張的下載吧，又或者是我們做個簡單的HTTP壓力測試肯定是要使用多個，進程或者線程去做（每個請求handler，會有一個參數（所有的參數生成一個隊列））然后把handler和隊列map到Pool里面。肯定要用多線程或者是多進程，然后把這100W的隊列丟給線程池或者進程池去處理在python中multiprocessing Pool進程池，以及multiprocessing.dummy非常好用，一般：

---

• from multiprocessing import Pool as ProcessPool
• from multiprocessing.dummy import Pool as ThreadPool

---

前者是多個進程，后者使用的是線程，之所以dummy（中文意思“假的”）
下面給出一個簡單的http壓力測試示例：

# _*_ coding:utf-8 _*_""" This file a sample demo to do http stress test """ import requests import time from multiprocessing.dummy import Pool as ThreadPool import urllibdef get_ret_from_http(url):"""cited from https://stackoverflow.com/questions/645312/what-is-the-quickest-way-to-http-get-in-python"""ret = requests.get(url)print ret.content# eg. result: {"error":false,"resultMap":{"check_ret":1},"success":true}def multi_process_stress_test():"""start up 4 thread to issue 1000 http requests to serverand test consume time:return:"""start = time.time()# 實際中url帶參數的一般使用下面的make_url函數生成，這里示例就不用（前面寫的現在懶得改了）url = """http://127.0.0.1:9325/shortvideo/checkBlack?url=http%3A%2F%2Fzbasecapture.bs2.yy.com%2F42269159_1499248536403_3.jpg&serial=abcdddddddd"""# generate task queue listlst_url = [url, url1]*50# use 5 threadspool = ThreadPool(5)# task and handles to poolret = pool.map(get_ret_from_http, lst_url)pool.close()pool.join()print 'time consume %s' % (time.time() - start)def make_url():"""generate url with parameterhttps://xy.com/index.php?url=http%3A//xy.xxx.com/22.jpg&SecretId=xy_123_movecited from https://stackoverflow.com/questions/2506379/add-params-to-given-url-in-pythonhttps://github.com/gruns/furl a good util for url operator:return:"""para = {"SecretId": "xy_123_move", "url": "http://xy.xxx.com/22.jpg"}print urllib.urlencode(para)#url=http%3A%2F%2Fxy.xxx.com%2F22.jpg&SecretId=xy_123_movebase_url = 'xy.com/index.php'return 'https://%s?%s' % (base_url, '&'.join('%s=%s' % (k, urllib.quote(str(v))) for k, v in para.iteritems()))if __name__ == '__main__':# get_ret_from_http()multi_process_stress_test()# print make_url()pass

下面在給出另一個簡單的示例，handler函數每次睡眠隨機的秒數（根據指定的參數），我們可以選擇使用進程或者是線程來完成隊列中所有的任務（一般CPU密集型的選擇用多進程，IO密集型的可以選擇多線程）

# _*_ coding:utf-8 _*_ """ This file is about thread(dummy)/process pool """ from multiprocessing import Pool as ProcessPool from multiprocessing.dummy import Pool as ThreadPool import logging from time import sleep, time from random import randrangelogging.basicConfig(level=logging.DEBUG,format='%(levelname)s %(asctime)s %(processName)s %(message)s',datefmt='%Y-%m-%d %I:%M:%S')def handler(sec):logging.debug('now I will sleep %s S', sec)sleep(sec)def get_pool(b_dummy=True, num=4):"""if b_dummy is True then get ThreadPool, or get process pool:param b_dummy: dummy thread Pool or Process pool:param num: thread or process num:return: pool object"""if b_dummy:pool = ThreadPool(num)else:pool = ProcessPool(num)return pooldef test_dummy_thread_pool():start_time = time()# generate task queue parameters listslst_sleep_sec = [randrange(3, 10) for i in xrange(10)]pool = get_pool(b_dummy=False)results = pool.map(handler, lst_sleep_sec)logging.debug(results)pool.close()pool.join()logging.debug('time consume %s', time() - start_time)passif __name__ == '__main__':test_dummy_thread_pool()pass

工作中使用的語言比較多寫過C++，java, 部分html+js, python的.由于用到語言的間歇性，比如還幾個月沒有使用python了許多技巧就忘記了，于是我把一些常用的python代碼分類項目在本人的github中，當實際中用到某一方法的時候就把常用的方法放到一個文件中方便查詢。

總結

以上是生活随笔為你收集整理的python multiprocessing dummy Pool 使用的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。