python 线程池及进度条

上一篇去爬壁纸的文章用的是每下载都使用一个新的线程去下，这种方式比较简单，但用起来好像不是很高效，而且进度显示也是很混乱，于是就想着使用线程池去管理下载任务，并使用更加好看的进度条显示进度。

先上一个效果图：

线程池

python我没找到有自带的线程池实现（类似于Java那种），但自己实现一个简单的其实也不难。线程池的说得简单点就是一堆线程集合，不过线程是可复用的，避免了重复开启线程的开销。

线程池主要包含几个部分

  • 线程管理* 工作线程
  • 任务队列

线程管理

基于此，首先得创建线程管理器。线程管理我把它定义成是一个创建工作线程以及分配任务到具体一个工作线程的过程，先上完整代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
__DEFAULT_NUM__=5

class WorkPool:
    def __init__(self,worker_num=__DEFAULT_NUM__):
        self.workers=[]

        for i in range(0,worker_num):
            self.workers.append(Worker(i))

    def selectWorker(self): 
        worker = None
        for w in self.workers:
            if worker == None or w.task_num() < worker.task_num():
                worker=w

        return worker

    def add_task(self,func,*args):
        availableWorker = self.selectWorker()
        if availableWorker:
            availableWorker.add_task(func,args)

    def wait(self):
        for worker in self.workers:
            worker.join()

Worker是具体的工作线程定义（见下），在WorkPool初始化时就将所有的工作线程都创建出来，比较简单粗暴，但也够用；add_task是对外添加任务用的，里面实现有一个调度过程，为了各个工作线程工作平均，selectWorker会根据当前各个工作线程的任务数据来决定当前新任务将被交给哪个工作线程去完成。

1
add_task(self,func,*args)

add_task接受一个函数指针和这个函数的参数作用参数，args是一个tuple，类似于Java中的

1
add_task(Object...args)

但tuple是不能修改的。

选择好哪个工作线程后，就将函数与参数都放到去工作线程对应的工作队列中。

wait函数是用于判断结束的，当所有的工作线程都退出时，调用方应该知道。

工作线程

工作线程是算是核心，也是先上代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class Worker(threading.Thread):
    def __init__(self,id):
        threading.Thread.__init__(self)
        self.id=id
        self.work_queue=Queue.Queue()
        self.size=0
        self.running=False

    def getId(self):
        return self.id

    def add_task(self,func,args):#由于这个args是从外边*args传进来的，不需要再取地址,否则多个参数会被打包成一个对象
        self.work_queue.put((func,args))
        self.size+=1

        if not self.running:
            self.start()

    def task_num(self):
        return self.size

    def isRunning(self):
        return self.running

    def stop(self):
        self.running=False

    def run(self):
        self.running=True
        while self.running:
            #self.bind_data[self.id].value+=1

            try:
                func,args=self.work_queue.get(block=False)
                func(*args)#重新取值，避免直接函数调用时参数个数匹配不上
                self.size-=1
                self.work_queue.task_done()
            except Exception,e:
                break

在构造函数里先创建一个工作队列（Queue），并为每一个工作线程分派一个id，每一个工作线程本质上都是一个Thread。

add_task接受函数与参数，并将其丢到工作队列中，并将func与args打包成一个tuple统一丢到工作队列中就好了。

当首次接收到工作任务时，工作线程开始工作 ，run()函数就开始执行了。因为工作线程需要不断从工作队列中获取任务并执行，所以需要一个循环。

1
self.work_queue.get(block=False)

这一句从工作队列中获取任务，并不阻塞，当获取不到时就会抛出异常，由于任务是开始时一次过添加进去的，当获取不到任务时抛异常退出后就会认为工作线程工作结束了。

当从工作线程中获取到函数与参数后，就可以直接调用执行了

1
func(*args)

进度条

由于在终端显示进度条比较麻烦，上一篇里使用的要sys.stdout.write并且不换行来达到刷新的进度的效果，但毕竟还是不太好看。这次我使用了progressive来实现。

从progressive提供的example里可以看出，它是使用字典和arDescriptor来创建一个或多个进度条的。

