python多线程学习二

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

本文希望达到的目标:

多线程同步原语：互斥锁
多线程队列queue
线程池threadpool

一、多线程同步原语：互斥锁

在多线程代码中，总有一些特定的函数或者代码块不应该被多个线程同时执行，通常包括修改数据库，更新文件或者其他会产生竞态的类似情况。当多个线程共享相同内存时，需要确保每个线程看到的数据是一致的，如果线程使用的变量是其他线程都不会去修改或读取的，那就不存在这个问题；或者数据变量只是只读的，那也不会出现数据不一致的问题；但是如果某个线程可以修改变量，其他线程也可以修改或者读取这个变量的时候，就需要对这些线程进行同步，以确保他们访问的变量的存储内容是不会访问到无效的值。

当一个线程修改变量时，其他线程在读取这个变量时就可能会出现不一致的数据。在变量修改时间多于一个存储器访问周期的处理器结构中，当存储器读跟存储器写这两个周期相交的时，这种潜在的不一致就会出现。这时，针对这一变量形成的临界区，应该在只有一个线程可以通过。线程的同步一般使用两种类型的同步原语，一种是互斥锁，一种是信号量。锁是比较简单和比较低级别的机制，而信号量用于多线程竞争有限资源的情况，两种使用场景有点差异，本文只针对互斥锁进行学习。

一个简单常用的demo引入这个概念，全局变量balance ，初始化值为0，创建2个子线程，每个线程都去调用change_it函数，子线程运行完成后打印出balance值应该是0才是正确的。

import time, threading
balance = def change_it(n):
global balance
balance = balance + n
balance = balance - ndef run_thread(n):
for i in range():
change_it(n)t1 = threading.Thread(target=run_thread, args=(,))
t2 = threading.Thread(target=run_thread, args=(,))
t1.start()
t2.start()
t1.join()
t2.join()
print balance

但是实际情况下，多次运行或当增大run_thread的for循环大小时，值会出现非0的情况，这是因为两个线程是有可能同时访问change_it函数的，可能会出现：当线程A运行到balance = balance + n时，线程B也在调用这个函数，此时拿到的balance值，线程B还未减去，这样才导致了最终结果后面不是0。

python的threading模块有一个Lock对象，这个锁是原始锁，它在锁定时不属于特定线程。是当前可用的最低级别的同步原语，由线程扩展模块直接实现。原始锁处于两种状态之一：“锁定”或“未锁定”。它是在未锁定状态下创建的。它有两种基本方法，即acquire和release。针对上面的demo，使用原始锁代码如下，在执行change_it函数修改全局变量时，先获取到锁，执行完整个过程后，再释放锁，其他线程才能执行这块临界区的代码。

import time, threadingbalance =
lock =threading.Lock()class MyThread(threading.Thread):
def __init__(self,func,args,name=''):
threading.Thread.__init__(self)
self.name = name
self.func = func
self.args = args def run(self):
self.func(*self.args)def change_it(n):
global balance
balance = balance + n
balance = balance - ndef run_thread(n):
for i in range():
lock.acquire()
try:
change_it(n)
finally:
lock.release()def main():
threads =[]
nloops = for i in range(nloops):
t = MyThread(run_thread,(i,))
threads.append(t) for i in range(nloops):
threads[i].start() for i in range(nloops):
threads[i].join() print balanceif __name__ == '__main__':
main()

总结：

A：如果某个线程可以修改变量，其他线程也可以修改或者读取这个变量的时候，都需要对这些线程进行同步加锁操作，否则是会出现数据不一致的情况。

B：这类情况下需要怎么测试加锁后数据的一致性呢？一个可以加大调用的次数，但是需要把基数调整很大，也只有万分之一必现的可能；还有一个方法，就是确保在写的时候，不能有其他的写和读的操作，此时启动多个线程，需要有日志打印或者打断点到线程

来赋值测试更好的观察结果，比如一个线程正在写时，进入临界区，锁住之后，线程被挂起，其他线程此时是无法读取该资源的，其他线程也无法正常运行下去。

二、线程队列Queue

Queue模块，在多生产者和多消费者模型的场景下特别适用，也特别的适用于多线程场景下线程之间的信息的安全交换，该模块中的队列类实现所有所需的锁定语义。Queue类有好几个不同的：

1、 Queue.Queue(maxsize=0)，是一个FIFO队列，先进队列就先出队列，如果maxsize<=0，表示队列的大小是无限的

2、Queue.LifoQueue，是一个LIFO队列，先进后出队列

3、Queue.PriorityQueue，是一个优先级队列，优先级高的先出。

实际使用用经常是FIFO队列，所以demo也是这个类为例子，包含主要的函数如下：

1、基本方法：

Queue.Queue(maxsize=0) FIFO，如果maxsize小于1就表示队列长度无限
       Queue.LifoQueue(maxsize=0)   LIFO，如果maxsize小于1就表示队列长度无限
       Queue.qsize()   返回队列的大小
       Queue.empty()   如果队列为空，返回True,反之False
       Queue.full()   如果队列满了，返回True,反之False
       Queue.get([block[, timeout]])   读队列，timeout等待时间
       Queue.put(item, [block[, timeout]])   写队列，timeout等待时间
       Queue.queue.clear()   清空队列

