第五篇 进程&线程&协程

标签：bar   lap   producer   inpu   sub   操作   close   生产者   gre   

• 多进程
  
  

   1 #方式一：
   2 # from multiprocessing import Process
   3 # import time
   4 #
   5 # def task(name):
   6 #     print(‘%s is running‘ %name)
   7 #     time.sleep(3)
   8 #     print(‘%s is done‘ %name)
   9 #
  10 # if __name__ == ‘__main__‘:
  11 #     # Process(target=task,kwargs={‘name‘:‘子进程1‘})
  12 #     p=Process(target=task,args=(‘子进程1‘,))
  13 #     p.start() #仅仅只是给操作系统发送了一个信号
  14 #
  15 #     print(‘主‘)
  16 
  17 #方式二
  18 from multiprocessing import Process
  19 import time
  20 
  21 class MyProcess(Process):
  22     def __init__(self,name):
  23         super().__init__()
  24         self.name=name
  25 
  26     def run(self):
  27         print(‘%s is running‘ %self.name)
  28         time.sleep(3)
  29         print(‘%s is done‘ %self.name)
  30 
  31 
  32 if __name__ == ‘__main__‘:
  33     p=MyProcess(‘子进程1‘)
  34     p.start()
  35     print(‘主‘)

  开启进程的2种方式
  

   1 from multiprocessing import Process
   2 import time,os
   3 
   4 def task():
   5     print(‘%s is running,parent id is <%s>‘ %(os.getpid(),os.getppid()))
   6     time.sleep(3)
   7     print(‘%s is done,parent id is <%s>‘ %(os.getpid(),os.getppid()))
   8 
   9 if __name__ == ‘__main__‘:
  10     p=Process(target=task,)
  11     p.start()
  12 
  13     print(‘主‘,os.getpid(),os.getppid())

  查看pid
  

    1 #join方法
    2 # from multiprocessing import Process
    3 # import time,os
    4 #
    5 # def task():
    6 #     print(‘%s is running,parent id is <%s>‘ %(os.getpid(),os.getppid()))
    7 #     time.sleep(3)
    8 #     print(‘%s is done,parent id is <%s>‘ %(os.getpid(),os.getppid()))
    9 #
   10 # if __name__ == ‘__main__‘:
   11 #     p=Process(target=task,)
   12 #     p.start()
   13 #
   14 #     p.join()
   15 #     print(‘主‘,os.getpid(),os.getppid())
   16 #     print(p.pid)
   17 
   18 
   19 # from multiprocessing import Process
   20 # import time,os
   21 #
   22 # def task(name,n):
   23 #     print(‘%s is running‘ %name)
   24 #     time.sleep(n)
   25 #
   26 #
   27 # if __name__ == ‘__main__‘:
   28 #     start=time.time()
   29 #     p1=Process(target=task,args=(‘子进程1‘,5))
   30 #     p2=Process(target=task,args=(‘子进程2‘,3))
   31 #     p3=Process(target=task,args=(‘子进程3‘,2))
   32 #     p_l=[p1,p2,p3]
   33 #
   34 #     # p1.start()
   35 #     # p2.start()
   36 #     # p3.start()
   37 #     for p in p_l:
   38 #         p.start()
   39 #
   40 #     # p1.join()
   41 #     # p2.join()
   42 #     # p3.join()
   43 #     for p in p_l:
   44 #         p.join()
   45 #
   46 #     print(‘主‘,(time.time()-start))
   47 
   48 
   49 
   50 
   51 
   52 # from multiprocessing import Process
   53 # import time,os
   54 #
   55 # def task(name,n):
   56 #     print(‘%s is running‘ %name)
   57 #     time.sleep(n)
   58 #
   59 #
   60 # if __name__ == ‘__main__‘:
   61 #     start=time.time()
   62 #     p1=Process(target=task,args=(‘子进程1‘,5))
   63 #     p2=Process(target=task,args=(‘子进程2‘,3))
   64 #     p3=Process(target=task,args=(‘子进程3‘,2))
   65 #
   66 #     p1.start()
   67 #     p1.join()
   68 #     p2.start()
   69 #     p2.join()
   70 #     p3.start()
   71 #     p3.join()
   72 #
   73 #     print(‘主‘,(time.time()-start))
   74 
   75 
   76 #了解:is_alive
   77 from multiprocessing import Process
   78 import time,os
   79 
   80 def task():
   81     print(‘%s is running,parent id is <%s>‘ %(os.getpid(),os.getppid()))
   82     time.sleep(3)
   83     print(‘%s is done,parent id is <%s>‘ %(os.getpid(),os.getppid()))
   84 
   85 if __name__ == ‘__main__‘:
   86     # p=Process(target=task,)
   87     # p.start()
   88     # # print(p.is_alive())
   89     # p.join()
   90     # print(‘主‘,os.getpid(),os.getppid())
   91     # print(p.pid)
   92     # # print(p.is_alive())
   93 
   94 
   95 
   96     p=Process(target=task,name=‘sub——Precsss‘)
   97     p.start()
   98     p.terminate()
   99     time.sleep(3)
  100     print(p.is_alive())
  101     print(‘主‘)
  102     print(p.name)

  Process对象的其他属性
  

   1 # from multiprocessing import Process
   2 # import time
   3 #
   4 # def task(name):
   5 #     print(‘%s is running‘ %name)
   6 #     time.sleep(2)
   7 #     p=Process(target=time.sleep,args=(3,))
   8 #     p.start()
   9 #
  10 #
  11 # if __name__ == ‘__main__‘:
  12 #     p=Process(target=task,args=(‘子进程1‘,))
  13 #     p.daemon=True
  14 #     p.start()
  15 #
  16 #     p.join()
  17 #     print(‘主‘)
  18 
  19 
  20 #练习题
  21 from multiprocessing import Process
  22 
  23 import time
  24 def foo():
  25     print(123)
  26     time.sleep(1)
  27     print("end123")
  28 
  29 def bar():
  30     print(456)
  31     time.sleep(3)
  32     print("end456")
  33 
  34 if __name__ == ‘__main__‘:
  35     p1=Process(target=foo)
  36     p2=Process(target=bar)
  37 
  38     p1.daemon=True
  39     p1.start()
  40     p2.start()
  41     print("main-------")

  守护进程
  

   1 from multiprocessing import Process,Lock
   2 import json
   3 import time
   4 
   5 def search(name):
   6     time.sleep(1)
   7     dic=json.load(open(‘db.txt‘,‘r‘,encoding=‘utf-8‘))
   8     print(‘<%s> 查看到剩余票数【%s】‘ %(name,dic[‘count‘]))
   9 
  10 
  11 def get(name):
  12     time.sleep(1)
  13     dic=json.load(open(‘db.txt‘,‘r‘,encoding=‘utf-8‘))
  14     if dic[‘count‘] > 0:
  15         dic[‘count‘]-=1
  16         time.sleep(3)
  17         json.dump(dic,open(‘db.txt‘,‘w‘,encoding=‘utf-8‘))
  18         print(‘<%s> 购票成功‘ %name)
  19 
  20 
  21 def task(name,mutex):
  22     search(name)
  23     mutex.acquire()
  24     get(name)
  25     mutex.release()
  26 
  27 if __name__ == ‘__main__‘:
  28     mutex=Lock()
  29     for i in range(10):
  30         p=Process(target=task,args=(‘路人%s‘ %i,mutex))
  31         p.start()

  互斥锁-模拟抢票示例
  

   1 from multiprocessing import Process,Lock
   2 import json
   3 import time
   4 
   5 def search(name):
   6     time.sleep(1)
   7     dic=json.load(open(‘db.txt‘,‘r‘,encoding=‘utf-8‘))
   8     print(‘<%s> 查看到剩余票数【%s】‘ %(name,dic[‘count‘]))
   9 
  10 
  11 def get(name):
  12     time.sleep(1)
  13     dic=json.load(open(‘db.txt‘,‘r‘,encoding=‘utf-8‘))
  14     if dic[‘count‘] > 0:
  15         dic[‘count‘]-=1
  16         time.sleep(3)
  17         json.dump(dic,open(‘db.txt‘,‘w‘,encoding=‘utf-8‘))
  18         print(‘<%s> 购票成功‘ %name)
  19     else:
  20         print(‘<%s> 购票失败‘ %name)
  21 
  22 def task(name,):
  23     search(name)
  24     # mutex.acquire()
  25     get(name)
  26     # mutex.release()
  27 
  28 if __name__ == ‘__main__‘:
  29     # mutex=Lock()
  30     for i in range(10):
  31         p=Process(target=task,args=(‘路人%s‘ %i,))
  32         p.start()
  33         p.join()

  互斥锁与join
  

   1 from multiprocessing import Queue
   2 
   3 
   4 q=Queue(3)
   5 
   6 q.put(‘hello‘)
   7 q.put({‘a‘:1})
   8 q.put([3,3,3,])
   9 print(q.full())
  10 
  11 # q.put(4)
  12 
  13 print(q.get())
  14 print(q.get())
  15 print(q.get())
  16 print(q.empty())
  17 
  18 print(q.get())

  队列使用
  

   1 from multiprocessing import Process,Queue
   2 import time
   3 
   4 def producer(q):
   5     for i in range(10):
   6         res=‘包子%s‘ %i
   7         time.sleep(0.5)
   8         print(‘生产者生产了%s‘ %res)
   9 
  10         q.put(res)
  11 
  12 def consumer(q):
  13     while True:
  14         res=q.get()
  15         if res is None:break
  16         time.sleep(1)
  17         print(‘消费者吃了%s‘ % res)
  18 
  19 
  20 
  21 if __name__ == ‘__main__‘:
  22     #容器
  23     q=Queue()
  24 
  25     #生产者们
  26     p1=Process(target=producer,args=(q,))
  27     p2=Process(target=producer,args=(q,))
  28     p3=Process(target=producer,args=(q,))
  29 
  30     #消费者们
  31     c1=Process(target=consumer,args=(q,))
  32     c2=Process(target=consumer,args=(q,))
  33 
  34     p1.start()
  35     p2.start()
  36     p3.start()
  37     c1.start()
  38     c2.start()
  39 
  40     p1.join()
  41     p2.join()
  42     p3.join()
  43     q.put(None)
  44     q.put(None)
  45     print(‘主‘)

  生产消费模型
  

   1 from multiprocessing import Process,JoinableQueue
   2 import time
   3 
   4 def producer(q):
   5     for i in range(2):
   6         res=‘包子%s‘ %i
   7         time.sleep(0.5)
   8         print(‘生产者生产了%s‘ %res)
   9 
  10         q.put(res)
  11     q.join()
  12 
  13 def consumer(q):
  14     while True:
  15         res=q.get()
  16         if res is None:break
  17         time.sleep(1)
  18         print(‘消费者吃了%s‘ % res)
  19         q.task_done()
  20 
  21 
  22 if __name__ == ‘__main__‘:
  23     #容器
  24     q=JoinableQueue()
  25 
  26     #生产者们
  27     p1=Process(target=producer,args=(q,))
  28     p2=Process(target=producer,args=(q,))
  29     p3=Process(target=producer,args=(q,))
  30 
  31     #消费者们
  32     c1=Process(target=consumer,args=(q,))
  33     c2=Process(target=consumer,args=(q,))
  34     c1.daemon=True
  35     c2.daemon=True
  36 
  37     p1.start()
  38     p2.start()
  39     p3.start()
  40     c1.start()
  41     c2.start()
  42 
  43 
  44     p1.join()
  45     p2.join()
  46     p3.join()
  47     print(‘主‘)

  JoinableQueue

   

