Python 多线程教程：并发与并行

阅读量：5349 次

发布时间：2019-06-15

本文共 10795 字，大约阅读时间需要 35 分钟。

在批评Python的讨论中，常常说起Python多线程是多么的难用。还有人对 global interpreter lock(也被亲切的称为“GIL”)指指点点，说它阻碍了Python的多线程程序同时运行。因此，如果你是从其他语言（比如C++或Java）转过来的话，Python线程模块并不会像你想象的那样去运行。必须要说明的是，我们还是可以用Python写出能并发或并行的代码，并且能带来性能的显著提升，只要你能顾及到一些事情。如果你还没看过的话，我建议你看看Eqbal Quran的文章《Ruby中的并发和并行》。

在本文中，我们将会写一个小的Python脚本，用于下载Imgur上最热门的图片。我们将会从一个按顺序下载图片的版本开始做起，即一个一个地下载。在那之前，你得注册一个Imgur上的应用。如果你还没有Imgur账户，请先注册一个。

本文中的脚本在Python3.4.2中测试通过。稍微改一下，应该也能在Python2中运行——urllib是两个版本中区别最大的部分。

快速使用Romanysoft LAB
的技术实现 HTML 开发Mac OS App，并销售到苹果应用商店中。

《HTML开发Mac OS App 视频教程》

土豆网同步更新：

百度网盘同步：

分享 [中文纪录片]互联网时代

官方QQ群：（申请加入，说是我推荐的
）

App实践出真知 434558944

App学习交流 452180823

1、开始动手

让我们从创建一个叫“download.py”的Python模块开始。这个文件包含了获取图片列表以及下载这些图片所需的所有函数。我们将这些功能分成三个单独的函数：

get_links

download_link

setup_download_dir

第三个函数，“setup_download_dir”，用于创建下载的目标目录（如果不存在的话）。

Imgur的API要求HTTP请求能支持带有client ID的“Authorization”头部。你可以从你注册的Imgur应用的面板上找到这个client ID，而响应会以JSON进行编码。我们可以使用Python的标准JSON库去解码。下载图片更简单，你只需要根据它们的URL获取图片，然后写入到一个文件即可。

代码如下：

                         
import
json
import
logging
import
os
from
pathlib 
import
Path
from
urllib.request 
import
urlopen, Request
 
logger 
=
logging.getLogger(__name__)
 
def
get_links(client_id):
   
headers 
=
{ 
'Authorization'
: 
'Client-ID {}'
.
format
(client_id)}
   
req 
=
Request(
'https://api.imgur.com/3/gallery/'
, headers
=
headers, method
=
'GET'
)
   
with urlopen(req) as resp:
       
data 
=
json.loads(resp.readall().decode(
'utf-8'
))
   
return
map
(
lambda
item: item[
'link'
], data[
'data'
])
 
def
download_link(directory, link):
   
logger.info(
'Downloading %s'
, link)
   
download_path 
=
directory 
/
os.path.basename(link)
   
with urlopen(link) as image, download_path.
open
(
'wb'
) as f:
       
f.write(image.readall())
 
def
setup_download_dir():
   
download_dir 
=
Path(
'images'
)
   
if
not
download_dir.exists():
       
download_dir.mkdir()
   
return
download_dir
        
      

接下来，你需要写一个模块，利用这些函数去逐个下载图片。我们给它命名为“single.py”。它包含了我们最原始版本的Imgur图片下载器的主要函数。这个模块将会通过环境变量“IMGUR_CLIENT_ID”去获取Imgur的client ID。它将会调用“setup_download_dir”去创建下载目录。最后，使用get_links函数去获取图片的列表，过滤掉所有的GIF和专辑URL，然后用“download_link”去将图片下载并保存在磁盘中。下面是“single.py”的代码：

                         
import
logging
import
os
from
time 
import
time
 
from
download 
import
setup_download_dir, get_links, download_link
 
