Python自学Day64 并发下载

多线程和多进程补充知识点

threading.local类

使用线程时最不愿意遇到的情况就是多个线程竞争资源，在这种情况下为了保证资源状态的正确性，我们可能需要对资源进行加锁保护的处理，这一方面会导致程序失去并发性，另外如果多个线程竞争多个资源时，还有可能因为加锁方式的不当导致死锁。要解决多个线程竞争资源的问题，其中一个方案就是让每个线程都持有资源的副本（拷贝），这样每个线程可以操作自己所持有的资源，从而规避对资源的竞争。

要实现将资源和持有资源的线程进行绑定的操作，最简单的做法就是使用threading模块的local类，在网络爬虫开发中，就可以使用local类为每个线程绑定一个MySQL数据库连接或Redis客户端对象，这样通过线程可以直接获得这些资源，既解决了资源竞争的问题，又避免了在函数和方法调用时传递这些资源。具体的请参考本章多线程爬取“手机搜狐网”（Redis版）的实例代码。

concurrent.futures模块

Python3.2带来了concurrent.futures 模块，这个模块包含了线程池和进程池、管理并行编程任务、处理非确定性的执行流程、进程/线程同步等功能。关于这部分的内容推荐大家阅读《Python并行编程》。

分布式进程

使用多进程的时候，可以将进程部署在多个主机节点上，Python的multiprocessing模块不但支持多进程，其中managers子模块还支持把多进程部署到多个节点上。当然，要部署分布式进程，首先需要一个服务进程作为调度者，进程之间通过网络进行通信来实现对进程的控制和调度，由于managers模块已经对这些做出了很好的封装，因此在无需了解网络通信细节的前提下，就可以编写分布式多进程应用。具体的请参照本章分布式多进程爬取“手机搜狐网”的实例代码。

协程和异步I/O

协程的概念

协程（coroutine）通常又称之为微线程或纤程，它是相互协作的一组子程序（函数）。所谓相互协作指的是在执行函数A时，可以随时中断去执行函数B，然后又中断继续执行函数A。注意，这一过程并不是函数调用（因为没有调用语句），整个过程看似像多线程，然而协程只有一个线程执行。协程通过yield关键字和 send()操作来转移执行权，协程之间不是调用者与被调用者的关系。

协程的优势在于以下两点：

执行效率极高，因为子程序（函数）切换不是线程切换，由程序自身控制，没有切换线程的开销。

不需要多线程的锁机制，因为只有一个线程，也不存在竞争资源的问题，当然也就不需要对资源加锁保护，因此执行效率高很多。

说明：协程适合处理的是I/O密集型任务，处理CPU密集型任务并不是它擅长的，如果要提升CPU的利用率可以考虑“多进程+多线程”或者“多进程+协程”的工作模式。

历史回顾

Python 2.2：第一次提出了生成器（最初称之为迭代器）的概念（PEP 255）。
Python 2.5：引入了将对象发送回暂停了的生成器这一特性即生成器的send()方法（PEP 342）。
Python 3.3：添加了yield from特性，允许从迭代器中返回任何值（注意生成器本身也是迭代器），这样我们就可以串联生成器并且重构出更好的生成器。
Python 3.4：引入asyncio.coroutine装饰器用来标记作为协程的函数，协程函数和asyncio及其事件循环一起使用，来实现异步I/O操作。
Python 3.5：引入了async和await，可以使用async def来定义一个协程函数，这个函数中不能包含任何形式的yield语句，但是可以使用return或await从协程中返回值。
协程实现了协作式并发，通过提高CPU的利用率来达到改善性能的目的。著名的三方库aiohttp就是通过协程的方式实现了HTTP客户端和HTTP服务器的功能，较之requests有更好的获取数据的性能，有兴趣可以阅读它的官方文档。

import asyncio

import aiohttp

async def download(url):

print('Fetch:', url)

async with aiohttp.ClientSession() as session:

async with session.get(url, ssl=False) as resp:

print(url, '--->', resp.status)

print(url, '--->', resp.headers)

print('\n\n', await resp.text())

def main():

loop = asyncio.get_event_loop()

urls = [

'https://www.baidu.com',

'http://www.sohu.com/',

'http://www.sina.com.cn/',

'https://www.taobao.com/',

'http://jd.com/'

]

tasks = [download(url) for url in urls]

loop.run_until_complete(asyncio.wait(tasks))

loop.close()

if __name__ == '__main__':

main()

实例 – 多线程爬取“手机搜狐网”所有页面