• 多线程
  
  

   1 # import time
   2 # import random
   3 # from threading import Thread
   4 #
   5 # def piao(name):
   6 #     print(‘%s piaoing‘ %name)
   7 #     time.sleep(random.randrange(1,5))
   8 #     print(‘%s piao end‘ %name)
   9 #
  10 # if __name__ == ‘__main__‘:
  11 #     t1=Thread(target=piao,args=(‘egon‘,))
  12 #     t1.start()
  13 #     print(‘主线程‘)
  14 #
  15 
  16 
  17 import time
  18 import random
  19 from threading import Thread
  20 
  21 class MyThread(Thread):
  22     def __init__(self,name):
  23         super().__init__()
  24         self.name=name
  25 
  26     def run(self):
  27         print(‘%s piaoing‘ %self.name)
  28 
  29         time.sleep(random.randrange(1,5))
  30         print(‘%s piao end‘ %self.name)
  31 
  32 if __name__ == ‘__main__‘:
  33     t1=MyThread(‘egon‘)
  34     t1.start()
  35     print(‘主‘)

  开启线程的2种方式
  

   1 # 1、开进程的开销远大于开线程
   2 # import time
   3 # from threading import Thread
   4 # from multiprocessing import Process
   5 #
   6 # def piao(name):
   7 #     print(‘%s piaoing‘ %name)
   8 #     time.sleep(2)
   9 #     print(‘%s piao end‘ %name)
  10 #
  11 # if __name__ == ‘__main__‘:
  12 #     # p1=Process(target=piao,args=(‘egon‘,))
  13 #     # p1.start()
  14 #
  15 #     t1=Thread(target=piao,args=(‘egon‘,))
  16 #     t1.start()
  17 #     print(‘主线程‘)
  18 
  19 
  20 
  21 # 2、同一进程内的多个线程共享该进程的地址空间
  22 # from threading import Thread
  23 # from multiprocessing import Process
  24 #
  25 # n=100
  26 # def task():
  27 #     global n
  28 #     n=0
  29 #
  30 # if __name__ == ‘__main__‘:
  31 #     # p1=Process(target=task,)
  32 #     # p1.start()
  33 #     # p1.join()
  34 #
  35 #     t1=Thread(target=task,)
  36 #     t1.start()
  37 #     t1.join()
  38 #
  39 #     print(‘主线程‘,n)
  40 
  41 
  42 # 3、瞅一眼pid
  43 # from threading import Thread
  44 # from multiprocessing import Process,current_process
  45 # import os
  46 #
  47 # def task():
  48 #     # print(current_process().pid)
  49 #     print(‘子进程PID:%s  父进程的PID:%s‘ %(os.getpid(),os.getppid()))
  50 #
  51 # if __name__ == ‘__main__‘:
  52 #     p1=Process(target=task,)
  53 #     p1.start()
  54 #
  55 #     # print(‘主线程‘,current_process().pid)
  56 #     print(‘主线程‘,os.getpid())
  57 
  58 
  59 from threading import Thread
  60 import os
  61 
  62 def task():
  63     print(‘子线程:%s‘ %(os.getpid()))
  64 
  65 if __name__ == ‘__main__‘:
  66     t1=Thread(target=task,)
  67     t1.start()
  68 
  69     print(‘主线程‘,os.getpid())

  进程与线程的区别
  

   1 from threading import Thread,currentThread,active_count,enumerate
   2 import time
   3 
   4 def task():
   5     print(‘%s is ruuning‘ %currentThread().getName())
   6     time.sleep(2)
   7     print(‘%s is done‘ %currentThread().getName())
   8 
   9 if __name__ == ‘__main__‘:
  10     t=Thread(target=task,name=‘子线程1‘)
  11     t.start()
  12     # t.setName(‘儿子线程1‘)
  13     # t.join()
  14     # print(t.getName())
  15     # currentThread().setName(‘主线程‘)
  16     # print(t.isAlive())
  17 
  18 
  19     # print(‘主线程‘,currentThread().getName())
  20 
  21     # t.join()
  22     # print(active_count())
  23     print(enumerate())

  Thread对象的属性和方法
  

   1 # from threading import Thread
   2 # import time
   3 #
   4 # def sayhi(name):
   5 #     time.sleep(2)
   6 #     print(‘%s say hello‘ %name)
   7 #
   8 # if __name__ == ‘__main__‘:
   9 #     t=Thread(target=sayhi,args=(‘egon‘,))
  10 #     # t.setDaemon(True) #必须在t.start()之前设置
  11 #     t.daemon=True
  12 #     t.start()
  13 #
  14 #     print(‘主线程‘)
  15 #     print(t.is_alive())
  16 #
  17 
  18 
  19 from threading import Thread
  20 import time
  21 
  22 def foo():
  23     print(123)
  24     time.sleep(1)
  25     print("end123")
  26 
  27 def bar():
  28     print(456)
  29     time.sleep(3)
  30     print("end456")
  31 
  32 if __name__ == ‘__main__‘:
  33     t1=Thread(target=foo)
  34     t2=Thread(target=bar)
  35 
  36     t1.daemon=True
  37     t1.start()
  38     t2.start()
  39     print("main-------")
  40 
  41 
  42     ‘‘‘
  43     123
  44     456
  45     main-------
  46     end123
  47     end456
  48     
  49     ‘‘‘

  守护线程
  

   1 #mutex
   2 from threading import Thread,Lock
   3 import time
   4 
   5 n=100
   6 
   7 def task():
   8     global n
   9     mutex.acquire()
  10     temp=n
  11     time.sleep(0.1)
  12     n=temp-1
  13     mutex.release()
  14     
  15 if __name__ == ‘__main__‘:
  16     mutex=Lock()
  17     t_l=[]
  18     for i in range(100):
  19         t=Thread(target=task)
  20         t_l.append(t)
  21         t.start()
  22 
  23     for t in t_l:
  24         t.join()
  25 
  26     print(‘主‘,n)

  互斥锁
  

   1 # 计算密集型：用多进程
   2 # from multiprocessing import Process
   3 # from threading import Thread
   4 # import os,time
   5 # def work():
   6 #     res=0
   7 #     for i in range(100000000):
   8 #         res*=i
   9 #
  10 #
  11 # if __name__ == ‘__main__‘:
  12 #     l=[]
  13 #     # print(os.cpu_count()) #本机为8核
  14 #     start=time.time()
  15 #     for i in range(8):
  16 #         # p=Process(target=work) #耗时8s多
  17 #         p=Thread(target=work) #耗时37s多
  18 #         l.append(p)
  19 #         p.start()
  20 #     for p in l:
  21 #         p.join()
  22 #     stop=time.time()
  23 #     print(‘run time is %s‘ %(stop-start))
  24 
  25 
  26 # IO密集型：用多线程
  27 from multiprocessing import Process
  28 from threading import Thread
  29 import threading
  30 import os,time
  31 
  32 def work():
  33     time.sleep(2)
  34 
  35 if __name__ == ‘__main__‘:
  36     l=[]
  37     # print(os.cpu_count()) #本机为4核
  38     start=time.time()
  39     for i in range(400):
  40         # p=Process(target=work) #耗时2.697多,大部分时间耗费在创建进程上
  41         p=Thread(target=work) #耗时2.02多
  42         l.append(p)
  43         p.start()
  44     for p in l:
  45         p.join()
  46     stop=time.time()
  47     print(‘run time is %s‘ %(stop-start))

  GIL解释器锁
  

   1 # 死锁
   2 # from threading import Thread,Lock
   3 # import time
   4 #
   5 # mutexA=Lock()
   6 # mutexB=Lock()
   7 #
   8 # class MyThread(Thread):
   9 #     def run(self):
  10 #         self.f1()
  11 #         self.f2()
  12 #
  13 #     def f1(self):
  14 #         mutexA.acquire()
  15 #         print(‘%s 拿到了A锁‘ %self.name)
  16 #
  17 #         mutexB.acquire()
  18 #         print(‘%s 拿到了B锁‘ %self.name)
  19 #         mutexB.release()
  20 #
  21 #         mutexA.release()
  22 #
  23 #
  24 #     def f2(self):
  25 #         mutexB.acquire()
  26 #         print(‘%s 拿到了B锁‘ % self.name)
  27 #         time.sleep(0.1)
  28 #
  29 #         mutexA.acquire()
  30 #         print(‘%s 拿到了A锁‘ % self.name)
  31 #         mutexA.release()
  32 #
  33 #         mutexB.release()
  34 #
  35 # if __name__ == ‘__main__‘:
  36 #     for i in range(10):
  37 #         t=MyThread()
  38 #         t.start()
  39 
  40 
  41 # 互斥锁只能acquire一次
  42 # from threading import Thread,Lock
  43 #
  44 # mutexA=Lock()
  45 #
  46 # mutexA.acquire()
  47 # mutexA.release()
  48 
  49 
  50 # 递归锁:可以连续acquire多次，每acquire一次计数器+1，只有计数为0时，才能被抢到acquire
  51 from threading import Thread,RLock
  52 import time
  53 
  54 mutexB=mutexA=RLock()
  55 
  56 class MyThread(Thread):
  57     def run(self):
  58         self.f1()
  59         self.f2()
  60 
  61     def f1(self):
  62         mutexA.acquire()
  63         print(‘%s 拿到了A锁‘ %self.name)
  64 
  65         mutexB.acquire()
  66         print(‘%s 拿到了B锁‘ %self.name)
  67         mutexB.release()
  68 
  69         mutexA.release()
  70 
  71 
  72     def f2(self):
  73         mutexB.acquire()
  74         print(‘%s 拿到了B锁‘ % self.name)
  75         time.sleep(7)
  76 
  77         mutexA.acquire()
  78         print(‘%s 拿到了A锁‘ % self.name)
  79         mutexA.release()
  80 
  81         mutexB.release()
  82 
  83 if __name__ == ‘__main__‘:
  84     for i in range(10):
  85         t=MyThread()
  86         t.start()

  死锁和递归锁
  

   1 from threading import Thread,Semaphore,currentThread
   2 import time,random
   3 
   4 sm=Semaphore(3)
   5 
   6 def task():
   7     # sm.acquire()
   8     # print(‘%s in‘ %currentThread().getName())
   9     # sm.release()
  10     with sm:
  11         print(‘%s in‘ %currentThread().getName())
  12         time.sleep(random.randint(1,3))
  13 
  14 
  15 if __name__ == ‘__main__‘:
  16     for i in range(10):
  17         t=Thread(target=task)
  18         t.start()

  信号量
  

   1 # from threading import Thread,Event
   2 # import time
   3 #
   4 # event=Event()
   5 # # event.wait()
   6 # # event.set()
   7 #
   8 #
   9 # def student(name):
  10 #     print(‘学生%s 正在听课‘ %name)
  11 #     event.wait(2)
  12 #     print(‘学生%s 课间活动‘ %name)
  13 #
  14 #
  15 # def teacher(name):
  16 #     print(‘老师%s 正在授课‘ %name)
  17 #     time.sleep(7)
  18 #     event.set()
  19 #
  20 #
  21 # if __name__ == ‘__main__‘:
  22 #     stu1=Thread(target=student,args=(‘alex‘,))
  23 #     stu2=Thread(target=student,args=(‘wxx‘,))
  24 #     stu3=Thread(target=student,args=(‘yxx‘,))
  25 #     t1=Thread(target=teacher,args=(‘egon‘,))
  26 #
  27 #     stu1.start()
  28 #     stu2.start()
  29 #     stu3.start()
  30 #     t1.start()
  31 
  32 
  33 
  34 from threading import Thread,Event,currentThread
  35 import time
  36 
  37 event=Event()
  38 
  39 def conn():
  40     n=0
  41     while not event.is_set():
  42         if n == 3:
  43             print(‘%s try too many times‘ %currentThread().getName())
  44             return
  45         print(‘%s try %s‘ %(currentThread().getName(),n))
  46         event.wait(0.5)
  47         n+=1
  48 
  49     print(‘%s is connected‘ %currentThread().getName())
  50 
  51 
  52 def check():
  53     print(‘%s is checking‘ %currentThread().getName())
  54     time.sleep(5)
  55     event.set()
  56 
  57 
  58 if __name__ == ‘__main__‘:
  59     for i in range(3):
  60         t=Thread(target=conn)
  61         t.start()
  62     t=Thread(target=check)
  63     t.start()

  Event事件
  

   1 # from threading import Timer
   2 #
   3 # def task(name):
   4 #     print(‘hello %s‘ %name)
   5 #
   6 #
   7 # t=Timer(5,task,args=(‘egon‘,))
   8 # t.start()
   9 
  10 from threading import Timer
  11 import random
  12 
  13 class Code:
  14     def __init__(self):
  15         self.make_cache()
  16 
  17     def make_cache(self,interval=5):
  18         self.cache=self.make_code()
  19         print(self.cache)
  20         self.t=Timer(interval,self.make_cache)
  21         self.t.start()
  22 
  23     def make_code(self,n=4):
  24         res=‘‘
  25         for i in range(n):
  26             s1=str(random.randint(0,9))
  27             s2=chr(random.randint(65,90))
  28             res+=random.choice([s1,s2])
  29         return res
  30 
  31     def check(self):
  32         while True:
  33             code=input(‘请输入你的验证码>>: ‘).strip()
  34             if code.upper() == self.cache:
  35                 print(‘验证码输入正确‘)
  36                 self.t.cancel()
  37                 break
  38 
  39 
  40 obj=Code()
  41 obj.check()

  定时器
  

   1 import queue
   2 
   3 # q=queue.Queue(3) #先进先出->队列
   4 #
   5 # q.put(‘first‘)
   6 # q.put(2)
   7 # q.put(‘third‘)
   8 # # q.put(4)
   9 # # q.put(4,block=False) #q.put_nowait(4)
  10 # # q.put(4,block=True,timeout=3)
  11 #
  12 #
  13 # #
  14 # print(q.get())
  15 # print(q.get())
  16 # print(q.get())
  17 # # print(q.get(block=False)) #q.get_nowait()
  18 # # print(q.get_nowait())
  19 #
  20 # # print(q.get(block=True,timeout=3))
  21 
  22 
  23 # q=queue.LifoQueue(3) #后进先出->堆栈
  24 # q.put(‘first‘)
  25 # q.put(2)
  26 # q.put(‘third‘)
  27 #
  28 # print(q.get())
  29 # print(q.get())
  30 # print(q.get())
  31 
  32 #
  33 # q=queue.PriorityQueue(3) #优先级队列
  34 #
  35 # q.put((10,‘one‘))
  36 # q.put((40,‘two‘))
  37 # q.put((30,‘three‘))
  38 #
  39 # print(q.get())
  40 # print(q.get())
  41 # print(q.get())

  线程queue

   

