使用python的threading中的Thread
下面是两种基本的实现线程的方式：
第一种方式————
#coding=utf-8 """ thread的第一种声明及调用方式 """ from threading import Thread import time def func(): while True: print time.ctime() time.sleep(1) def func1(): while True: print "func1", time.ctime() time.sleep(2) #将要执行的方法作为参数传给Thread的构造方法 t = Thread(target=func) t.start() t1 = Thread(target=func1) t1.start()
第二种方式————
#coding=utf-8 """ thread的第二种声明及调用方法 """ import time from threading import Thread from time import sleep #从Thread继承，并重写run() class myThread(Thread): def __init__(self, threadname): Thread.__init__(self, name=threadname) def run(self): while True: print self.getName() + " " + time.ctime() time.sleep(1) t = myThread("thread_1") t.start() t1 = myThread("thread_2") t1.start()
接下来是：join和setDaemon的方法调用及具体区别：
join的方法调用：
#coding=utf-8 """ join方法的含义是：存在A主线程，B子线程，在A主线程中生成了一个子线程B， 如果子线程B调用了join方法，则主线程需要等待子线程操作结束，才能够继续往下执行，主线程会被阻塞 """ import time from threading import Thread class myThread(Thread): def __init__(self, threadname): Thread.__init__(self, name = threadname) def run(self): time.sleep(10) print self.getName() + " " + time.ctime() def mainfunc(): for i in range(5): print i if __name__ == "__main__": t = myThread("childthread") t.start() #增加join方法的调用，则程序运行时的表现是：先子线程（会先sleep 10s，之后打印子线程的内容），再运行主线程中的打印从0到4的逻辑 #不增加join方法的调用，则程序运行时的表现是：会先走主线程的打印从0到4的逻辑，之后子线程（sleep 10s，再打印子线程中的内容），但其实这里就是没有等待子线程 # t.join() mainfunc()
setDaemon方法的调用：
#coding=utf-8 """ daemon方法的含义是：存在主线程A，和子线程B，子线程B是从主线程A中生成的 如果设置子线程B为daemon即守护进程，则主线程A不存在的话，子线程B就退出 如果没有设置子线程B为daemon，则主线程A结束运行退出之后，子线程B也不退出 """ import time from threading import Thread class myThread(Thread): def __init__(self, threadname): Thread.__init__(self, name=threadname) def run(self): time.sleep(10) print self.getName() + " " + time.ctime() def mainfunc(): for i in range(5): print i if __name__ == "__main__": t = myThread("zixiancheng") #不设置子线程为守护进程，则主线程退出，子线程继续运行 #设置子线程为守护进程，则主线程退出，子线程直接退出 #并且，需要在start之前调用setDaemon # t.setDaemon(True) t.start() mainfunc()
接下来是线程同步相关的内容：
首先是没有线程同步机制的情况下，多个线程同时操作一个公共数据：
代码如下：
#coding=utf-8 """ @function: test Thread的Lock @author: LiXia @date: 2017-0303 """ """ 所有的锁都是为了在线程之间进行通信 例如对一个公共变量进行操作，此例子为没有使用线程同步机制对公共变量进行操作 """ from threading import Thread import time data = 0 class myThread(Thread): def __init__(self): Thread.__init__(self) def run(self): global data time.sleep(1) data += 1 print self.getName() + " data: " + str(data) if __name__ == "__main__": for i in range(5): t = myThread() t.start()
运行结果见下方：
为了能够对公共操作的数据，在特定逻辑内保持一致，需要保证在该逻辑内，只有单个线程可以操作，其他线程需要等待该线程处理结束才可以。
所以下面是关于python线程同步机制提供的指令：
1、Lock：
但是将以上代码中增加上lock处理，注意一定要增加在对公共数据的逻辑判断和处理的前面和后面，所以这里必须添加到while判断之前，释放锁需要在输出操作结束之后添加
即：
#coding=utf-8 """ 所有的锁都是为了在线程之间进行通信 例如对一个公共变量进行操作 """ from threading import Thread import threading import time data = 0 lock = threading.Lock() class myThread(Thread): def __init__(self): Thread.__init__(self) def run(self): global data if lock.acquire(): time.sleep(1) data += 1 print self.getName() + " data: " + str(data) lock.release() if __name__ == "__main__": for i in range(5): t = myThread() t.start()
运行结果见下方：
2、RLock
代码如下：
#coding=utf-8 """ 通过RLock（可重入锁）进行线程间的安全通信,RLock为可重入锁，即线程在拥有RLock之后可以继续acquire()，只要保证acquire()次数与 release()次数一致即可； 继续对公共变量进行操作 """ from threading import Thread import threading import time rlock = threading.RLock() data = 0 class myThread(Thread): def __init__(self): Thread.__init__(self) def run(self): global data if rlock.acquire(): time.sleep(1) data += 1 print self.getName() + " data:" + str(data) if rlock.acquire(): time.sleep(1) data -= 1 print self.getName() + " data:" + str(data) rlock.release() rlock.release() if __name__ == "__main__": for i in range(5): t = myThread() t.start()
运行结果：
3、Condition
Condition就是条件变量，condition通常与一个锁关联。
acquire([timeout])/release(): 调用关联的锁的相应方法。
wait([timeout]): 调用这个方法将使线程进入Condition的等待池等待通知，并释放锁。使用前线程必须已获得锁定，否则将抛出异常。
notify(): 调用这个方法将从等待池挑选一个线程并通知，收到通知的线程将自动调用acquire()尝试获得锁定（进入锁定池）；其他线程仍然在等待池中。调用这个方法不会释放锁定。使用前线程必须已获得锁定，否则将抛出异常。
notifyAll(): 调用这个方法将通知等待池中所有的线程，这些线程都将进入锁定池尝试获得锁定。调用这个方法不会释放锁定。使用前线程必须已获得锁定，否则将抛出异常。
#coding=utf-8 """ 以最经典的生产者和消费者的问题为例，两者需要针对一个公共的数据区域做操作，并且两端需要进行通信 生产者在生产之前，需要先占领地盘，然后开始生产，生产结束通知消费者来取； 消费者在消费的时候，也需要先占领地盘，然后开始消费，消费结束通知生产者继续生产； 当然还有多个生产者和消费者的情况，会更加复杂，可以用python的Queue来实现 """ from threading import Thread import threading import time import random product = [] con = threading.Condition() #得到一个Condition的对象 #生产者线程 class producerThread(Thread): def __init__(self, threadname): Thread.__init__(self, name=threadname) def run(self): global product if con.acquire(): while True: r = random.randint(0, 20) product.append(r) print "product len and product: ", self.getName(), product con.notify() con.wait() # time.sleep(2) #消费者线程 class consumerThread(Thread): def __init__(self, threadname): Thread.__init__(self, name=threadname) def run(self): global product if con.acquire(): while True: if product != []: product.pop() print "consumer len and product: ", self.getName(), product con.notify() con.wait() time.sleep(2) if __name__ == "__main__": t1 = producerThread("producer") t2 = consumerThread("consumer") t1.start() t2.start()
运行结果如下：
4、对多个生产者消费者支持的很好的Queue
Queue是什么？Queue实现了多生产者-多消费者的队列形式，分别包含先入先出（默认）FIFO、后进先出LIFO、以及优先级队列PriorityQueue（优先级低的先出）
Queue模块的class及exception：
#coding=utf-8 """ Queue的类和异常 """ import Queue q = Queue.Queue(maxsize = 0) #FIFO的Queue，maxsize即Queue的size值，maxsize≤0表示无size限制 lq = Queue.LifoQueue(maxsize = 0) #LIFO的Queue，先进后出，maxsiz的含义同上 pq = Queue.PriorityQueue(maxsize = 0) #优先级Queue，优先级低的先出，maxsize的含义同上 Queue.Empty() #当Queue的对象为空时，使用非阻塞get时会抛出异常 Queue.Full() #当Queue的对象已满时，使用非阻塞put时会抛出异常
Queue的对象：
Queue（Queue、LifoQueue、PriorityQueue）的对象提供的公共方法如下：
#coding=utf-8 """ Queue（Queue、LifoQueue、PriorityQueue）的对象的公共方法 """ import Queue q = Queue.Queue() q.qsize() #获取Queue的大小 q.empty() #是否为空，是则返回True，否则返回False q.full() #是否已满，是则返回True，否则返回False q.put(item[, block[, timeout]]) #put元素到Queue中，可选参数block默认值为True,timeout默认值为None，即为阻塞方法， #接上，若timeout为正值，则代表会阻塞timeout的时长之后引发一个full的异常（如果已满） q.put_nowait(item) #相当于q.put(item, False) q.get([block[, timeout]]) #get元素到Queue中，可选参数block默认值为True，timeout默认值为None，即为阻塞方法， #接上，若timeout为正数，则代表会则色timeouu的时长之后会引发一个empty的异常（如果为空） q.get_nowait() #相当于q.get(False) q.task_done() #用在Queue的消费者模式下，即用在Queue.get()方法之后，通过Queue.task_done()告诉Queue当前任务完成 #如果当前join正在阻塞，则它将在所有item都被处理之后恢复 q.join() #会保持阻塞直到队列中的所有item都会处理 #接上，未完成的task数目会增加当一个item被add到Queue中，未完成的task数目会降低，每当消费者线程调用task_done来表示所有的 #接上，work都完成了，当未完成的数目降低到0时，join就自动取消阻塞了
Queue的主要用法是什么？
看一下Queue的源码实现，然后在另一篇博客中更新一下。
#coding=utf-8 """ 理解Queue的用法，理解Queue衍生出来的不同的class """ from Queue import Queue from threading import Thread import time import random q = Queue() class producerQueue(Thread): def __init__(self, queue): Thread.__init__(self) self.queue = queue def run(self): while True: r = random.randint(0, 99) self.queue.put(r) print self.getName() + "produce " + str(r) time.sleep(1) class consumerQueue(Thread): def __init__(self, queue): Thread.__init__(self) self.queue = queue def run(self): while True: r = self.queue.get() print self.getName() + "consume " + str(r) time.sleep(3) self.queue.task_done() if __name__ == "__main__": threads = [] for i in range(5): t = producerQueue(q) t.start() # threads.append(t) for i in range(3): t = consumerQueue(q) t.start() # threads.append(t) q.join()
之后得到的结果如下：
