Day28：Event对象、队列、multiprocessing模块

event.isSet()：   返回event的状态值True或者False；

event.wait()：    如果 event.isSet()==False将阻塞线程；

event.set()：     设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；

event.clear()：   恢复event的状态值为False。

import threading,time

event = threading.Event()

def foo():
    while not event.is_set():
        print(‘wait....‘)
        event.wait()

    print(‘Connect to redis server‘)


print(‘attempt to start redis server‘)


for i in range(5):
    t = threading.Thread(target=foo)
    t.start()

time.sleep(10)
event.set()

‘‘‘
运行结果：
attempt to start redis server
wait....
wait....
wait....
wait....
wait....
Connect to redis server
Connect to redis server
Connect to redis server
Connect to redis server
Connect to redis server

‘‘‘

import threading,time

event = threading.Event()

def foo():
    while not event.is_set():
        print(‘wait....‘)
        event.wait(2)

    print(‘Connect to redis server‘)


print(‘attempt to start redis server‘)


for i in range(2):
    t = threading.Thread(target=foo)
    t.start()

time.sleep(5)
event.set()
‘‘‘
运行结果：
attempt to start redis server
wait....
wait....
wait....
wait....
wait....
wait....
Connect to redis server
Connect to redis server
‘‘‘

def worker(event):
    while not event.is_set():
        logging.debug(‘Waiting for redis ready...‘)
        event.wait(2)
    logging.debug(‘redis ready, and connect to redis server and do some work [%s]‘, time.ctime())
    time.sleep(1)

import  queue
q = queue.Queue(3)

q.put(11)
q.put(‘hello‘)
q.put(3.123)

print(q.get())
print(q.get())
print(q.get())
‘‘‘
运行结果：
11
hello
3.123
‘‘‘

‘‘‘
join() 阻塞进程，直到所有任务完成，需要配合另一个方法task_done。

    def join(self):
     with self.all_tasks_done:
      while self.unfinished_tasks:
       self.all_tasks_done.wait()

task_done() 表示某个任务完成。每一条get语句后需要一条task_done。


import queue
q = queue.Queue(5)
q.put(10)
q.put(20)
print(q.get())
q.task_done()
print(q.get())
q.task_done()

q.join()

print("ending!")
‘‘‘

import  queue,threading
q = queue.Queue(3)
def foo():
    q.put(11)
    q.put(‘hello‘)
    q.put(3.123)
    q.join()
def bar():

    print(q.get())
    q.task_done()    #注释掉本行，程序将不会结束。

t1 = threading.Thread(target=foo)
t1.start()

for i in range(3):

    t = threading.Thread(target=bar)
    t.start()
‘‘‘
运行结果：
11
hello
3.123
‘‘‘

‘‘‘

此包中的常用方法(q = queue.Queue()):

q.qsize() 返回队列的大小
q.empty() 如果队列为空，返回True,反之False
q.full() 如果队列满了，返回True,反之False
q.full 与 maxsize 大小对应
q.get([block[, timeout]]) 获取队列，timeout等待时间
q.get_nowait() 相当q.get(False)非阻塞 
q.put(item) 写入队列，timeout等待时间
q.put_nowait(item) 相当q.put(item, False)
q.task_done() 在完成一项工作之后，q.task_done() 函数向任务已经完成的队列发送一个信号
q.join() 实际上意味着等到队列为空，再执行别的操作

‘‘‘

Python queue模块有三种队列及构造函数: 

1、Python queue模块的FIFO队列先进先出。  class queue.Queue(maxsize) 
2、LIFO类似于堆栈，即先进后出。           class queue.LifoQueue(maxsize) 
3、还有一种是优先级队列级别越低越先出来。 class queue.PriorityQueue(maxsize) 


import queue
#先进后出

q=queue.LifoQueue()

q.put(34)
q.put(56)
q.put(12)

print(q.get())
print(q.get())
print(q.get())
‘‘‘
运行结果：
12
56
34
‘‘‘

#优先级
q=queue.PriorityQueue()
q.put([5,100])
q.put([7,200])
q.put([3,"hello"])
q.put([4,{"name":"alex"}])

while 1:
  data=q.get()
  print(data)

‘‘‘
运行结果：
[3, ‘hello‘]
[4, {‘name‘: ‘alex‘}]
[5, 100]
[7, 200]
‘‘‘

import time,random
import queue,threading

q = queue.Queue()

def Producer(name):
  count = 0
  while count <10:
    print("making........")
    time.sleep(random.randrange(3))
    q.put(count)
    print(‘Producer %s has produced %s baozi..‘ %(name, count))
    count +=1

    print("ok......")