• 线程池&进程池
  
  线程池与进程池的开启
  
  

   1 #提交任务的两种方式
   2 #1、同步调用:提交完任务后，就在原地等待任务执行完毕，拿到结果，再执行下一行代码,导致程序是串行执行
   3 #
   4 # from concurrent.futures import ThreadPoolExecutor
   5 # import time
   6 # import random
   7 #
   8 # def la(name):
   9 #     print(‘%s is laing‘ %name)
  10 #     time.sleep(random.randint(3,5))
  11 #     res=random.randint(7,13)*‘#‘
  12 #     return {‘name‘:name,‘res‘:res}
  13 #
  14 # def weigh(shit):
  15 #     name=shit[‘name‘]
  16 #     size=len(shit[‘res‘])
  17 #     print(‘%s 拉了 《%s》kg‘ %(name,size))
  18 #
  19 #
  20 # if __name__ == ‘__main__‘:
  21 #     pool=ThreadPoolExecutor(13)
  22 #
  23 #     shit1=pool.submit(la,‘alex‘).result()
  24 #     weigh(shit1)
  25 #
  26 #     shit2=pool.submit(la,‘wupeiqi‘).result()
  27 #     weigh(shit2)
  28 #
  29 #     shit3=pool.submit(la,‘yuanhao‘).result()
  30 #     weigh(shit3)
  31 
  32 
  33 #2、异步调用：提交完任务后，不地等待任务执行完毕，
  34 
  35 from concurrent.futures import ThreadPoolExecutor
  36 import time
  37 import random
  38 
  39 def la(name):
  40     print(‘%s is laing‘ %name)
  41     time.sleep(random.randint(3,5))
  42     res=random.randint(7,13)*‘#‘
  43     return {‘name‘:name,‘res‘:res}
  44 
  45 
  46 def weigh(shit):
  47     shit=shit.result()
  48     name=shit[‘name‘]
  49     size=len(shit[‘res‘])
  50     print(‘%s 拉了 《%s》kg‘ %(name,size))
  51 
  52 
  53 if __name__ == ‘__main__‘:
  54     pool=ThreadPoolExecutor(13)
  55 
  56     pool.submit(la,‘alex‘).add_done_callback(weigh)
  57 
  58     pool.submit(la,‘wupeiqi‘).add_done_callback(weigh)
  59 
  60     pool.submit(la,‘yuanhao‘).add_done_callback(weigh)

  异步调用&回调机制
  

   1 from concurrent.futures import ThreadPoolExecutor
   2 import requests
   3 import time
   4 
   5 def get(url):
   6     print(‘GET %s‘ %url)
   7     response=requests.get(url)
   8     time.sleep(3)
   9     return {‘url‘:url,‘content‘:response.text}
  10 
  11 
  12 def parse(res):
  13     res=res.result()
  14     print(‘%s parse res is %s‘ %(res[‘url‘],len(res[‘content‘])))
  15 
  16 
  17 if __name__ == ‘__main__‘:
  18     urls=[
  19         ‘http://www.cnblogs.com/linhaifeng‘,
  20         ‘https://www.python.org‘,
  21         ‘https://www.openstack.org‘,
  22     ]
  23 
  24     pool=ThreadPoolExecutor(2)
  25 
  26     for url in urls:
  27         pool.submit(get,url).add_done_callback(parse)

  线程池 示例
