Python的并发并行[2] -> 队列 -> 使用队列进行任务控制

 1 from random import randint
 2 from time import sleep, ctime
 3 from queue import Queue, LifoQueue
 4 from threading import Thread
 5 
 6 COUNT = 0
 7 
 8 
 9 class MyThread(Thread):
10     """
11     Bulid up a Module to make this subclass more general
12     And get return value by add a function named ‘getResult()‘
13     """
14     def __init__(self, func, args, name=‘‘):
15         Thread.__init__(self)
16         self.name = name
17         self.func = func
18         self.args = args
19 
20     def getResult(self):
21         return self.res
22 
23     def run(self):
24         print(‘Starting‘, self.name, ‘at:‘, ctime())
25         # Call function here and calculate the running time
26         self.res = self.func(*self.args)
27         print(self.name, ‘finished at:‘, ctime())
28 
29 
30 class MyQueue():
31     def __init__(self):
32         self.funcs = [self.writer, self.reader]
33         self.nfuncs = range(len(self.funcs))
34 
35     def writeQ(self, queue):
36         global COUNT
37         print(‘Producing object OBJ_%d for Q...‘ % COUNT, end=‘ ‘)
38         queue.put(‘OBJ_%d‘ % COUNT, True)
39         print(‘size now:‘, queue.qsize())
40         COUNT += 1
41     
42     def readQ(self, queue):
43         # If queue is empty, block here until queue available
44         val = queue.get(True)
45         print(‘Consumed object %s from Q... size now:‘ % val, queue.qsize())
46     
47     def writer(self, queue, loops):
48         for i in range(loops):
49             self.writeQ(queue)
50             sleep(randint(1, 3))
51     
52     def reader(self, queue, loops):
53         for i in range(loops):
54             self.readQ(queue)
55             sleep(randint(2, 5))
56     
57     def main(self):
58         nloops = randint(2, 5)
59         fifoQ = Queue(32)
60         lifoQ = LifoQueue(32)
61     
62         # First In First Out mode for Queue
63         print(‘-----Start FIFO Queue-----‘)
64         threads = []
65         for i in self.nfuncs:
66             threads.append(MyThread(self.funcs[i], (fifoQ, nloops), self.funcs[i].__name__))
67         for t in threads:
68             t.start()
69         for t in threads:
70             t.join()
71         # Last In First Out mode for LifoQueue
72         print(‘-----Start LIFO Queue-----‘)
73         threads = []
74         for i in self.nfuncs:
75             threads.append(MyThread(self.funcs[i], (lifoQ, nloops), self.funcs[i].__name__))
76         for t in threads:
77             t.start()
78         for t in threads:
79             t.join()
80     
81         print(‘All DONE‘)
82 
83 if __name__ == ‘__main__‘:
84      MyQueue().main()

 1 from Queue_FIFO_LIFO import *
 2 
 3 class NewQueue(MyQueue):
 4     def __init__(self):
 5         MyQueue.__init__(self)
 6 
 7     def writer(self, queue, loops):
 8         for i in range(loops):
 9             self.writeQ(queue)
10             sleep(randint(1, 3))
11         print(‘Producing join here, waiting consumer‘)
12         queue.join()
13     
14     def reader(self, queue, loops):
15         for i in range(loops):
16             self.readQ(queue)
17             sleep(randint(2, 5))
18             print(‘OBJ_%d task done‘ % i)
19             queue.task_done()
20 
21     def main(self):
22         nloops = randint(2, 5)
23         fifoQ = Queue(32)
24 
25         print(‘-----Start FIFO Queue-----‘)
26         threads = []
27         for i in self.nfuncs:
28             threads.append(MyThread(self.funcs[i], (fifoQ, nloops), self.funcs[i].__name__))
29         for t in threads:
30             t.start()
31         for t in threads:
32             t.join()
33 
34         print(‘All DONE‘)
35 
36 if __name__ == ‘__main__‘:
37     NewQueue().main()