def Consumer(name):
  count = 0
  while count <10:
    time.sleep(random.randrange(3))
    if not q.empty():
        data = q.get()

        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()

# Process类调用
from multiprocessing import Process
import time
def f(name):

    print(‘hello‘, name,time.ctime())
    time.sleep(1)

if __name__ == ‘__main__‘:
    p_list=[]
    for i in range(3):
        p = Process(target=f, args=(‘alvin:%s‘%i,))
        p_list.append(p)
        p.start()
    for i in p_list:
        p.join()
    print(‘end‘)
‘‘‘
运行结果：
hello alvin:0 Wed Jul 19 16:06:40 2017
hello alvin:2 Wed Jul 19 16:06:40 2017
hello alvin:1 Wed Jul 19 16:06:40 2017
end
‘‘‘

#继承Process类调用
from multiprocessing import Process
import time

class MyProcess(Process):
    def __init__(self):
        super(MyProcess, self).__init__()
        # self.name = name

    def run(self):

        print (‘hello‘, self.name,time.ctime())
        time.sleep(1)

if __name__ == ‘__main__‘:
    p_list=[]
    for i in range(3):
        p = MyProcess()
        p.start()
        p_list.append(p)

    for p in p_list:
        p.join()

    print(‘end‘)


‘‘‘
运行结果：
hello MyProcess-3 Wed Jul 19 16:09:39 2017
hello MyProcess-1 Wed Jul 19 16:09:39 2017
hello MyProcess-2 Wed Jul 19 16:09:39 2017
end
‘‘‘

from multiprocessing import Process
import os
import time
def info(name):


    print("name:",name)
    print(‘parent process:‘, os.getppid())
    print(‘process id:‘, os.getpid())
    print("------------------")
    time.sleep(1)


if __name__ == ‘__main__‘:

    info(‘main process line‘)


    p1 = Process(target=info, args=(‘alvin‘,))
    p2 = Process(target=info, args=(‘egon‘,))
    p1.start()
    p2.start()

    p1.join()
    p2.join()

    print("ending")
‘‘‘
运行结果：
name: main process line
parent process: 3400
process id: 1712
------------------
name: alvin
parent process: 1712
process id: 8428
------------------
name: egon
parent process: 1712
process id: 8212
------------------
ending

‘‘‘

from multiprocessing import Process, Queue

def f(q,n):
    q.put(n*n+1)
    print("son process",id(q))

if __name__ == ‘__main__‘:
    q = Queue()  #如果使用线程间的队列queue.Queue则无法运行
    print("main process",id(q))

    for i in range(3):
        p = Process(target=f, args=(q,i))
        p.start()

    print(q.get())
    print(q.get())
    print(q.get())
‘‘‘
运行结果：
main process 41655376
son process 45073408
1
son process 44942336
2
son process 44942392
5

‘‘‘

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([12, {"name": "yuan"}, ‘hello‘])
    response = conn.recv()
    print("response", response)
    conn.close()

if __name__ == ‘__main__‘:

    parent_conn, child_conn = Pipe()    #管道两个对象

    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())  
    parent_conn.send("儿子你好!")
    p.join()
‘‘‘
运行结果：
[12, {‘name‘: ‘yuan‘}, ‘hello‘]
response 儿子你好!
‘‘‘

from multiprocessing import Process, Manager

def f(d, l, n):

    d[n] = n
    d["name"] ="alvin"
    l.append(n)

    #print("l",l)

if __name__ == ‘__main__‘:

    with Manager() as manager:

        d = manager.dict()      　　 #字典

        l = manager.list(range(5))  #列表

        print(d,‘\n‘,l)
        p_list = []

        for i in range(10):
            p = Process(target=f, args=(d,l,i))
            p.start()
            p_list.append(p)

        for res in p_list:
            res.join()

        print(d)
        print(l)

‘‘‘
运行结果：
{} 　　初始化的字典
[0, 1, 2, 3, 4]　　初始化的列表
{3: 3, ‘name‘: ‘alvin‘, 0: 0, 2: 2, 7: 7, 5: 5, 4: 4, 1: 1, 6: 6, 8: 8, 9: 9}
[0, 1, 2, 3, 4, 3, 0, 2, 7, 5, 4, 1, 6, 8, 9]

from multiprocessing import Pool
import time

def foo(args):
 time.sleep(5)
 print(args)

if __name__ == ‘__main__‘:
     p = Pool(5)
     for i in range(30):
         p.apply_async(func=foo, args= (i,))

     p.close()   # 等子进程执行完毕后关闭进程池
     # time.sleep(2)
     # p.terminate()  # 立刻关闭进程池
     p.join()         # 没有join会立即结束