logging.basicConfig(level
=
logging.DEBUG, 
format
=
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logging.getLogger(
'requests'
).setLevel(logging.CRITICAL)
logger 
=
logging.getLogger(__name__)
 
def
main():
   
ts 
=
time()
   
client_id 
=
os.getenv(
'IMGUR_CLIENT_ID'
)
   
if
not
client_id:
       
raise
Exception(
"Couldn't find IMGUR_CLIENT_ID environment variable!"
)
   
download_dir 
=
setup_download_dir()
   
links 
=
[l 
for
l 
in
get_links(client_id) 
if
l.endswith(
'.jpg'
)]
   
for
link 
in
links:
       
download_link(download_dir, link)
   
print
(
'Took {}s'
.
format
(time() 
-
ts))
 
if
__name__ 
=
=
'__main__'
:
   
main()
        
      

注：为了测试方便，上面两段代码可以用如下代码替代演示：

                         
# coding=utf-8
#测试utf-8编码
from
time 
import
sleep, time
import
sys, threading
reload
(sys)
sys.setdefaultencoding(
'utf-8'
)
def
getNums(N):
    
return
xrange
(N)
def
processNum(num):
    
num_add 
=
num 
+
1
    
sleep(
1
)
    
print
str
(threading.current_thread()) 
+
": "
+
str
(num) 
+
" → "
+
str
(num_add)
if
__name__ 
=
=
"__main__"
:
    
t1 
=
time()
    
for
i 
in
getNums(
3
):
        
processNum(i)
    
print
"cost time is: {:.2f}s"
.
format
(time() 
-
t1)
        
      

结果：

                         
<_MainThread(MainThread, started 4436)>: 0 → 1
<_MainThread(MainThread, started 4436)>: 1 → 2
<_MainThread(MainThread, started 4436)>: 2 → 3
cost 
time
is: 3.00s
        
      

在我的笔记本上，这个脚本花了19.4秒去下载91张图片。请注意这些数字在不同的网络上也会有所不同。19.4秒并不是非常的长，但是如果我们要下载更多的图片怎么办呢？或许是900张而不是90张。平均下载一张图片要0.2秒，900张的话大概需要3分钟。那么9000张图片将会花掉30分钟。好消息是使用了并发或者并行后，我们可以将这个速度显著地提高。

接下来的代码示例将只会显示导入特有模块和新模块的import语句。所有相关的Python脚本都可以在这方便地找到。

2、使用线程

线程是最出名的实现并发和并行的方式之一。操作系统一般提供了线程的特性。线程比进程要小，而且共享同一块内存空间。

在这里，我们将写一个替代“single.py”的新模块。它将创建一个有八个线程的池，加上主线程的话总共就是九个线程。之所以是八个线程，是因为我的电脑有8个CPU内核，而一个工作线程对应一个内核看起来还不错。在实践中，线程的数量是仔细考究的，需要考虑到其他的因素，比如在同一台机器上跑的的其他应用和服务。

下面的脚本几乎跟之前的一样，除了我们现在有个新的类，DownloadWorker，一个Thread类的子类。运行无限循环的run方法已经被重写。在每次迭代时，它调用“self.queue.get()”试图从一个线程安全的队列里获取一个URL。它将会一直堵塞，直到队列中出现一个要处理元素。一旦工作线程从队列中得到一个元素，它将会调用之前脚本中用来下载图片到目录中所用到的“download_link”方法。下载完成之后，工作线程向队列发送任务完成的信号。这非常重要，因为队列一直在跟踪队列中的任务数。如果工作线程没有发出任务完成的信号，“queue.join()”的调用将会令整个主线程都在阻塞状态。

                         
from
queue 
import
Queue
from
threading 
import
Thread
 
class
DownloadWorker(Thread):
   
def
__init__(
self
, queue):
       
Thread.__init__(
self
)
       
self
.queue 
=
queue
 
   
def
run(
self
):
       
while
True
:
           
# Get the work from the queue and expand the tuple
           
# 从队列中获取任务并扩展tuple
           
directory, link 
=
self
.queue.get()
           
download_link(directory, link)
           
self
.queue.task_done()
 
def
main():
   
ts 
=
time()
   
client_id 
=
os.getenv(
'IMGUR_CLIENT_ID'
)
   
if
not
client_id:
       
raise
Exception(
"Couldn't find IMGUR_CLIENT_ID environment variable!"
)
   
download_dir 
=
setup_download_dir()
   
links 
=
[l 
for
l 
in
get_links(client_id) 
if
l.endswith(
'.jpg'
)]
   
# Create a queue to communicate with the worker threads
   
queue 
=
Queue()
   
# Create 8 worker threads
   
# 创建八个工作线程
   
for
x 
in
range
(
8
):
       
worker 
=
DownloadWorker(queue)
       
# Setting daemon to True will let the main thread exit even though the workers are blocking
       
# 将daemon设置为True将会使主线程退出，即使worker都阻塞了
       
worker.daemon 
=
True
       
worker.start()
   
# Put the tasks into the queue as a tuple
   
# 将任务以tuple的形式放入队列中
   
for
link 
in
links:
       
logger.info(
'Queueing {}'
.
format
(link))
       
queue.put((download_dir, link))
   
# Causes the main thread to wait for the queue to finish processing all the tasks
   
# 让主线程等待队列完成所有的任务
   
queue.join()
   
print
(
'Took {}'
.
format
(time() 
-
ts))
        
      

注：为了测试方便，上面的代码可以用如下代码替代演示：

                         
# coding=utf-8
#测试utf-8编码
from
Queue 
import
Queue
from
threading 
import
Thread
from
single 
import
*
import
sys
reload
(sys)
sys.setdefaultencoding(
'utf-8'
)
class
ProcessWorker(Thread):
    
def
__init__(
self
, queue):
        
Thread.__init__(
self
)
        
self
.queue 
=
queue
    
def
run(
self
):
        
while
True
:
            
# Get the work from the queue
            
num 
=
self
.queue.get()
            
processNum(num)
            
self
.queue.task_done()
def
main():
    
ts 
=
time()
    
nums 
=
getNums(
4
)
    
# Create a queue to communicate with the worker threads
    
queue 
=
Queue()
    
# Create 4 worker threads
    
# 创建四个工作线程
    
for
x 
in
range
(
4
):
        
worker 
=
ProcessWorker(queue)
        
# Setting daemon to True will let the main thread exit even though the workers are blocking
        
# 将daemon设置为True将会使主线程退出，即使worker都阻塞了
        
worker.daemon 
=
True
        
worker.start()
    
# Put the tasks into the queue
    
for
num 
in
nums:
        
queue.put(num)
    
# Causes the main thread to wait for the queue to finish processing all the tasks
    
# 让主线程等待队列完成所有的任务
    
queue.join()
    
print
(
"cost time is: {:.2f}s"
.
format
(time() 
-
ts))
if
__name__ 
=
=
"__main__"
:
    
main()
        
      

结果：

                         
<ProcessWorker(Thread-4, started daemon 3900)>: 3 → 4<ProcessWorker(Thread-1, started daemon 3436)>: 2 → 3<ProcessWorker(Thread-3, started daemon 4576)>: 1 → 2
 
<ProcessWorker(Thread-2, started daemon 396)>: 0 → 1
cost 
time
is: 1.01s

在同一个机器上运行这个脚本，下载时间变成了4.1秒！即比之前的例子快4.7倍。虽然这快了很多，但还是要提一下，由于GIL的缘故，在这个进程中同一时间只有一个线程在运行。因此，这段代码是并发的但不是并行的。而它仍然变快的原因是这是一个IO密集型的任务。进程下载图片时根本毫不费力，而主要的时间都花在了等待网络上。这就是为什么线程可以提供很大的速度提升。每当线程中的一个准备工作时，进程可以不断转换线程。使用Python或其他有GIL的解释型语言中的线程模块实际上会降低性能。如果你的代码执行的是CPU密集型的任务，例如解压gzip文件，使用线程模块将会导致执行时间变长。对于CPU密集型任务和真正的并行执行，我们可以使用多进程（multiprocessing）模块。

官方的Python实现——CPython——带有GIL，但不是所有的Python实现都是这样的。比如，IronPython，使用.NET框架实现的Python就没有GIL，基于Java实现的Jython也同样没有。你可以查看现有的Python实现。

3、生成多进程

多进程模块比线程模块更易使用，因为我们不需要像线程示例那样新增一个类。我们唯一需要做的改变在主函数中。

为了使用多进程，我们得建立一个多进程池。通过它提供的map方法，我们把URL列表传给池，然后8个新进程就会生成，它们将并行地去下载图片。这就是真正的并行，不过这是有代价的。整个脚本的内存将会被拷贝到各个子进程中。在我们的例子中这不算什么，但是在大型程序中它很容易导致严重的问题。

                         
from
functools 
import
partial
from
multiprocessing.pool 
import
Pool
 
def
main():
   
ts 
=
time()
   
client_id 
=
os.getenv(
'IMGUR_CLIENT_ID'
)
   
if
not
client_id:
       
raise
Exception(
"Couldn't find IMGUR_CLIENT_ID environment variable!"
)
   
download_dir 
=
setup_download_dir()
   
links 
=
[l 
for
l 
in
get_links(client_id) 
if
l.endswith(
'.jpg'
)]
   
download 
=
partial(download_link, download_dir)
   
with Pool(
8
) as p:
       
p.
map
(download, links)
   
print
(
'Took {}s'
.
format
(time() 
-
ts))
        
      

注：为了测试方便，上面的代码可以用如下代码替代演示：

                         
# coding=utf-8
#测试utf-8编码
from
functools 
import
partial
from
multiprocessing.pool 
import
Pool
from
single 
import
*
from
time 
import
time
import
sys
reload
(sys)
sys.setdefaultencoding(
'utf-8'
)
def
main():
    
ts 
=
time()
    
nums 
=
getNums(
4
)
    
p 
=
Pool(
4
)
    
p.
map
(processNum, nums)
    
print
(
"cost time is: {:.2f}s"
.
format
(time() 
-
ts))
if
__name__ 
=
=
"__main__"
:
    
main()
        
      

结果：

                         
<_MainThread(MainThread, started 6188)>: 0 → 1
<_MainThread(MainThread, started 3584)>: 1 → 2
<_MainThread(MainThread, started 2572)>: 3 → 4<_MainThread(MainThread, started 4692)>: 2 → 3
cost 
time
is: 1.21s
        
      

4、分布式任务

你已经知道了线程和多进程模块可以给你自己的电脑跑脚本时提供很大的帮助，那么在你想要在不同的机器上执行任务，或者在你需要扩大规模而超过一台机器的的能力范围时，你该怎么办呢？一个很好的使用案例是网络应用的长时间后台任务。如果你有一些很耗时的任务，你不会希望在同一台机器上占用一些其他的应用代码所需要的子进程或线程。这将会使你的应用的性能下降，影响到你的用户们。如果能在另外一台甚至很多台其他的机器上跑这些任务就好了。

Python库非常适用于这类任务。它是一个简单却很强大的库。首先将一个函数和它的参数放入队列中。它将函数调用的表示，然后将这些表示添加到一个列表中。任务进入队列只是第一步，什么都还没有做。我们至少还需要一个能去监听任务队列的worker（工作线程）。

第一步是在你的电脑上安装和使用Redis服务器，或是拥有一台能正常的使用的Redis服务器的使用权。接着，对于现有的代码只需要一些小小的改动。先创建一个RQ队列的实例并通过传给一台Redis服务器。然后，我们执行“q.enqueue(download_link, download_dir, link)”，而不只是调用“download_link” 。enqueue方法的第一个参数是一个函数，当任务真正执行时，其他的参数或关键字参数将会传给该函数。

最后一步是启动一些worker。RQ提供了方便的脚本，可以在默认队列上运行起worker。只要在终端窗口中执行“rqworker”，就可以开始监听默认队列了。请确认你当前的工作目录与脚本所在的是同一个。如果你想监听别的队列，你可以执行“rqworker queue_name”，然后将会开始执行名为queue_name的队列。RQ的一个很好的点就是，只要你可以连接到Redis，你就可以在任意数量上的机器上跑起任意数量的worker；因此，它可以让你的应用扩展性得到提升。下面是RQ版本的代码：

                         
from
redis 
import
Redis
from
rq 
import
Queue
 
def
main():
   
client_id 
=
os.getenv(
'IMGUR_CLIENT_ID'
)
   
if
not
client_id:
       
raise
Exception(
"Couldn't find IMGUR_CLIENT_ID environment variable!"
)
   
download_dir 
=
setup_download_dir()
   
links 
=
[l 
for
l 
in
get_links(client_id) 
if
l.endswith(
'.jpg'
)]
   
q 
=
Queue(connection
=
Redis(host
=
'localhost'
, port
=
6379
))
   
for
link 
in
links:
       
q.enqueue(download_link, download_dir, link)
        
      