本章介绍了从Python3.2以来引入的concurrent.futures模块，阻塞性I/O与GIL，以及期物的概念。
这一章给的示例代码中的网页http://flupy.org/data/flags目前无法访问，这里就不再复制无法执行的示例代码了，反正在这里也可直接查看。
作为看完后的应用，试着对一段实际代码进行改进。
下面是一段通过访问网易财经的接口得到上证和深证股票历史数据，写入csv文件的程序
# -*- coding: utf-8 -*-
import csv
import datetime
import requests
from lxml import etree
now = datetime.datetime.now()
today = now.strftime('%Y%m%d')
before = now + datetime.timedelta(-1)
yesterday = before.strftime('%Y%m%d')
def getHTMLtext(url, param={}, code='ANSI'):
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.115 Safari/537.36'}
try:
r = requests.get(url, param, headers=headers)
r.encoding = code
r.raise_for_status()
return r.text
except:
return ''
def stockList(url, param={}):
stockSH, stockSZ = [], []
text = getHTMLtext(url, param)
html = etree.HTML(text)
for url in html.xpath('//a/@href'):
if 'http://quote.eastmoney.com/sh' in url:
stockSH.append(url.split('sh')[1][:6])
elif 'http://quote.eastmoney.com/sz' in url:
stockSZ.append(url.split('sz')[1][:6])
print('all stock number got!')
return stockSH, stockSZ
def stockInfo(lst, strnum):
# 依序下载，无并发
print('start get all stock history, will take a long time, please wait...')
L = []
for i in lst:
# code=0000001分开看，0是上证，1是深证，000001是股票代码
history_url = "http://quotes.money.163.com/service/chddata.html?code={0}&start={1}&end={2}&fields=TCLOSE;HIGH;LOW;TOPEN;LCLOSE;CHG;PCHG;VOTURNOVER;VATURNOVER".format(strnum+i, '20170801', yesterday)
perday = getHTMLtext(history_url).split('\r\n')
if len(perday) <= 2:
continue
perday.pop()
for day in perday[1:]:
L.append(day.split(','))
print('all stock info got!')
with open(today+'stock.csv', 'a+', newline='', encoding='gb18030') as file:
w = csv.writer(file)
w.writerows(L)
return L
def main():
stockList_url = "http://quote.eastmoney.com/stocklist.html"
SH, SZ = stockList(stockList_url)
stockInfo(SH, '0')
stockInfo(SZ, '1')
if __name__ == "__main__":
main()
由于是依序下载，速度比较慢，这段代码要运行十分钟以上才执行完毕。
那么，现在是利用本章内容加速代码运行的时候了。
很明显，stockInfo函数中的for循环互相之间没有联系，可以改成函数以便并发调用。
# -*- coding: utf-8 -*-
import csv
import datetime
from concurrent import futures
import requests
from lxml import etree
MAX_WORKERS = 20
L = []
now = datetime.datetime.now()
today = now.strftime('%Y%m%d')
before = now + datetime.timedelta(-1)
yesterday = before.strftime('%Y%m%d')
def getHTMLtext(url, param={}, code='ANSI'):
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.115 Safari/537.36'}
try:
r = requests.get(url, param, headers=headers)
r.encoding = code
r.raise_for_status()
return r.text
except:
return ''
def stockList(url, param={}):
stockSH, stockSZ = [], []
text = getHTMLtext(url, param)
html = etree.HTML(text)
for url in html.xpath('//a/@href'):
if 'http://quote.eastmoney.com/sh' in url:
stockSH.append(url.split('sh')[1][:6])
elif 'http://quote.eastmoney.com/sz' in url:
stockSZ.append(url.split('sz')[1][:6])
print('all stock number got!')
return stockSH, stockSZ
def stockInfo(i, num: str):
# code=0000001应该分开看，0是上证，1是深证，000001是股票代码
history_url = "http://quotes.money.163.com/service/chddata.html?code={0}&start={1}&end={2}&fields=TCLOSE;HIGH;LOW;TOPEN;LCLOSE;CHG;PCHG;VOTURNOVER;VATURNOVER".format(num+i, '20170820', yesterday)
perday = getHTMLtext(history_url).split('\r\n')
if len(perday) > 2:
perday.pop()
for day in perday[1:]:
L.append(day.split(','))
return L
def write2CSV(lst: list):
with open(today+'stock.csv', 'a+', newline='', encoding='gb18030') as file:
w = csv.writer(file)
w.writerows(lst)
def downloadOne(i):
luca = stockInfo(i, '0')
write2CSV(luca)
# stockInfo(lst, '1') 没有跑深证的股票
def downloadMany(lst: list):
workers = min(MAX_WORKERS, len(lst))
with futures.ThreadPoolExecutor(workers) as e:
res = e.map(downloadOne, sorted(lst))
return len(list(res))
def main():
stockList_url = "http://quote.eastmoney.com/stocklist.html"
SH, SZ = stockList(stockList_url)
downloadMany(SH)
# downloadMany(SZ)
if __name__ == "__main__":
main()
由于开了workers个线程，速度近似提高了workers倍，明显感觉到一下子就执行完毕了。
事实上，很多时候Python的多线程是无法提速的。这是因为CPython解释器本身不是线程安全的，因此存在全局解释器锁(GIL, global interpreter lock)，1次只允许1个线程执行Python代码，因此，1个Python进程通常不能使用多个CPU核心。与Python语言本身无关，Jython等没有这个限制。
不过，标准库中所有执行阻塞型I/O操作的函数在等待操作系统返回结果时都会释放GIL，这意味着I/O密集型Python程序在多线程下可以正常运转：1个Python线程等待网络响应时，阻塞型I/O函数会释放GIL，从而再运行1个线程。
包括 time.sleep()函数，即使sleep(0)，也会释放GIL。
那么，如果是CPU密集型Python程序，无法使用多线程时，怎么办？
这时可以使用多进程。
使用concurrent.futures模块启动多进程非常简单，只需要把
def downloadMany(lst: list):
workers = min(MAX_WORKERS, len(lst))
with futures.ThreadPoolExecutor(workers) as e:
res = e.map(downloadOne, sorted(lst))
改成
def downloadMany(lst: list):
with futures.ProcessPoolExecutor() as e:
res = e.map(downloadOne, sorted(lst))
就可以了。
ThreadPoolExecutor()函数需要1个参数指定线程池中线程的数量，而ProcessPoolExecutor()函数通常不需要指定进程数，默认是os.cpu_count()函数返回的CPU数量，也可以自行指定其他值，但对CPU密集型的处理而言，进程数不得超过CPU数量。
实际上，对于I/O密集型程序，这个框架提供的多进程方式做不到进程间通信，不会节省运行时间，只是同时在跑4个完全一样的程序最终得到4组相同数据而已。要实现进程间通信应该使用multiprocessing模块。
译者把future翻译成期物，future是一种对象，表示异步执行的操作，通常自己不应该创建它，而是由并发框架实例化。
要创建的话，可以这样写
def downloadMany(lst: list):
workers = min(MAX_WORKERS, len(lst))
with futures.ThreadPoolExecutor(workers) as e:
to_do = []
for i in sorted(lst):
future = e.submit(downloadOne, i)
to_do.append(future)
results = []
for future in futures.as_completed(to_do):
res = future.result()
results.append(res)
return len(results)
其实这个函数的return并没有多大作用，发生异常的话会在此抛出而已。
上面有一点没有提到，如何向map函数传入多个参数？不过网上有一堆教程，暂时就不写了。