• 协程
  
  

   1 #并发执行
   2 import time
   3 
   4 def producer():
   5     g=consumer()
   6     next(g)
   7     for i in range(10000000):
   8         g.send(i)
   9 
  10 
  11 
  12 def consumer():
  13     while True:
  14         res=yield
  15 
  16 
  17 start_time=time.time()
  18 producer()
  19 stop_time=time.time()
  20 print(stop_time-start_time)
  21 
  22 
  23 
  24 #串行
  25 import time
  26 
  27 def producer():
  28     res=[]
  29     for i in range(10000000):
  30         res.append(i)
  31     return res
  32 
  33 
  34 def consumer(res):
  35     pass
  36 
  37 
  38 start_time=time.time()
  39 res=producer()
  40 consumer(res)
  41 stop_time=time.time()
  42 print(stop_time-start_time)

  协程
  

   1 #pip3 install greenlet
   2 from greenlet import greenlet
   3 import time
   4 
   5 def eat(name):
   6     print(‘%s eat 1‘ %name)
   7     time.sleep(10)
   8     g2.switch(‘egon‘)
   9     print(‘%s eat 2‘ %name)
  10     g2.switch()
  11 
  12 def play(name):
  13     print(‘%s play 1‘ %name )
  14     g1.switch()
  15     print(‘%s play 2‘ %name )
  16 
  17 
  18 g1=greenlet(eat)
  19 g2=greenlet(play)
  20 
  21 g1.switch(‘egon‘)

  greenlet模块
  

   1 #pip3 install gevent
   2 # from gevent import monkey;monkey.patch_all()
   3 # import gevent
   4 # import time
   5 #
   6 #
   7 # def eat(name):
   8 #     print(‘%s eat 1‘ % name)
   9 #     time.sleep(3)
  10 #     print(‘%s eat 2‘ % name)
  11 #
  12 #
  13 # def play(name):
  14 #     print(‘%s play 1‘ % name)
  15 #     time.sleep(4)
  16 #     print(‘%s play 2‘ % name)
  17 #
  18 #
  19 # start_time=time.time()
  20 # g1=gevent.spawn(eat,‘egon‘)
  21 # g2=gevent.spawn(play,‘alex‘)
  22 #
  23 #
  24 # g1.join()
  25 # g2.join()
  26 # stop_time=time.time()
  27 # print(stop_time-start_time)
  28 
  29 
  30 from gevent import monkey;monkey.patch_all()
  31 import gevent
  32 import time
  33 
  34 
  35 def eat(name):
  36     print(‘%s eat 1‘ % name)
  37     time.sleep(3)
  38     print(‘%s eat 2‘ % name)
  39 
  40 
  41 def play(name):
  42     print(‘%s play 1‘ % name)
  43     time.sleep(4)
  44     print(‘%s play 2‘ % name)
  45 
  46 
  47 g1=gevent.spawn(eat,‘egon‘)
  48 g2=gevent.spawn(play,‘alex‘)
  49 
  50 
  51 # time.sleep(5)
  52 
  53 
  54 # g1.join()
  55 # g2.join()
  56 
  57 gevent.joinall([g1,g2])

  gevent模块（自动切换）

   

第五篇 进程&线程&协程

标签：bar   lap   producer   inpu   sub   操作   close   生产者   gre   

原文地址：https://www.cnblogs.com/huahuawang/p/14725519.html