下面我们把之间讲的所有知识结合起来，用面向对象的方式实现一个爬取“手机搜狐网”的多线程爬虫。

import pickle
import zlib
from enum import Enum, unique
from hashlib import sha1
from random import random
from threading import Thread, current_thread, local
from time import sleep
from urllib.parse import urlparse

import pymongo
import redis
import requests
from bs4 import BeautifulSoup
from bson import Binary

@unique
class SpiderStatus(Enum):
IDLE = 0
WORKING = 1

def decode_page(page_bytes, charsets=('utf-8',)):
page_html = None
for charset in charsets:
try:
page_html = page_bytes.decode(charset)
break
except UnicodeDecodeError:
pass
return page_html

class Retry(object):

def __init__(self, *, retry_times=3,
wait_secs=5, errors=(Exception, )):
self.retry_times = retry_times
self.wait_secs = wait_secs
self.errors = errors

def __call__(self, fn):

def wrapper(*args, **kwargs):
for _ in range(self.retry_times):
try:
return fn(*args, **kwargs)
except self.errors as e:
print(e)
sleep((random() + 1) * self.wait_secs)
return None

return wrapper

class Spider(object):

def __init__(self):
self.status = SpiderStatus.IDLE

@Retry()
def fetch(self, current_url, *, charsets=('utf-8', ),
user_agent=None, proxies=None):
thread_name = current_thread().name
print(f'[{thread_name}]: {current_url}')
headers = {'user-agent': user_agent} if user_agent else {}
resp = requests.get(current_url,
headers=headers, proxies=proxies)
return decode_page(resp.content, charsets) \
if resp.status_code == 200 else None

def parse(self, html_page, *, domain='m.sohu.com'):
soup = BeautifulSoup(html_page, 'lxml')
for a_tag in soup.body.select('a[href]'):
parser = urlparse(a_tag.attrs['href'])
scheme = parser.scheme or 'http'
netloc = parser.netloc or domain
if scheme != 'javascript' and netloc == domain:
path = parser.path
query = '?' + parser.query if parser.query else ''
full_url = f'{scheme}://{netloc}{path}{query}'
redis_client = thread_local.redis_client
if not redis_client.sismember('visited_urls', full_url):
redis_client.rpush('m_sohu_task', full_url)

def extract(self, html_page):
pass

def store(self, data_dict):
# redis_client = thread_local.redis_client
# mongo_db = thread_local.mongo_db
pass

class SpiderThread(Thread):

def __init__(self, name, spider):
super().__init__(name=name, daemon=True)
self.spider = spider

def run(self):
redis_client = redis.Redis(host='1.2.3.4', port=6379, password='1qaz2wsx')
mongo_client = pymongo.MongoClient(host='1.2.3.4', port=27017)
thread_local.redis_client = redis_client
thread_local.mongo_db = mongo_client.msohu
while True:
current_url = redis_client.lpop('m_sohu_task')
while not current_url:
current_url = redis_client.lpop('m_sohu_task')
self.spider.status = SpiderStatus.WORKING
current_url = current_url.decode('utf-8')
if not redis_client.sismember('visited_urls', current_url):
redis_client.sadd('visited_urls', current_url)
html_page = self.spider.fetch(current_url)
if html_page not in [None, '']:
hasher = hasher_proto.copy()
hasher.update(current_url.encode('utf-8'))
doc_id = hasher.hexdigest()
sohu_data_coll = mongo_client.msohu.webpages
if not sohu_data_coll.find_one({'_id': doc_id}):
sohu_data_coll.insert_one({
'_id': doc_id,
'url': current_url,
'page': Binary(zlib.compress(pickle.dumps(html_page)))
})
self.spider.parse(html_page)
self.spider.status = SpiderStatus.IDLE

def is_any_alive(spider_threads):
return any([spider_thread.spider.status == SpiderStatus.WORKING
for spider_thread in spider_threads])

thread_local = local()
hasher_proto = sha1()

def main():
redis_client = redis.Redis(host='1.2.3.4', port=6379, password='1qaz2wsx')
if not redis_client.exists('m_sohu_task'):
redis_client.rpush('m_sohu_task', 'http://m.sohu.com/')

spider_threads = [SpiderThread('thread-%d' % i, Spider())
for i in range(10)]
for spider_thread in spider_threads:
spider_thread.start()

while redis_client.exists('m_sohu_task') or is_any_alive(spider_threads):
sleep(5)

print('Over!')

if __name__ == '__main__':
main()

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

import pickle

import zlib

from enum import Enum, unique

from hashlib import sha1

from random import random

from threading import Thread, current_thread, local

