【进程池与线程池 协程】

本日内容概要 - 2024.5.31

---

• 死锁与递归锁（了解）
  
• 信号量（了解）
  
• Event事件（了解）
  
• 线程q（了解）
  
• 进程池与线程池（掌握）
  
• 协程（了解）
  
• 协程实现TCP服务端的并发效果（了解）
  

本日内容具体

---

死锁与递归锁
当你知道锁的使用枪锁必须要释放锁，其实你在操作锁的时候也极其轻易产生死锁现象。

1. from threading import Thread, Lock
2. import time
4. mutexA = Lock()
5. mutexB = Lock()
6. # 类被实例化多次，产生的对象肯定是不相等的
7. # 如果要实现实例化多次类，产生的对象是相等的 只能是单利模式
10. class MyThread(Thread):
11.     def run(self):
12.         self.func1()
13.         self.func2()
15.     def func1(self):
16.         mutexA.acquire()
17.         print('%s 抢到了锁A锁' % self.name) # 当前线程的名称
18.         mutexB.acquire()
19.         print('%s 抢到了锁B锁' % self.name)
20.         mutexB.release()
21.         mutexA.release()
23.     def func2(self):
24.         mutexB.acquire()
25.         print('%s 抢到了锁B锁' % self.name)
26.         time.sleep(2)
27.         mutexA.acquire()
28.         print('%s 抢到了锁A锁' % self.name)
29.         mutexA.release()
30.         mutexB.release()
33. if __name__ == '__main__':
34.     for i in range(10):
35.         t = MyThread()
36.         t.start()

复制代码

递归锁

1. """
2. 递归锁的特点
3.         可以被连续的acquire和release
4.         内部有一个计数器 每acquire一次计数加1 没release一次计数器减一
5. """

复制代码

信号量
信号量在不同阶段大概对应不同的技能点
在并发编程中信号量指的就是锁！！

1. """
2. 如果我们将互斥锁比喻成一个厕所
3. 那么信号量就是公共厕所
4. """
5. from threading import Thread, Semaphore
6. import time
7. import random
9. sm = Semaphore(5)  # 数字写几就代表几个坑位
12. def task(name):
13.     sm.acquire()
14.     print('%s正在蹲坑' % name)
15.     time.sleep(random.randint(1, 3))
16.     sm.release()
19. if __name__ == '__main__':
20.     for i in range(20):
21.         t = Thread(target=task, args=('伞兵%s号' % i,))
22.         t.start()

复制代码

Event事件
一些进程/线程必要期待另外一些进程/线程完毕之后才能运行，类似于发射信号一样

1. from threading import Thread, Event
2. import time
4. event = Event()
7. def light():
8.     print('红灯开始了')
9.     time.sleep(3)
10.     print('红灯结束了')
11.     event.set()  # 告诉等待红灯的人可以走了
14. def car(name):
15.     print('%s 车正在等红灯' % name)
16.     event.wait()  # 等待别人给你发信号
17.     print('%s 车油门飙走了' % name)
20. if __name__ == '__main__':
21.     t = Thread(target=light)
22.     t.start()
24.     for i in range(1, 11):
25.         t = Thread(target=car, args=('%s' % i,))
26.         t.start()

复制代码

线程q(了解)

1. """
2. 同一个进程多个线程数据是共享的
3. 为什么同一个进程还要使用队列
4. 因为队列是： 管道 + 锁
5. 所以队列还是为了保证数据的安全
6. """
7. import queue
9. # 1 队列q 先进先出
10. # q = queue.Queue()
11. # q.put(1)
12. # q.get()
13. # q.get_nowait()
14. # q.full()
15. # q.empty()
16. # print(q.get())
18. # 2 堆栈q 后进先出
19. # q = queue.LifoQueue(3)
20. # q.put(2)
21. # q.put(3)
22. # print(q.get()) # 3
24. # 优先级q 你可以给放入队列中的数据设置进出优先级
25. q = queue.PriorityQueue(4)
26. q.put((10, '11'))
27. q.put((100, '22'))
28. q.put((0, '33'))
29. q.put((-5, '44'))
30. print(q.get())  # (-5, '44')
31. # 第一个数字代表优先级 数字越小，优先级越高

复制代码

进程池与线程池（掌握）
先回顾一下吧，之前TCP服务端实现并发的效果，每来一个人就开设一个进程大概线程行止理。

1. """
2. 无论是开设进程还是开设线程，都需要消耗资源，只不过开设线程的消耗比开设进程的消耗稍微小一点而已。
3. 我们不可能无限制的开设进程和线程，因为计算机的硬件资源跟不上。
5. 我们的宗旨保证计算机硬件正常工作的情况下最大限度的利用它。
6. """
7. # 池的概念
8. """
9. 什么是池？
10.         池是用来保障计算机硬件安全的情况下最大限度的利用计算机
11.         它降低了程序的运行效果但是保证了计算机硬件的安全，从而让程序能够正常运行。
12. """

复制代码

基本操作