2、两个比较重要的函数：

task_done()：意味着之前入队的一个任务已经完成。由队列的消费者线程调用。每一个get()调用得到一个任务，接下来的task_done()调用告诉队列该任务已经处理完毕。如果当前一个join()正在阻塞，它将在队列中的所有任务都处理完时恢复执行（即每一个由put()调用入队的任务都有一个对应的task_done()调用）。

join()：阻塞调用线程，直到队列中的所有任务被处理掉。只要有数据被加入队列，未完成的任务数就会增加。当消费者线程调用task_done()（意味着有消费者取得任务并完成任务），未完成的任务数就会减少。当未完成的任务数降到0，join()解除阻塞。

# encoding: utf-
from Queue import Queue
import time
import threadingqueue = Queue()def do_job():
while True:
i = queue.get()
time.sleep()
print 'index %s' % (i)
queue.task_done()if __name__ == '__main__':
# 创建1个线程的线程池
for i in range():
t = threading.Thread(target=do_job)
t.daemon = True # 设置线程daemon 主线程退出，daemon线程也会推出，即时正在运行
t.start() # 模拟创建线程池3秒后塞进10个任务到队列
time.sleep()
for i in range():
queue.put(i) queue.join()

因为是Queue队列，主线程put数据到queue后，只有一个子线程去get，所以是按照先进先出的顺序读取的

如果有3个子线程去读取，顺序就不是这样的了，但是也不会出现多次取出同一个，或者有数据在queue内未读取的情况，因为queue的内部实现了对queue数据的加锁的代码逻辑，实现了所需的锁的定义。

三、线程池threadpool

传统多线程方案会使用“即时创建，即时销毁”的策略。一个线程的运行时间可以分为3部分：线程的启动时间、线程体的运行时间和线程的销毁时间。在多线程处理的情景中，如果线程不能被重用，就意味着每次创建都需要经过启动、销毁和运行3个过程。如果提交给线程的任务是执行时间较短，但执行次数极其频繁，那么服务器将处于不停的创建线程，销毁线程的状态，这必然会增加系统相应的时间，降低了效率。

线程池预先创建线程放如线程池中，然后把需要处理的任务放入工作请求队列中，在这些任务队列中，线程在运行过程中，获取到该任务并处理完成，可以将结果放入另一个队列中。然后，线程池对象可以在所有线程可用时或在所有线程完成工作之后立即从该队列收集所有线程的结果。然后可以定义回调来处理每个结果。每个处理完当前任务之后并不销毁而是被安排处理下一个任务，所以能够避免多次创建线程，从而节省线程创建和销毁的开销，能带来更好的性能和系统稳定性。

上面的例子，实际也是一个threadpool的引例，简单的工作逻辑如下：

创建任务队列，实例化Queue.Queue类。
生成守护线程池，并把线程设置成了daemon守护线程。
定义线程处理函数do_job，每个线程处理时，都从queue中读取数据，如果没有数据，则无限循环阻塞。
每次完成一次工作后，使用queue.task_done()函数向任务已经完成的队列发送一个信号
主线程设置queue.join()阻塞，直到任务队列已经清空了，解除阻塞，向下执行

这里面已经能说明线程池涉及到的基本类了，Queue用户线程间的通讯，Thread模块用户实现多线程机制，但是对于Queue和Thread之间的关联关系没有提现，对任务的结构没有定义，不方便整个任务的管理。于是就引入了线程池的概念，线程池最初我的理解是

在外面再包装了一层，把多个线程的创建和数据队列的创建统一封装到了线程池内进行调度和工作。参考了下常用的自定义线程池的demo，代码如下：

# !/usr/bin/env python
# -*- coding:utf- -*-import Queue
import threading
import timeclass WorkManager(object):
def __init__(self, work_num=,thread_num=):
self.work_queue = Queue.Queue()
self.threads = []
self.__init_work_queue(work_num)
self.__init_thread_pool(thread_num) """
初始化线程
"""
def __init_thread_pool(self,thread_num):
for i in range(thread_num):
self.threads.append(Work(self.work_queue)) """
初始化工作队列
"""
def __init_work_queue(self, jobs_num):
for i in range(jobs_num):
self.add_job(do_job, i) """
添加一项工作入队
"""
def add_job(self, func, *args):
self.work_queue.put((func, list(args)))#任务入队，Queue内部实现了同步机制 """
等待所有线程运行完毕
"""
def wait_allcomplete(self):
for item in self.threads:
if item.isAlive():item.join()class Work(threading.Thread):
def __init__(self, work_queue):
threading.Thread.__init__(self)
self.work_queue = work_queue
self.start() def run(self):
#死循环，从而让创建的线程在一定条件下关闭退出
while True:
try:
do, args = self.work_queue.get(block=False)#任务异步出队，Queue内部实现了同步机制
do(args)
self.work_queue.task_done()#通知系统任务完成
except:
break#具体要做的任务
def do_job(args):
time.sleep(0.1)#模拟处理时间
print threading.current_thread(), list(args)if __name__ == '__main__':
start = time.time()
work_manager = WorkManager(, )#或者work_manager = WorkManager(, )
work_manager.wait_allcomplete()
end = time.time()
print "cost all time: %s" % (end-start)