Python Thread related

1.Thread.join([timeout])

Wait until the thread terminates. This blocks the calling thread until the thread whose join() method is called terminates – either normally or through an unhandled exception – or until the optional timeout occurs.

等待進程結束。也就是說，其屏蔽調用線程，直到此線程方法終止（要麼正常執行完畢，或者未處理的異常，或者時間超時）

下面經過例子來講明：

沒有設定timeout狀況，Main 線程啓動，work1 和work2 線程執行，完畢退出，Main線程執行終止

 1 import os
 2 import threading
 3 import time
 4 import logging
 5 import random
 6 
 7 def work1():
 8     count=0
 9     while count<=5:
10         threadname= threading.currentThread()
11         wait_time=random.randrange(1,4)
12         print("%s,count =%s wait for =%s s,time %s "%(threadname,count,wait_time,time.ctime()[-13:]))
13         time.sleep(wait_time)
14         count +=1
15      
16 def work2():
17     i=0
18     while i<=5:
19         threadname= threading.currentThread()
20         wait_time=random.randrange(1,4)
21         print("%s,i     =%s wait for =%s s,time %s "%(threadname,i,wait_time,time.ctime()[-13:]))
22         time.sleep(wait_time)
23         i +=1
24 
25 if __name__ =="__main__":
26     mainthread= threading.currentThread()
27     print '%s main thread is waiting for exit'% mainthread
28     test1=threading.Thread(name='work1',target=work1)
29     test2=threading.Thread(name='work2',target=work2)
30     test1.start()
31     test2.start()
32     test1.join()
33     test2.join()    
34     print 'main thread finish'

View Code

 

2個線程設定超時時間work1 5s,work2 4s，9s以後調用線程結束而不等待超時的線程：

 1 import os
 2 import threading
 3 import time
 4 import logging
 5 import random
 6 
 7 def work1():
 8     count=0
 9     while count<=5:
10         threadname= threading.currentThread()
11         wait_time=random.randrange(1,4)
12         print("%s,count =%s wait for =%s s,time %s "%(threadname,count,wait_time,time.ctime()[-13:]))
13         time.sleep(wait_time)
14         count +=1
15      
16 def work2():
17     i=0
18     while i<=5:
19         threadname= threading.currentThread()
20         wait_time=random.randrange(1,4)
21         print("%s,i     =%s wait for =%s s,time %s "%(threadname,i,wait_time,time.ctime()[-13:]))
22         time.sleep(wait_time)
23         i +=1
24 
25 if __name__ =="__main__":
26     mainthread= threading.currentThread()
27     print '%s main thread is waiting for exit'% mainthread
28     test1=threading.Thread(name='work1',target=work1)
29     test2=threading.Thread(name='work2',target=work2)
30     test1.start()
31     test2.start()
32     test1.join(4)
33     test2.join(5)    
34     print 'main thread finish

View Code

 2.Producer and comsumer

 1 import Queue
 2 import threading
 3 import random
 4 
 5 #writelock =threading.Lock()
 6 class Producer(threading.Thread):
 7     def __init__(self,q,con,name):
 8         super(Producer,self).__init__()
 9         self.q = q
10         self.con = con
11         self.name = name
12         print "produce" +self.name+"started"
13     def run(self):
14         while 1:
15             #global writelock      
16             self.con.acquire()#acquire the lock   
17             if self.q.full():#if queue is full
18                #with writelock:#output info
19                print "queue is full, producer wait"
20                self.con.wait()#wait for resource
21             else:
22                 value = random.ranint(0,10)
23                 #with writelock:
24                 print self.name+"put value"+self.name+":"+str(value)+"into queue"
25             self.q.put((self.name+":"+str(value)))#put to queue
26             self.con.notify()#inform consumer
27         self.con.release()#release the lock
28         
29 class Consumer(threading.Thread):
30     def __init__(self,q,con,name):
31         super(Consumer,self).__init__()
32         self.q = q
33         self.con = con
34         self.name = name
35         print "consume" +self.name+"started\n"
36     def run(self):
37         while 1:
38             #global writelock
39             self.con.acquire()
40             if self.q.empty():#if empty
41                 #with writelock:
42                 print "queue is empty,consumer wait"
43                 self.con.wait()#wait the resource ready
44             else:
45                 value = self.q.get()#get one element from queue
46                 #with writelock:
47                 print self.name +"get value"+ value+"from queue"
48                 self.q.notify()#inform producer
49             self.con.release()#release the lock
50 
51 
52 if __name__ == "__main__":
53     print "start to run\n"
54     q = Queue.Queue(10)
55     con = threading.Condition()
56     p = Producer(q,con,"p1")
57     p.start()
58     p1 = Producer(q,con,"p2")
59     p1.start()
60     c1 = Consumer(q,con,"c1")
61     c1.start()
62 
63 
64         
65 
66         
67     

View Code

3.Queue

programming python 4th 205頁

Queue 提供標準的隊列數據結構，實現python對象的先進先出，其可包含基本類型（string，list，dictionary......），類實例，任何可調用函數或綁定的方法等。可是Queue不像正常的list，由於其自動被線程獲取和釋放鎖操做。

#coding:utf-8"
import logging,threading,time
import Queue

def fibo_task(cond):
    with cond:
        while shared_queue.empty():
            logger.info("[%s]- waiting for element in queue......" % threading.current_thread().name)
            cond.wait()
        else:
            value =shared_queue.get()
            a,b=0,1
            for item in range(value):
                a,b=b,a+b
                fibo_dict[value] = a
        shared_queue.task_done()
        time.sleep(2)
        logger.debug("[%s] fibo of key[%d] with result[%d]" % (threading.current_thread().name,value,fibo_dict[value]))
                     
def queue_task(cond):
    logging.debug('starting  list data to queue......')
    with cond:
        for data in impit_list:
            shared_queue.put(data)
        #[shared_queue.put(data) for data in  impit_list]
        cond.notifyAll()


if __name__ == "__main__":
    logger =logging.getLogger()
    logger.setLevel(logging.DEBUG)
    formatter =logging.Formatter('%(asctime)s =%(message)s')
    
    ch=logging.StreamHandler()
    ch.setLevel(logging.DEBUG)
    ch.setFormatter(formatter)
    logger.addHandler(ch)
    fibo_dict={}
    shared_queue =Queue.Queue()
    impit_list =[3,10,5,7]
    queue_cond=threading.Condition()
    print "main thread starting......"
    threads =[threading.Thread(target=fibo_task,args=(queue_cond,)) for i in range(4)]
    #for thread in threads:
        #thread.setDaemon(True)
        #print 'daemon is %d' % thread.isDaemon()
    
    [thread.start() for thread in threads]
    
    prod = threading.Thread(name='queue_task_thread',target=queue_task,args=(queue_cond,))
    prod.setDaemon(True)
    prod.start()
   
    [thread.join() for thread in threads]    
    print "main thread done"

View Code