from time import sleep

from urllib.parse import urlparse

import pymongo

import redis

import requests

from bs4 import BeautifulSoup

from bson import Binary

@unique

class SpiderStatus(Enum):

IDLE = 0

WORKING = 1

def decode_page(page_bytes, charsets=('utf-8',)):

page_html = None

for charset in charsets:

try:

page_html = page_bytes.decode(charset)

break

except UnicodeDecodeError:

pass

return page_html

class Retry(object):

def __init__(self, *, retry_times=3,

wait_secs=5, errors=(Exception, )):

self.retry_times = retry_times

self.wait_secs = wait_secs

self.errors = errors

def __call__(self, fn):

def wrapper(*args, **kwargs):

for _ in range(self.retry_times):

try:

return fn(*args, **kwargs)

except self.errors as e:

print(e)

sleep((random() + 1) * self.wait_secs)

return None

return wrapper

class Spider(object):

def __init__(self):

self.status = SpiderStatus.IDLE

@Retry()

def fetch(self, current_url, *, charsets=('utf-8', ),

user_agent=None, proxies=None):

thread_name = current_thread().name

print(f'[{thread_name}]: {current_url}')

headers = {'user-agent': user_agent} if user_agent else {}

resp = requests.get(current_url,

headers=headers, proxies=proxies)

return decode_page(resp.content, charsets) \

if resp.status_code == 200 else None

def parse(self, html_page, *, domain='m.sohu.com'):

soup = BeautifulSoup(html_page, 'lxml')

for a_tag in soup.body.select('a[href]'):

parser = urlparse(a_tag.attrs['href'])

scheme = parser.scheme or 'http'

netloc = parser.netloc or domain

if scheme != 'javascript' and netloc == domain:

path = parser.path

query = '?' + parser.query if parser.query else ''

full_url = f'{scheme}://{netloc}{path}{query}'

redis_client = thread_local.redis_client

if not redis_client.sismember('visited_urls', full_url):

redis_client.rpush('m_sohu_task', full_url)

def extract(self, html_page):

pass

def store(self, data_dict):

# redis_client = thread_local.redis_client

# mongo_db = thread_local.mongo_db

pass

class SpiderThread(Thread):

def __init__(self, name, spider):

super().__init__(name=name, daemon=True)

self.spider = spider

def run(self):

redis_client = redis.Redis(host='1.2.3.4', port=6379, password='1qaz2wsx')

mongo_client = pymongo.MongoClient(host='1.2.3.4', port=27017)

thread_local.redis_client = redis_client

thread_local.mongo_db = mongo_client.msohu

while True:

current_url = redis_client.lpop('m_sohu_task')

while not current_url:

current_url = redis_client.lpop('m_sohu_task')

self.spider.status = SpiderStatus.WORKING

current_url = current_url.decode('utf-8')

if not redis_client.sismember('visited_urls', current_url):

redis_client.sadd('visited_urls', current_url)

html_page = self.spider.fetch(current_url)

if html_page not in [None, '']:

hasher = hasher_proto.copy()

hasher.update(current_url.encode('utf-8'))

doc_id = hasher.hexdigest()

sohu_data_coll = mongo_client.msohu.webpages

if not sohu_data_coll.find_one({'_id': doc_id}):

sohu_data_coll.insert_one({

'_id': doc_id,

'url': current_url,

'page': Binary(zlib.compress(pickle.dumps(html_page)))

})

self.spider.parse(html_page)

self.spider.status = SpiderStatus.IDLE

def is_any_alive(spider_threads):

return any([spider_thread.spider.status == SpiderStatus.WORKING

for spider_thread in spider_threads])

thread_local = local()

hasher_proto = sha1()

def main():

redis_client = redis.Redis(host='1.2.3.4', port=6379, password='1qaz2wsx')

if not redis_client.exists('m_sohu_task'):

redis_client.rpush('m_sohu_task', 'http://m.sohu.com/')

spider_threads = [SpiderThread('thread-%d' % i, Spider())

for i in range(10)]

for spider_thread in spider_threads:

spider_thread.start()

while redis_client.exists('m_sohu_task') or is_any_alive(spider_threads):

sleep(5)

print('Over!')

if __name__ == '__main__':

main()

本文来自这个系列长期转载Python-100-Days ，本文观点不代表蓝洛水深立场，转载请联系原作者。

Python自学Day64 并发下载

多线程和多进程补充知识点

threading.local类

concurrent.futures模块

分布式进程

协程和异步I/O

协程的概念

历史回顾

相关推荐

发表回复