Python入门day41——进程线程高阶
作者：互联网
1.死锁 与 递归锁（了解）
死锁现象
from threading import Thread
from threading import Lock
from threading import RLock # 递归锁
import time
mutexA = Lock()
mutexB = Lock()
# print(mutexA is mutexB) # False
# 类 只要加括号多次 产生的肯定是不同的对象
# 如果你想要实现多次加括号得到的是相同的对象 要用到单例模式
class MyThread(Thread):
def run(self):
self.func1()
self.func2()
def func1(self):
mutexA.acquire()
print(f'{self.name} 抢到了 func1的A锁') # 获取当前线程名
mutexB.acquire()
print(f'{self.name} 抢到了 func1的B锁') # 获取当前线程名
mutexB.release()
mutexA.release()
def func2(self):
mutexB.acquire()
print(f'{self.name} 抢到了 func2的B锁') # 获取当前线程名
time.sleep(2)
mutexA.acquire()
print(f'{self.name} 抢到了 func2的A锁') # 获取当前线程名
mutexA.release()
mutexB.release()
if __name__ == '__main__':
for i in range(10):
t = MyThread()
t.start()
# Thread-1 抢到了 func1的A锁
# Thread-1 抢到了 func1的B锁
# Thread-1 抢到了 func2的B锁
# Thread-2 抢到了 func1的A锁
递归锁（了解）
递归锁特点：
可以被连续地acquir和release
但是只能被第1个抢到的这把锁 执行上述操作
它的内部有1个计数器 每acquire 1次 计数+1， 每release 1次 计数—1
只要计数不为0 ，那么其他人都无法抢到该锁
from threading import Thread
from threading import RLock # 递归锁
import time
mutexA = mutexB = RLock()
class MyThread(Thread):
def run(self):
self.func1()
self.func2()
def func1(self):
mutexA.acquire()
print(f'{self.name} 抢到了 func1的A锁') # 获取当前线程名
mutexB.acquire()
print(f'{self.name} 抢到了 func1的B锁') # 获取当前线程名
mutexB.release()
mutexA.release()
def func2(self):
mutexB.acquire()
print(f'{self.name} 抢到了 func2的B锁') # 获取当前线程名
time.sleep(2)
mutexA.acquire()
print(f'{self.name} 抢到了 func2的A锁') # 获取当前线程名
mutexA.release()
mutexB.release()
if __name__ == '__main__':
for i in range(10):
t = MyThread()
t.start()
# Thread-1 抢到了 func1的A锁
# Thread-1 抢到了 func1的B锁
# Thread-1 抢到了 func2的B锁
# Thread-1 抢到了 func2的A锁
# Thread-2 抢到了 func1的A锁
# Thread-2 抢到了 func1的B锁
# Thread-2 抢到了 func2的B锁
# Thread-2 抢到了 func2的A锁
# Thread-4 抢到了 func1的A锁
# ......
2.信号量（了解）
信号量 在不同的阶段 可能对应不同的技术点
在并发编程中 信号量 指的是 锁！！！
如果我们将互斥锁 比喻成 一个厕所的话（家庭厕所）
那么 信号量 就相当于 多个厕所（公共厕所）
代码实现：
from threading import Thread
from threading import Semaphore
import time
import random
'''
利用random模块实现打印随机验证码（搜狗的一道笔试题）
'''
sm = Semaphore(5) # 括号内写数字 写几 就表示开设几个坑位
def task(name):
sm.acquire()
print(f'{name} 正在蹲坑')
# time.sleep(3)
time.sleep(random.randint(1, 5))
sm.release()
if __name__ == '__main__':
for i in range(20):
t = Thread(target=task, args=(f'伞兵{i}号',))
t.start()
3.Event事件（了解）
一些进程/线程 需要等待另外一些进程/线程运行完毕之后 才能运行，类似于发射信号一样。
from threading import Thread
from threading import Event
import time
event = Event() # 造了一个红绿灯
def light():
print('红灯亮')
time.sleep(3)
print('绿灯亮')
# 告诉等待红灯的人 可以走了
event.set()
def car(name):
print(f'{name} 正在等红灯')
event.wait() # 等待别人给你发消息
print(f'{name} 加油门 飙车走了')
if __name__ == '__main__':
t = Thread(target=light)
t.start()
for i in range(20):
# t = Thread(target=car, args=(f'{i}'))
t = Thread(target=car, args=('%s' % i))
t.start()
4.线程q（了解）
同一个进程下 多个线程数据是共享的
为什么同一个进程下 还会去使用队列呢？
因为：
队列 = 管道 + 锁
所以：
用队列还是为了保证数据的安全
'''
同一个进程下 多个线程数据是共享的
为什么同一个进程下 还会去使用队列呢？
因为：
队列 = 管道 + 锁
所以：
用队列还是为了保证数据的安全
'''
import queue
# 我们现在使用的队列 都是只能在本地测试使用的
# 1.队列q 先进先出
# q = queue.Queue(3)
# q.put(1)
# q.get()
# q.get_nowait()
# q.get(timeout=3)
# q.full()
# q.empty()
# 2.后进先出q
# q = queue.LifoQueue(3) # Last in first out
# q.put(1)
# q.put(2)
# q.put(3)
# print(q.get()) # 3
# 3.优先级q 你可以给放入队列中的数据 设置进出的优先级
q = queue.PriorityQueue(4)
q.put((10, '111')) # (优先级, '数据')
q.put((101, '222')) # (优先级, '数据')
q.put((0, '333')) # (优先级, '数据')
q.put((1, '444')) # (优先级, '数据')
print(q.get()) # (0, '333')
# put括号内放一个元组 第一个数字表示优先级
# 注意：数字越小 优先级越高！！！
5.进程池 与 线程池（掌握）
先回顾之前TCP服务端实现并发的效果是什么玩儿的
每来一个人 就要开设一个进程 或者 线程 去处理
无论是开设进程也好 还是开设线程也好 是不是都需要消耗资源
只不过 开设线程的消耗 比 开设进程的稍微小一点而已
我们是不可能做到无限制的开设进程和线程的 因为计算机硬件的资源跟不上！！！
硬件的开发速度 远远赶不上软件
我们的宗旨 应该是：在保证计算机硬件 能够正常工作的情况下 最大限度地利用它
# 池的概念 == > 接待室
池 是用来保证计算机硬件安全的情况下 最大限度地利用计算机
它 降低了程序的运行效率 但是保证了计算机硬件的安全 从而让你写的程序能够正常运行
初始：
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time
# 线程池
pool = ThreadPoolExecutor(5) # 池子里面 固定只有5个线程
# 括号内可以传数字 不传的话 默认会开设当前计算机CPU核心数的 5倍的线程
'''
池子造出来之后 里面会固定存在5个线程
这5个线程 不会出现重复创建和销毁的过程
池子的使用 非常的简单
你只需要将需要做的任务 往池子中提交即可 自动会有人来服务你
'''
def task(n):
print(n)
time.sleep(2)
return n**n
'''
任务的提交方式：
同步：提交任务之后 原地等待任务的返回结果 期间不做任何事
异步：提交任务之后 不等待任务的返回结果 继续往下执行
返回结果 如何获取？？？
'''
# pool.submit(task, 1) # 朝池子中 提交任务 异步提交
# print('主') # 1主
for i in range(20): # 朝池子中 提交20个任务
res = pool.submit(task, i) # 0<Future at 0x32e8190 state=running>
print(res.result()) # result方法 同步提交 join方法
'''
程序 从 并发 变成了 串行
任务的结果 为什么打印的是None
打印的是任务的return返回值
print(res.result()) 拿到的就是异步提交的任务的返回结果
'''
进阶：
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time
# 线程池
pool = ThreadPoolExecutor(5) # 池子里面 固定只有5个线程
# 括号内可以传数字 不传的话 默认会开设当前计算机CPU核心数的 5倍的线程
'''
池子造出来之后 里面会固定存在5个线程
这5个线程 不会出现重复创建和销毁的过程
池子的使用 非常的简单
你只需要将需要做的任务 往池子中提交即可 自动会有人来服务你
'''
def task(n):
print(n)
time.sleep(2)
return n**n
'''
任务的提交方式：
同步：提交任务之后 原地等待任务的返回结果 期间不做任何事
异步：提交任务之后 不等待任务的返回结果 继续往下执行
返回结果 如何获取？？？
'''
# pool.submit(task, 1) # 朝池子中 提交任务 异步提交
# print('主') # 1主
t_list = []
for i in range(20): # 朝池子中 提交20个任务
res = pool.submit(task, i) # 0<Future at 0x32e8190 state=running>
# print(res.result()) # result方法 同步提交 join方法
t_list.append(res)
for t in t_list:
print('>>>>:', t.result()) # 肯定是有序的
'''
程序 从 并发 变成了 串行
任务的结果 为什么打印的是None
打印的是任务的return返回值
print(res.result()) 拿到的就是异步提交的任务的返回结果
'''
高阶：
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time
# 线程池
pool = ThreadPoolExecutor(5) # 池子里面 固定只有5个线程
# 括号内可以传数字 不传的话 默认会开设当前计算机CPU核心数的 5倍的线程
'''
池子造出来之后 里面会固定存在5个线程
这5个线程 不会出现重复创建和销毁的过程
池子的使用 非常的简单
你只需要将需要做的任务 往池子中提交即可 自动会有人来服务你
'''
def task(n):
print(n)
time.sleep(2)
return n**n
'''
任务的提交方式：
同步：提交任务之后 原地等待任务的返回结果 期间不做任何事
异步：提交任务之后 不等待任务的返回结果 继续往下执行
返回结果 如何获取？？？
'''
# pool.submit(task, 1) # 朝池子中 提交任务 异步提交
# print('主') # 1主
t_list = []
for i in range(20): # 朝池子中 提交20个任务
res = pool.submit(task, i) # 0<Future at 0x32e8190 state=running>
# print(res.result()) # result方法 同步提交 join方法
t_list.append(res)
# 等待线程池中 所有的任务 执行完毕之后 再继续往下执行
pool.shutdown() # 关闭线程池 等待线程池中所有的任务进行完毕
for t in t_list:
print('>>>>:', t.result()) # 肯定是有序的
'''
程序 从 并发 变成了 串行
任务的结果 为什么打印的是None
打印的是任务的return返回值
print(res.result()) 拿到的就是异步提交的任务的返回结果
'''
终极：
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time
import os
# 线程池
# pool = ThreadPoolExecutor(5) # 池子里面 固定只有5个线程
# 括号内可以传数字 不传的话 默认会开设当前计算机CPU核心数的 5倍的线程
pool = ProcessPoolExecutor(5)
# 括号内可以传数字 不传的话 默认会开设当前计算机CPU核心数的线程
'''
池子造出来之后 里面会固定存在5个线程
这5个线程 不会出现重复创建和销毁的过程
池子的使用 非常的简单
你只需要将需要做的任务 往池子中提交即可 自动会有人来服务你
'''
def task(n):
print(n)
time.sleep(2)
return n**n
def callback(n):
print('返回的是：', n)
'''
任务的提交方式：
同步：提交任务之后 原地等待任务的返回结果 期间不做任何事
异步：提交任务之后 不等待任务的返回结果 继续往下执行
返回结果 如何获取？？？
异步提交任务的返回结果 应该通过回调机制来获取
回调机制：
一旦该任务有结果 立刻触发爆炸
'''
# pool.submit(task, 1) # 朝池子中 提交任务 异步提交
# print('主') # 1主
if __name__ == '__main__':
t_list = []
for i in range(20): # 朝池子中 提交20个任务
# res = pool.submit(task, i) # 0<Future at 0x32e8190 state=running>
res = pool.submit(task, i).add_done_callback(callback)
# print(res.result()) # result方法 同步提交 join方法
t_list.append(res)
# # 等待线程池中 所有的任务 执行完毕之后 再继续往下执行
# pool.shutdown() # 关闭线程池 等待线程池中所有的任务进行完毕
#
# for t in t_list:
# print('>>>>:', t.result()) # 肯定是有序的
'''
程序 从 并发 变成了 串行
任务的结果 为什么打印的是None
打印的是任务的return返回值
print(res.result()) 拿到的就是异步提交的任务的返回结果
'''
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
pool = ProcessPoolExecutor(5)
pool.submit(task, i).add_done_callback(callback)
6.协程（了解）
进程：资源单位
线程：执行单位
协程：这个概念 完全是程序员自己YY出来的 根本不存在
单线程下 实现并发（欺骗CPU）
我们程序员自己在代码层面上 检测我们所有的IO操作
一旦遇到了IO 我们在代码级别完成切换
这样给CPU的感觉 是你这个程序一直在运行 没有IO
从而提升程序的运行效率
多道技术：
切换 + 保存状态
CPU两种切换：
1.程序遇到IO
2.程序长时间占用
TCP服务端：
accept
recv
代码如何做到：
切换 + 保存状态
切换：
切换不一定是提升效率 也有可能是降低效率
IO切换：提升效率
没有IO切：降低效率
保存状态：
保存上一次我执行的状态 下一次来接着上一次的操作继续往后执行
yield
验证 切换是否就一定提升效率
import time
# 串行执行计算密集型的任务 耗时：1.8605678081512451秒
def func1():
for i in range(10000000):
i + 1
def func2():
for i in range(10000000):
i + 1
start_time = time.time()
func1()
func2()
print(time.time() - start_time)
# 切换 + yield 1.9720776081085205
import time
def func1():
while True:
10000000 + 1
yield
def func2():
g = func1() # 先初始化出 生成器
for i in range(10000000):
i + 1
next(g)
start_time = time.time()
func2()
print(time.time() - start_time)
gevent模块（了解）
安装：在终端输入 pip3 install gevent
from gevent import monkey;monkey.patch_all()
import time
from gevent import spawn
'''
gevent模块 本身无法检测常见的一些IO操作
在使用的时候 需要你额外地导入一句话：
from gevent import monkey
monkey.patch_all()
又由于上面的2句话 在使用gevent模块的时候 肯定是要导入的
所以，还支持简写：
from gevent import monkey;monkey.patch_all()
'''
def heng():
print('哼!')
time.sleep(2)
print('哼！！')
def ha():
print('哈!')
time.sleep(3)
print('哈！！')
def heiheihei():
print('嘿嘿嘿!')
time.sleep(5)
print('嘿嘿嘿！！')
start_time = time.time()
g1 = spawn(heng)
g2 = spawn(ha)
g3 = spawn(heiheihei)
g1.join() # 等待被监测的任务执行完毕 再往后继续执行
g2.join()
g3.join()
# heng()
# ha()
print(time.time() - start_time)
# 哼!
# 哈!
# 嘿嘿嘿!
# 哼！！
# 哈！！
# 嘿嘿嘿！！
# 5.022841453552246
7.协程实现TCP服务端的并发效果（了解）
服务端：
from gevent import monkey;monkey.patch_all()
import socket
from gevent import spawn
def communication(conn):
while True:
try:
data = conn.recv(1024)
if len(data) == 0:
break
conn.send(data.upper())
except ConnectionResetError as e:
print(e)
break
conn.close()
def server(ip, port):
server = socket.socket()
server.bind((ip, port))
server.listen(5)
while True:
conn, addr = server.accept()
spawn(communication, conn)
if __name__ == '__main__':
g1 = spawn(server, '127.0.0.1', 8080)
g1.join()
客户端：
from threading import Thread, current_thread
import socket
def x_client():
client = socket.socket()
client.connect(('127.0.0.1', 8080))
n = 0
while True:
msg = f'{current_thread().name} say hello {n}'
n += 1
client.send(msg.encode('UTF-8'))
data = client.recv(1024)
print(data.decode('UTF-8'))
if __name__ == '__main__':
for i in range(500):
t = Thread(target=x_client)
t.start()
总结：
理想状态：
我们可以通过
多进程下面 开设多线程
多线程下面 开设协程
从而使我们的程序 执行效率提升标签：__,Thread,Python,线程,time,print,import,day41 来源： https://www.cnblogs.com/yding/p/12790442.html