例如,我们通过 https://intrinio.com/tutorial/web_api
这个网址提供的api获取股市信息的csv数据,现在要下载大量csv数据文件,并将其转换为xml文件。
要求:使用多线程来提高下载并处理的效率。
解决方案:
使用标准库threading.Thread
类创建多个线程,在每个线程中下载并转换一只csv数据。
threading.Thread
类:class threading.Thread(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)
调用该类时,必需带有关键字参数。参数如下:
group应该为None;为了日后扩展ThreadGroup类实现而保留。 target是用于run()方法调用的可调用对象。默认是None,表示不需要调用任何方法。 name是线程名称。默认情况下,由"Thread-N" 格式构成一个唯一的名称,其中N是小的十进制数。 args是用于调用目标函数的参数元组。默认是()。 kwargs是用于调用目标函数的关键字参数字典。默认是{}。 如果daemon不是None,线程将被显式的设置为守护模式,不管该线程是否是守护模式。如果是None(默认值),线程将继承当前线程的守护模式属性。
threading.Thread
类有以下方法:
start()
开始线程活动。 它在一个线程里最多只能被调用一次。它安排对象的run()方法在一个独立的控制进程中调用。如果同一个线程对象中调用这个方法的次数大于一次,会抛出RuntimeError。
run()
代表线程活动的方法。 你可以在子类型里重载这个方法。 标准的run()方法会对作为target参数传递给该对象构造器的可调用对象(如果存在)发起调用,并附带从args和kwargs参数分别获取的位置和关键字参数。
join(timeout=None)
等待,直到线程终结。这会阻塞调用这个方法的线程,直到被调用join()的线程终结,不管是正常终结还是抛出未处理异常,或者直到发生超时,超时选项是可选的。 当timeout参数存在而且不是None时,它应该是一个用于指定操作超时的以秒为单位的浮点数(或者分数)。因为join()总是返回None,所以你一定要在join()后调用is_alive()才能判断是否发生超时,如果线程仍然存货,则join()超时。 当timeout参数不存在或者是None ,这个操作会阻塞直到线程终结。一个线程可以被join()很多次。 如果尝试加入当前线程会导致死锁,join()会引起RuntimeError异常。如果尝试join()一个尚未开始的线程,也会抛出相同的异常。
import requestsimport base64import csvimport timefrom io import StringIOfrom xml.etree.ElementTree import ElementTree, Element, SubElement USERNAME = b'7f304a2df40829cd4f1b17d10cda0304'PASSWORD = b'aff978c42479491f9541ace709081b99'def download_csv(page_number): print('download csv data [page=%s]' % page_number) url = "https://api.intrinio.com/price.csv?ticker=AAPL&hide_paging=true&page_size=200&page_number=%s" % page_number auth = b'Basic' + base64.b64encode(b'%s:%s' % (USERNAME, PASSWORD)) headers = {'Authorization' : auth} response = requests.get(url, headers=headers) if response.ok: return StringIO(response.text)def csv_to_xml(csv_file, xml_path): print('Convert csv data to %s' % xml_path) reader = csv.reader(csv_file) headers = next(reader) root = Element('Data') root.text = '\n\t' root.tail = '\n' for row in reader: book = SubElement(root, 'Row') book.text = '\n\t\t' book.tail = '\n\t' for tag, text in zip(headers, row): e = SubElement(book, tag) e.text = text e.tail = '\n\t\t' e.tail = '\n\t' book.tail = '\n' ElementTree(root).write(xml_path, encoding='utf8')def download_and_save(page_number, xml_path): # IO csv_file = None while not csv_file: csv_file = download_csv(page_number) # CPU csv_to_xml(csv_file, 'data%s.xml' % page_number)if __name__ == '__main__': t0 = time.time() for i in range(1, 6): download_and_save(i, 'data%s.xml' % i) print(time.time() - t0) print('main thread end.')
改进:使用threading.Thread
类创建多个线程,同时下载转换。
import requestsimport base64import csvimport timefrom io import StringIOfrom xml.etree.ElementTree import ElementTree, Element, SubElementfrom threading import Thread USERNAME = b'7f304a2df40829cd4f1b17d10cda0304'PASSWORD = b'aff978c42479491f9541ace709081b99'def download_csv(page_number): print('download csv data [page=%s]' % page_number) url = "https://api.intrinio.com/price.csv?ticker=AAPL&hide_paging=true&page_size=200&page_number=%s" % page_number auth = b'Basic' + base64.b64encode(b'%s:%s' % (USERNAME, PASSWORD)) headers = {'Authorization' : auth} response = requests.get(url, headers=headers) if response.ok: return StringIO(response.text)def csv_to_xml(csv_file, xml_path): print('Convert csv data to %s' % xml_path) reader = csv.reader(csv_file) headers = next(reader) root = Element('Data') root.text = '\n\t' root.tail = '\n' for row in reader: book = SubElement(root, 'Row') book.text = '\n\t\t' book.tail = '\n\t' for tag, text in zip(headers, row): e = SubElement(book, tag) e.text = text e.tail = '\n\t\t' e.tail = '\n\t' book.tail = '\n' ElementTree(root).write(xml_path, encoding='utf8')def download_and_save(page_number, xml_path): # IO csv_file = None while not csv_file: csv_file = download_csv(page_number) # CPU csv_to_xml(csv_file, 'data%s.xml' % page_number)class MyThread(Thread): def __init__(self, page_number, xml_path): super().__init__() self.page_number = page_number self.xml_path = xml_path def run(self): #线程运行download_and_save()方法 download_and_save(self.page_number, self.xml_path)if __name__ == '__main__': t0 = time.time() thread_list = [] for i in range(1, 6): t = MyThread(i, 'data%s.xml' % i) t.start() #启动线程 thread_list.append(t) for t in thread_list: t.join() #阻塞线程,主线程等待所有子线程结束 print(time.time() - t0) print('main thread end.')
上面url已失效,无法看到实际耗时效果,但通过多个线程可以节约IO时间,Python中的多线程不适合处理CPU密集型的任务。