Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
moyewuji
Python-100-Days
提交
2c2c096a
P
Python-100-Days
项目概览
moyewuji
/
Python-100-Days
与 Fork 源项目一致
从无法访问的项目Fork
通知
1
Star
1
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
P
Python-100-Days
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
2c2c096a
编写于
6月 04, 2018
作者:
骆昊的技术专栏
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
修正了代码中的bug
上级
ccae929b
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
50 addition
and
7 deletion
+50
-7
Day66-75/04.并发下载.md
Day66-75/04.并发下载.md
+38
-1
Day66-75/code/main_redis.py
Day66-75/code/main_redis.py
+12
-6
未找到文件。
Day66-75/04.并发下载.md
浏览文件 @
2c2c096a
...
@@ -2,9 +2,46 @@
...
@@ -2,9 +2,46 @@
### 多线程和多进程回顾
### 多线程和多进程回顾
在前面的
[
《进程和线程》
](
../Day01-15/Day13/进程和线程.md
)
一文中,我们已经对在Python中使用多进程和多线程实现并发编程进行了简明的讲解,在此我们补充几个知识点。
#### threading.local类
### 实例 - 多线程下载“手机搜狐网”所有页面。
使用线程时最不愿意遇到的情况就是多个线程竞争资源,在这种情况下为了保证资源状态的正确性,我们可能需要对资源进行加锁保护的处理,这一方面会导致程序失去并发性,另外如果多个线程竞争多个资源时,还有可能因为加锁方式的不当导致
[
死锁
](
https://zh.wikipedia.org/wiki/%E6%AD%BB%E9%94%81
)
。要解决多个线程竞争资源的问题,其中一个方案就是让每个线程都持有资源的副本(拷贝),这样每个线程可以操作自己所持有的资源,从而规避对资源的竞争。
要实现将资源和持有资源的线程进行绑定的操作,最简单的做法就是使用threading模块的local类,在网络爬虫开发中,就可以使用local类为每个线程绑定一个MySQL数据库连接或Redis客户端对象,这样通过线程可以直接获得这些资源,既解决了资源竞争的问题,又避免了在函数和方法调用时传递这些资源。具体的请参考本章多线程爬取“手机搜狐网”(Redis版)的实例代码。
#### concurrent.futures模块
Python3.2带来了
`concurrent.futures`
模块,这个模块包含了线程池和进程池、管理并行编程任务、处理非确定性的执行流程、进程/线程同步等功能。关于这部分的内容推荐大家阅读
[
《Python并行编程》
](
http://python-parallel-programmning-cookbook.readthedocs.io/zh_CN/latest/index.html
)
。
#### 分布式进程
使用多进程的时候,可以将进程部署在多个主机节点上,Python的
`multiprocessing`
模块不但支持多进程,其中
`managers`
子模块还支持把多进程部署到多个节点上。当然,要部署分布式进程,首先需要一个服务进程作为调度者,进程之间通过网络进行通信来实现对进程的控制和调度,由于
`managers`
模块已经对这些做出了很好的封装,因此在无需了解网络通信细节的前提下,就可以编写分布式多进程应用。具体的请参照本章分布式多进程爬取“手机搜狐网”的实例代码。
### 协程和异步I/O
#### 协程的概念
协程(coroutine)通常又称之为微线程或纤程,它是相互协作的一组子程序(函数)。所谓相互协作指的是在执行函数A时,可以随时中断去执行函数B,然后又中断继续执行函数A。注意,这一过程并不是函数调用(因为没有调用语句),整个过程看似像多线程,然而协程只有一个线程执行。协程通过
`yield`
关键字和
`send()`
操作来转移执行权,协程之间不是调用者与被调用者的关系。
协程的优势在于以下两点:
1.
执行效率极高,因为子程序(函数)切换不是线程切换,由程序自身控制,没有切换线程的开销。
2.
不需要多线程的锁机制,因为只有一个线程,也不存在竞争资源的问题,当然也就不需要对资源加锁保护,因此执行效率高很多。
> 说明:协程适合处理的是I/O密集型任务,处理CPU密集型任务并不是它的长处,如果要提升CPU的利用率可以考虑“多进程+协程”的模式。
#### 历史回顾
1.
Python 2.2:第一次提出了生成器(最初称之为迭代器)的概念(PEP 255)。
2.
Python 2.5:引入了将对象发送回暂停了的生成器这一特性即生成器的
`send()`
方法(PEP 342)。
3.
Python 3.3:添加了
`yield from`
特性,允许从迭代器中返回任何值(注意生成器本身也是迭代器),这样我们就可以串联生成器并且重构出更好的生成器。
4.
Python 3.4:引入
`asyncio.coroutine`
装饰器用来标记作为协程的函数,协程函数和
`asyncio`
及其事件循环一起使用,来实现异步I/O操作。
5.
Python 3.5:引入了
`async`
和
`await`
,可以使用
`async def`
来定义一个协程函数,这个函数中不能包含任何形式的
`yield`
语句,但是可以使用
`return`
或
`await`
从协程中返回值。
### 实例 - 多线程爬取“手机搜狐网”所有页面。
```
Python
```
Python
...
...
Day66-75/code/main_redis.py
浏览文件 @
2c2c096a
...
@@ -3,7 +3,7 @@ import zlib
...
@@ -3,7 +3,7 @@ import zlib
from
enum
import
Enum
,
unique
from
enum
import
Enum
,
unique
from
hashlib
import
sha1
from
hashlib
import
sha1
from
random
import
random
from
random
import
random
from
threading
import
Thread
,
current_thread
from
threading
import
Thread
,
current_thread
,
local
from
time
import
sleep
from
time
import
sleep
from
urllib.parse
import
urlparse
from
urllib.parse
import
urlparse
...
@@ -79,6 +79,7 @@ class Spider(object):
...
@@ -79,6 +79,7 @@ class Spider(object):
path
=
parser
.
path
path
=
parser
.
path
query
=
'?'
+
parser
.
query
if
parser
.
query
else
''
query
=
'?'
+
parser
.
query
if
parser
.
query
else
''
full_url
=
f
'
{
scheme
}
://
{
netloc
}{
path
}{
query
}
'
full_url
=
f
'
{
scheme
}
://
{
netloc
}{
path
}{
query
}
'
redis_client
=
thread_local
.
redis_client
if
not
redis_client
.
sismember
(
'visited_urls'
,
full_url
):
if
not
redis_client
.
sismember
(
'visited_urls'
,
full_url
):
redis_client
.
rpush
(
'm_sohu_task'
,
full_url
)
redis_client
.
rpush
(
'm_sohu_task'
,
full_url
)
...
@@ -86,6 +87,8 @@ class Spider(object):
...
@@ -86,6 +87,8 @@ class Spider(object):
pass
pass
def
store
(
self
,
data_dict
):
def
store
(
self
,
data_dict
):
# redis_client = thread_local.redis_client
# mongo_db = thread_local.mongo_db
pass
pass
...
@@ -96,6 +99,10 @@ class SpiderThread(Thread):
...
@@ -96,6 +99,10 @@ class SpiderThread(Thread):
self
.
spider
=
spider
self
.
spider
=
spider
def
run
(
self
):
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
:
while
True
:
current_url
=
redis_client
.
lpop
(
'm_sohu_task'
)
current_url
=
redis_client
.
lpop
(
'm_sohu_task'
)
while
not
current_url
:
while
not
current_url
:
...
@@ -109,6 +116,7 @@ class SpiderThread(Thread):
...
@@ -109,6 +116,7 @@ class SpiderThread(Thread):
hasher
=
hasher_proto
.
copy
()
hasher
=
hasher_proto
.
copy
()
hasher
.
update
(
current_url
.
encode
(
'utf-8'
))
hasher
.
update
(
current_url
.
encode
(
'utf-8'
))
doc_id
=
hasher
.
hexdigest
()
doc_id
=
hasher
.
hexdigest
()
sohu_data_coll
=
mongo_client
.
msohu
.
webpages
if
not
sohu_data_coll
.
find_one
({
'_id'
:
doc_id
}):
if
not
sohu_data_coll
.
find_one
({
'_id'
:
doc_id
}):
sohu_data_coll
.
insert_one
({
sohu_data_coll
.
insert_one
({
'_id'
:
doc_id
,
'_id'
:
doc_id
,
...
@@ -124,17 +132,15 @@ def is_any_alive(spider_threads):
...
@@ -124,17 +132,15 @@ def is_any_alive(spider_threads):
for
spider_thread
in
spider_threads
])
for
spider_thread
in
spider_threads
])
redis_client
=
redis
.
Redis
(
host
=
'1.2.3.4'
,
thread_local
=
local
()
port
=
6379
,
password
=
'1qaz2wsx'
)
mongo_client
=
pymongo
.
MongoClient
(
host
=
'120.77.222.217'
,
port
=
27017
)
db
=
mongo_client
.
msohu
sohu_data_coll
=
db
.
webpages
hasher_proto
=
sha1
()
hasher_proto
=
sha1
()
def
main
():
def
main
():
redis_client
=
redis
.
Redis
(
host
=
'1.2.3.4'
,
port
=
6379
,
password
=
'1qaz2wsx'
)
if
not
redis_client
.
exists
(
'm_sohu_task'
):
if
not
redis_client
.
exists
(
'm_sohu_task'
):
redis_client
.
rpush
(
'm_sohu_task'
,
'http://m.sohu.com/'
)
redis_client
.
rpush
(
'm_sohu_task'
,
'http://m.sohu.com/'
)
spider_threads
=
[
SpiderThread
(
'thread-%d'
%
i
,
Spider
())
spider_threads
=
[
SpiderThread
(
'thread-%d'
%
i
,
Spider
())
for
i
in
range
(
10
)]
for
i
in
range
(
10
)]
for
spider_thread
in
spider_threads
:
for
spider_thread
in
spider_threads
:
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录