1. from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
2. import time
3. import os
5. # pool = ThreadPoolExecutor(5) # 池子里面固定只有五个线程
6. # 括号内可以传数字 不穿的话默认会开设当前计算机cpu个数 + 4 与 32 之间取最小值
7. pool = ProcessPoolExecutor()
8. # 括号内可以传数字 不传的话默认会开设当前计算机CPU个数进程
9. """
10. 池子造出来以后 里面会固定存在5个线程
11. 这五个线程不会出现重复创建和销毁的过程
12. """
15. def task(n):
16.     print(n,os.getpid())
17.     time.sleep(2)
18.     return n * n
20. def callback(n):
21.     print('call_back>>>:',n.result())
23. """
24. 提交任务
25.     同步：提交任务之后原地等待返回结果，期间不做任何操作
26.     异步：提交任务之后不原地等待返回结果，继续往下执行
27.         返回结果如何获取？？？
28.         异步提交任务的返回结果 应该通过回调机制来获取
29.         回调机制
30.             就相当于给每个异步任务绑定了一个定时炸弹
31.             一旦该任务有结果立刻触发爆炸
32. """
33. if __name__ == '__main__':
34.     # pool.submit(task, 1)  # 向池子中提交任务  异步提交
35.     # print('主')
36.     t_list = []
37.     for i in range(20):
38.         res = pool.submit(task, i).add_done_callback(callback)  # <Future at 0x286a8ade310 state=running>
39.         # print(res.result())  # result()方法 同步提交
40.         # t_list.append(res)
41.     # 等待线程池中所有的任务执行完毕之后再继续往下执行
42.     # pool.shutdown()  # 关闭线程池 等待线程池中所有任务运行完毕
43.     # for t in t_list:
44.     #     print('>>>:', t.result())
46.     """
47.     程序由并发变成了串行
48.     res.result() 拿到的是异步提交任务的返回结果
49.     """

复制代码

总结

1. from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
2. pool = ProcessPoolExecutor()
3. res = pool.submit(task, i).add_done_callback(callback)

复制代码

协程

1. """
2. 进程：资源单位
3. 线程：执行单位
4. 协程：这个概念完全是程序员自己想出来的 根本不存在
5.         单线程下实现并发
6.         程序员自己在代码层面上检查我们所有遇到的IO操作
7.         一旦遇到IO了 我们在代码级别完成切换
8.         这样给CPU的感觉是你这个程序一直在运行 没有IO
9.         从而提升程序的运行效率
10.        
11. 多道技术
12.         切换+保存状态
13.         CPU两种切换
14.                 1.程序遇到IO
15.                 2.程序长时间占用
16.                
17. TCP服务器
18.         accept
19.         recv
20.        
21. 代码如何做到
22.         切换+保存状态
24. 切换
25.         切换不一定是提升效率 也有可能是降低效率
26.         IO切    提升
27.         没有IO切 降低
28.        
29. 保存状态
30.         保存上一次执行的状态 下一次接着上一次的操作继续往后执行
31.         yield
32. """

复制代码

严重切换是否就肯定提升效率

1. import time
3. # 串行执行计算机密集型的任务 0.17453885078430176
4. # def func1():
5. #     for i in range(1000000):
6. #         i + 1
7. #
8. #
9. # def func2():
10. #     for i in range(1000000):
11. #         i + 1
12. # start_time = time.time()
13. # func1()
14. # func2()
15. # print(time.time() - start_time)
18. # 切换 + yield
19. import time
22. def func1():
23.     while True:
24.         100000 + 1
25.         yield
28. def func2():
29.     g = func1()
30.     for i in range(100000):
31.         i + 1
32.         next(g)
35. start_time = time.time()
36. # func1()
37. func2()
38. print(time.time() - start_time) # 0.012012958526611328

复制代码

gevent模块

1. from gevent import monkey;monkey.patch_all()
2. # monkey.patch_all()
3. import time
4. from gevent import spawn
5. """
6. gevent模块本身无法检查常见的一些IO操作
7. 在使用时需额外的导入一句话，可支持简写
8. """
11. def heng():
12.     print('哼')
13.     time.sleep(2)
14.     print('哼')
17. def ha():
18.     print('哈')
19.     time.sleep(3)
20.     print('哈')
22. def heihei():
23.     print('hahaha')
24.     time.sleep(5)
25.     print('hahahha')
27. start_time = time.time()
28. g1 = spawn(heng)
29. g2 = spawn(ha)
30. g3 = spawn(heihei)
31. g1.join() # 等待被检测的任务执行完毕 再往后继续执行 3.0827553272247314
32. g2.join()
33. g3.join()
34. # heng()
35. # ha()
36. print(time.time() - start_time)  # 5.043513774871826

复制代码

协程实现TCP服务端的并发

1. from gevent import monkey;monkey.patch_all()
2. from gevent import spawn
3. import socket
6. def connection(conn):
7.     while True:
8.         try:
9.             data = conn.recv(1024)
10.             if len(data) == 0: break
11.             conn.send(data.upper())
12.         except ConnectionError as e:
13.             print(e)
14.             break
15.     conn.close()
18. def server(ip, port):
19.     server = socket.socket()
20.     server.bind((ip, port))
21.     server.listen(5)
22.     while True:
23.         conn, addr = server.accept()
24.         spawn(connection, conn)
27. if __name__ == '__main__':
28.     g = spawn(server, '127.0.0.1', 8080)
29.     g.join()
31.    
32.    
33.    
34. # 客服端
35. import socket
36. from threading import Thread, current_thread
39. def x_client():
40.     client = socket.socket()
41.     client.connect(('127.0.0.1', 8080))
42.     n = 0
43.     while True:
44.         msg = '%s say hello %s' % (current_thread().name, n)
45.         n += 1
46.         client.send(msg.encode('utf-8'))
47.         data = client.recv(1024)
48.         print(data.decode('utf-8'))
51. if __name__ == '__main__':
52.     for i in range(500):
53.         t = Thread(target=x_client)
54.         t.start()

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。