首页 > 代码库 > day9---多线程，线程锁，队列
day9---多线程，线程锁，队列
进程、线程
http://www.ruanyifeng.com/blog/2013/04/processes_and_threads.html
使用threading模块实现多线程编程[综述]
Python这门解释性语言也有专门的线程模型，Python虚拟机使用GIL（Global Interpreter Lock，全局解释器锁）来互斥线程对共享资源的访问，但暂时无法利用多处理器的优势。
在Python中我们主要是通过thread和 threading这两个模块来实现的，其中Python的threading模块是对thread做了一些包装的，可以更加方便的被使用，所以我们使用 threading模块实现多线程编程。这篇文章我们主要来看看Python对多线程编程的支持。
在语言层面，Python对多线程提供了很好的支持，可以方便地支持创建线程、互斥锁、信号量、同步等特性。下面就是官网上介绍threading模块的基本资料及功能：
实现模块
thread：多线程的底层支持模块，一般不建议使用；
threading：对thread进行了封装，将一些线程的操作对象化。
threading模块
Thread 线程类，这是我们用的最多的一个类，你可以指定线程函数执行或者继承自它都可以实现子线程功能；
Timer与Thread类似，但要等待一段时间后才开始运行；
Lock 锁原语，这个我们可以对全局变量互斥时使用；
RLock 可重入锁，使单线程可以再次获得已经获得的锁；
Condition 条件变量，能让一个线程停下来，等待其他线程满足某个“条件”；
Event 通用的条件变量。多个线程可以等待某个事件发生，在事件发生后，所有的线程都被激活；
Semaphore为等待锁的线程提供一个类似“等候室”的结构；
BoundedSemaphore 与semaphore类似，但不允许超过初始值；
Queue：实现了多生产者（Producer）、多消费者（Consumer）的队列，支持锁原语，能够在多个线程之间提供很好的同步支持。
其中Thread类
是你主要的线程类，可以创建进程实例。该类提供的函数包括：
getName(self) 返回线程的名字
isAlive(self) 布尔标志，表示这个线程是否还在运行中
isDaemon(self) 返回线程的daemon标志
join(self, timeout=None) 程序挂起，直到线程结束，如果给出timeout，则最多阻塞timeout秒
run(self) 定义线程的功能函数
setDaemon(self, daemonic) 把线程的daemon标志设为daemonic
setName(self, name) 设置线程的名字
start(self) 开始线程执行
其中Queue提供的类
Queue队列
LifoQueue后入先出（LIFO）队列
PriorityQueue 优先队列
Python threading模块
#!/usr/bin/env python
#-*- coding:utf-8 -*-
# Author:DCC
import threading
import time
def run(n):
print("task",n)
time.sleep(2)
t1 = threading.Thread(target=run,args=("t1",))
t2 = threading.Thread(target=run,args=("t2",))
t1.start()
t2.start()
print(t1.getName())
print(t2.getName())#!/usr/bin/env python
#-*- coding:utf-8 -*-
# Author:DCC
import threading
import time
class MyThread(threading.Thread):
def __init__(self,n):
super(MyThread,self).__init__()
self.n = n
def run(self):
print("running task",self.n)
time.sleep(3)
t1 = MyThread("t1")
t2 = MyThread("t2")
if __name__ == ‘__main__‘:
t1.start()
t2.start()
print(t1.getName())
print(t2.getName())Join & Daemon
一般情况下 主线程是不等待子线程是否执行完成的，只是触发一下，就不管了。
1、join ()方法：主线程A中，创建了子线程B，并且在主线程A中调用了B.join()，那么，主线程A会在调用的地方等待，直到子线程B完成操作后，才可以接着往下执行，那么在调用这个线程时可以使用被调用线程的join方法。
原型：join([timeout])
里面的参数时可选的，代表线程运行的最大时间，即如果超过这个时间，不管这个此线程有没有执行完毕都会被回收，然后主线程或函数都会接着执行的。
import threading import time class MyThread(threading.Thread): def __init__(self,id): threading.Thread.__init__(self) self.id = id def run(self): x = 0 time.sleep(10) print self.id if __name__ == "__main__": t1=MyThread(999) t1.start() for i in range(5): print i #执行结果 0 1 2 3 4 999
机器上运行时，4和999之间，有明显的停顿。解释：线程t1 start后，主线程并没有等线程t1运行结束后再执行，而是先把5次循环打印执行完毕（打印到4），然后sleep（10）后，线程t1把传进去的999才打印出来。
现在，我们把join()方法加进去(其他代码不变)，看看有什么不一样，例子：
import threading import time class MyThread(threading.Thread): def __init__(self,id): threading.Thread.__init__(self) self.id = id def run(self): x = 0 time.sleep(10) print self.id if __name__ == "__main__": t1=MyThread(999) t1.start() t1.join() for i in range(5): print i #执行结果 999 0 1 2 3 4
2、setDaemon()方法。主线程A中，创建了子线程B，并且在主线程A中调用了B.setDaemon(),这个的意思是，把主线程A设置为守护线程，这时候，要是主线程A执行结束了，就不管子线程B是否完成,一并和主线程A退出.这就是setDaemon方法的含义，这基本和join是相反的。此外，还有个要特别注意的：必须在start() 方法调用之前设置，如果不设置为守护线程，程序会被无限挂起。
例子：就是设置子线程随主线程的结束而结束：
#!/usr/bin/env python
#-*- coding:utf-8 -*-
# Author:DCC
import threading
import time
def run(n):
print("task",n)
time.sleep(2)
print("thread done...",n)
start_time = time.time()
t_objs = [] #存线程实例
for i in range(50):
t = threading.Thread(target=run,args=("t-%s"% i,))
t.setDaemon(True) #把当前线程设置为守护线程
t.start()
t_objs.append(t) # 为了不阻塞后面线程的启动，不在这里join，先放到-个列表里
# print(t.getName())
#time.sleep(2)
print(threading.active_count())
# for i in t_objs: #循环线程实例列表，等待所有线程执行完毕
# t.join()
print(time.time()-start_time) 线程锁
CPU执行任务时，在线程之间是进行随机调度的，并且每个线程可能只执行n条代码后就转而执行另外一条线程。由于在一个进程中的多个线程之间是共享资源和数据的，这就容易造成资源抢夺或脏数据，于是就有了锁的概念，限制某一时刻只有一个线程能访问某个指定的数据。
未枷锁
import threading import time NUM = 0 def show(): global NUM NUM += 1 name = t.getName() time.sleep(1) # 注意，这行语句的位置很重要，必须在NUM被修改后，否则观察不到脏数据的现象。 print(name, "执行完毕后，NUM的值为： ", NUM) for i in range(10): t = threading.Thread(target=show) t.start() print(‘main thread stop‘)
LOCK锁
普通锁，也叫互斥锁，是独占的，同一时刻只有一个线程被放行。
import time import threading NUM = 10 def func(lock): global NUM lock.acquire() # 让锁开始起作用 NUM -= 1 time.sleep(1) print(NUM) lock.release() # 释放锁 lock = threading.Lock() # 实例化一个锁对象 for i in range(10): t = threading.Thread(target=func, args=(lock,)) # 记得把锁当作参数传递给func参数 t.start()
RLock（递归锁）
说白了就是在一个大锁中还要再包含子锁
threading模块的Lock类，它不支持嵌套锁。RLcok类的用法和Lock一模一样，但它支持嵌套，因此我们一般直接使用RLcok类。
import threading, time
def run1():
print("grab the first part data")
lock.acquire()
global num
num += 1
lock.release()
return num
def run2():
print("grab the second part data")
lock.acquire()
global num2
num2 += 1
lock.release()
return num2
def run3():
lock.acquire()
res = run1()
print(‘--------between run1 and run2-----‘)
res2 = run2()
lock.release()
print(res, res2)
if __name__ == ‘__main__‘:
num, num2 = 0, 0
lock = threading.RLock()
for i in range(10):
t = threading.Thread(target=run3)
t.start()
while threading.active_count() != 1:
print(threading.active_count())
else:
print(‘----all threads done---‘)
print(num, num2)时器（Timer）
定时器，指定n秒后执行某操作。很简单但很使用的东西。
from threading import Timer
def hello():
print("hello, world")
t = Timer(1, hello) # 表示1秒后执行hello函数
t.start()信号量(Semaphore)
这种锁允许一定数量的线程同时更改数据，它不是互斥锁。比如地铁安检，排队人很多，工作人员只允许一定数量的人进入安检区，其它的人继续排队。
互斥锁 同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据 ，比如厕所有3个坑，那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去。
import time
import threading
def run(n):
semaphore.acquire()
print("run the thread: %s" % n)
time.sleep(1)
semaphore.release()
num = 0
semaphore = threading.BoundedSemaphore(5) # 最多允许5个线程同时运行
for i in range(20):
t = threading.Thread(target=run, args=(i,))
t.start()事件(Event)
事件主要提供了三个方法 set、wait、clear。
事件机制：全局定义了一个“Flag”，如果“Flag”的值为False，那么当程序执行wait方法时就会阻塞，如果“Flag”值为True，那么wait方法时便不再阻塞。这种锁，类似交通红绿灯（默认是红灯），它属于在红灯的时候一次性阻挡所有线程，在绿灯的时候，一次性放行所有的排队中的线程。
clear：将“Flag”设置为False
set：将“Flag”设置为True
!/usr/bin/env python
#-*- coding:utf-8 -*-
# Author:DCC
import time
import threading
import random
event = threading.Event()
def lighter():
count = 0
event.set() #先设置成绿灯
while True:
if count > 5 and count < 10:
#改成红灯
event.clear() #把标志位清除
print("\033[41;1m red.....\033[0m")
elif count > 10:
event.set() #设置成路灯
count = 0
else:
print("\033[42;1m green \033[0m")
time.sleep(1)
count +=1
def car(name):
while True:
if event.is_set(): #代表绿灯
print("[%s] is running " % name)
time.sleep(2)
else:
print("[%s] is waitting....... " % name)
event.wait()
print("[%s] green light is on ,start going" % name)
light = threading.Thread(target=lighter,)
light.start()
car1 = threading.Thread(target=car,args=("Tesla",))
car1.start()队列
通常而言，队列是一种先进先出的数据结构，与之对应的是堆栈这种后进先出的结构。但是在python中，它内置了一个queue模块，它不但提供普通的队列，还提供一些特殊的队列
Queue：先进先出队列
import queue q = queue.Queue(5) q.put(11) q.put(22) q.put(33) print(q.get()) print(q.get()) print(q.get())
Queue类的参数和方法：
qsize() 获取当前队列中元素的个数，也就是队列的大小
empty() 判断当前队列是否为空，返回True或者False
full() 判断当前队列是否已满，返回True或者False
put(self, block=True, timeout=None)
get(self, block=True, timeout=None)
LifoQueue：后进先出队列
import queue q = queue.LifoQueue() q.put(123) q.put(456) print(q.get())
PriorityQueue：优先级队列
q = queue.PriorityQueue() q.put((1,"alex1")) q.put((1,"alex2")) q.put((1,"alex3")) q.put((3,"alex3")) print(q.get())
生产者消费者模型
import time,random
import queue,threading
q = queue.Queue()
def Producer(name):
count = 0
while count <20:
time.sleep(random.randrange(3))
q.put(count)
print(‘Producer %s has produced %s baozi..‘ %(name, count))
count +=1
def Consumer(name):
count = 0
while count <20:
time.sleep(random.randrange(4))
if not q.empty():
data = q.get()
print(data)
print(‘\033[32;1mConsumer %s has eat %s baozi...\033[0m‘ %(name, data))
else:
print("-----no baozi anymore----")
count +=1
p1 = threading.Thread(target=Producer, args=(‘A‘,))
c1 = threading.Thread(target=Consumer, args=(‘B‘,))
p1.start()
c1.start()本文出自 “东风十里柔情” 博客，请务必保留此出处http://xiaofengcanyue.blog.51cto.com/6671161/1855615
day9---多线程，线程锁，队列