看示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
test_d = {
    "Warp Jump": {
        "1) Prepare fuel": {
            "Load Tanks": {
                "Tank 1": BarDescriptor(value=leaf_values[0], **bd_defaults),
                "Tank 2": BarDescriptor(value=leaf_values[1], **bd_defaults),
            },
            "Refine tylium ore": BarDescriptor(
                value=leaf_values[2], **bd_defaults
            ),
        },
        "2) Calculate jump co-ordinates": {
            "Resolve common name to co-ordinates": {
                "Querying resolution from baseship": BarDescriptor(
                    value=leaf_values[3], **bd_defaults
                ),
            },
        },
        "3) Perform jump": {
            "Check FTL drive readiness": BarDescriptor(
                value=leaf_values[4], **bd_defaults
            ),
            "Juuuuuump!": BarDescriptor(value=leaf_values[5],
                                        **bd_defaults)
        }
    }
}

最终出来的效果是

其使用的是leaf_values表示的是对应进度条的进度，是一个Value对象，当修改此值是就会反映到进度条上，不过需要不断地刷新。

1
2
3
4
5
6
7
  # Create blessings.Terminal instance
t = Terminal()
# Initialize a ProgressTree instance
n = ProgressTree(term=t)
# We'll use the make_room method to make sure the terminal
#   is filled out with all the room we need
n.make_room(test_d)

创建一个ProgressTree用将字典数据构建出具有层级关系的进度条。但如果需要将进度条画在终端上或者进行刷新，则需要调用

1
2
3
4
5
    # After the cursor position is first saved (in the first draw call)
#   this will restore the cursor back to the top so we can draw again
n.cursor.restore()
# Actually draw out the bars
n.draw(test_d, BarDescriptor(bd_defaults))

理解了progressTree的使用方法后，就可以将其应用到线程池上了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
    class TaskGraphics:
def __init__(self,data,num=__DEFAULT_NUM__):
    self.bind_data=[Value(0) for i in range(num)] #创建出进度数值数组
    self.terminal=False
    self.bd_defaults=dict(type=Bar,kwargs=dict(max_value=100))#进度条样式 
    self.graph_data={}

    for i in range(0,num):
        task_name='task_%d'%i
        self.graph_data[task_name]=BarDescriptor(value=self.bind_data[i],**self.bd_defaults)

    self.__init_tree__()
    threading.Thread(target=self.invalidate,args=()).start()

def __init_tree__(self):
    t=Terminal()
    self.n=ProgressTree(term=t)
    self.n.make_room(self.graph_data)

#更新进度
def updateValue(self,id,value):
    self.bind_data[id].value=value

#不断地刷新进度
def invalidate(self):
    while not self.terminal:
        time.sleep(0.1)
        self.n.cursor.restore()
        self.n.draw(self.graph_data,BarDescriptor(self.bd_defaults))

def stop(self):
    self.terminal=True

结合上面的注释，大概也能看出运行原理了。另外，由于刷新进度需要不断地循环，所以不能放在主线程中，需要另开线程去刷新，否则其他工作就没法做了。

带进度条的线程池

结合上面两节，完整代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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
#!/usr/bin/python
# -*- coding:utf-8 -*-

import Queue
import threading
import time
import traceback

import random
from time import sleep

from blessings import Terminal

from progressive.bar import Bar
from progressive.tree import ProgressTree, Value, BarDescriptor

__DEFAULT_NUM__=5

class WorkPool:
    def __init__(self,worker_num=__DEFAULT_NUM__):
        self.workers=[]
        self.leaf_values=[Value(0) for i in range(worker_num)]

                #创建进度条显示对象
        self.taskGraphics=TaskGraphics(self.leaf_values,worker_num)

        for i in range(0,worker_num):
            self.workers.append(Worker(i,self.taskGraphics))

    def selectWorker(self): 
        worker = None
        for w in self.workers:
            if worker == None or w.task_num() < worker.task_num():
                worker=w

        return worker

    def add_task(self,func,*args):
        availableWorker = self.selectWorker()
        if availableWorker:
            availableWorker.add_task(func,args)

    def wait(self):
        for worker in self.workers:
            worker.join()

                #当所有任务都完成后，应该停止刷新进度显示
        self.taskGraphics.stop()

class Worker(threading.Thread):
    def __init__(self,id,graphics):
        threading.Thread.__init__(self)
        self.id=id
        self.work_queue=Queue.Queue()
        self.size=0
        self.running=False
        self.graphics=graphics

    def getId(self):
        return self.id

    def add_task(self,func,args):#由于这个args是从外边*args传进来的，不需要再取地址,否则多个参数会被打包成一个对象
        self.work_queue.put((func,args))
        self.size+=1

        if not self.running:
            self.start()

    def task_num(self):
        return self.size

    def isRunning(self):
        return self.running

    def stop(self):
        self.running=False

    def run(self):
        self.running=True
        while self.running:
            #self.bind_data[self.id].value+=1

            try:
                func,args_ori=self.work_queue.get(block=False)

                                #添加进度反馈
                args_callback=(self.update,)
                args=args_ori+args_callback

                func(*args)#重新取值，避免直接函数调用时参数个数匹配不上
                self.size-=1
                self.work_queue.task_done()
            except Exception,e:
                traceback.print_exc()
                break

    def update(self,fileName,percent):
        self.graphics.updateTask(self.id,'task_%d: %s'%(self.id,fileName))
        self.graphics.updateValue(self.id,percent)

class TaskGraphics:
    def __init__(self,data,num=__DEFAULT_NUM__):
        self.bind_data=[Value(0) for i in range(num)]
        #self.bind_data=data
        self.terminal=False
        self.bd_defaults=dict(type=Bar,kwargs=dict(max_value=100))
        self.indicator={}
        self.graph_data={}

        for i in range(0,num):
            task_name='task_%d'%i
            self.indicator[i]=task_name
            self.graph_data[task_name]=BarDescriptor(value=self.bind_data[i],**self.bd_defaults)

        self.__init_tree__()
        threading.Thread(target=self.invalidate,args=()).start()

    def __init_tree__(self):
        t=Terminal()
        self.n=ProgressTree(term=t)
        self.n.make_room(self.graph_data)

    def updateTask(self,id,task_name):
        old_task_name=self.indicator[id]
        if task_name == old_task_name:
            return

        self.graph_data[task_name]=self.graph_data[old_task_name]
        self.indicator[id]=task_name
        del self.graph_data[old_task_name]

    def updateValue(self,id,value):
        self.bind_data[id].value=value

    def invalidate(self):
        while not self.terminal:
            time.sleep(0.1)
            self.n.cursor.restore()
            self.n.draw(self.graph_data,BarDescriptor(self.bd_defaults))

    def stop(self):
        self.terminal=True

def test(i,j):
    #print threading.current_thread(),i
    #print i+j
    time.sleep(0.1)

if __name__ == '__main__':
    workPool=WorkPool(10)
    for k in range(1,100):
        workPool.add_task(test,k,k+1)

    workPool.wait()

需要讲几个点

1
2
3
4
5
6
7
8
9
10
11
12
13
14
                    ……

            func,args_ori=self.work_queue.get(block=False)

                            #添加进度反馈
            args_callback=(self.update,)#后面的“，”不能少
            args=args_ori+args_callback

            func(*args)#重新取值，避免直接函数调用时参数个数匹配不上
                ……

def update(self,fileName,percent):
    self.graphics.updateTask(self.id,'task_%d: %s'%(self.id,fileName))
    self.graphics.updateValue(self.id,percent)

由于具体的进度需要在具体的任务中才能获取，比方说下载进度，下载进度的计算需要由具体任务去算，作用回调到工作线程中去更新进度条，但因为任务加进线程池时是不可能知道线程池里的回调函数的，所以在真正调用任务函数时，需要添加一个新的回调函数，由于参数本来是tuple，所以需要重新创建一个新的tuple作用参数传给任务函数，注意args_callback=(self.update,)里括号的逗号不能少，否则连接出来的结果不对。当然，具体的任务函数的参数里也要预留一个这样的回调，如

1
def download(self,dirName,fileName,url,rateHook=None):

rateHook就是回调的函数指针。

另外，由于progressive里的字典参数需要变换名字时，作为字典的key不能直接删除，且由于进度条是使用key来作用进度条的名字的，我需要动态地修改这个，但也不能使用index来定位具体是哪个，所以需要一个 indicator={} 来定义key是在什么位置，以便之后需要换key名字的时候可以直接根据id来定位。另外，由于这是一个字典，里面是按key来排序的，我不能因为换了key后就把进度条的顺序给乱了，所以需要给每个进度条都安排一个统一前缀的key,如“task_1,task_2”。

最后，再结合上一篇的爬壁纸操作，就能实现文章开头的效果了，task的名字是正在下载的文件名：

结合updateTask 和 updateValue 就可以实现进度条的更新和下载名的更新了。

源码地址

依赖

No module named progressive.bar
pip install progressive

No module named blessings
pip install blessings

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

mail:kofmax28@gmail.com