1.多进程实现TCP服务端并发
1.之前我们学习了一个服务端对应一个客户端的操作,但是我们无法实现一个服务端对应多个客户端的操作。因此我们需要在pycharm的右上角点击倒三角,选择Edit Configurations,我们便可进行多客户端操作。
客户端代码:
import socket
client = socket.socket()
client.connect(('127.0.0.1', 8080))
while True:
client.send(b'hello world')
data = client.recv(1024)
print(data)
服务端:
import socket
sever = socket.socket()
sever.bind(('127.0.0.1', 8080))
sever.listen(5)
while True:
sock, addr = sever.accept()
while True:
data = sock.recv(1024)
print(data.decode('utf8'))
sock.send(data.upper())
"""
但是执行此操作就会发现第一个客户端有内容,后面的客户端都没有内容。因此单进程无法实现上述效果,我们必须使用多进程来实现上述并发的效果
"""
服务端:
import socket
from multiprocessing import Process
def get_sever():
server = socket.socket()
server.bind(('127.0.0.1', 8080))
server.listen(5)
return server
def get_talk(sock):
while True:
data = sock.recv(1024)
print(data.decode('utf8'))
sock.send(data.upper())
if __name__ == '__main__':
server = get_sever()
while True:
sock, addr = server.accept()
p = Process(target=get_talk, args=(sock,))
p.start()
'''每新增一个客户端就会新增一个人进程,相当于雇了服务员,每新增一个客户端就会新雇一名服务员'''
客户端:
import socket
client = socket.socket()
client.connect(('127.0.0.1', 8080))
while True:
client.send(b'hello world')
data = client.recv(1024)
print(data)
2.互斥锁代码实操
1.互斥锁主要用于抢票等操作,高并发的情况下多个用户同事抢一张票,而最终只有一个用户抢票成功,每一个用户相当于新开一个进程。按照以往思维我们只要有一个用户抢票成功,那么文件余票数量就会被修改,现在执行上述操作:
代码:
import random
from multiprocessing import Process
import json
import time
def search():
with open(r'data.json', 'r', encoding='utf8') as f:
data = json.load(f)
print(f'目前余票{data.get("ticket_num")}')
def buys(name):
# 查票
with open(r'data.json', 'r', encoding='utf8') as f:
data = json.load(f)
# 模拟网络延迟
time.sleep(random.randint(1,3))
if data.get('ticket_num') > 0:
with open(r'data.json', 'w', encoding='utf8') as f:
data['ticket_num'] -= 1
json.dump(data, f)
print(f'{name}购票成功')
else:
print('余票不足,无法购买')
def run(name):
search()
buys(name)
'''接下来我们模拟多个消费者同时抢票,模拟多进程'''
if __name__ == '__main__':
for i in range(10):
p = Process(target=run, args=('用户%s号' % i,))
p.start()
文件以json格式来保存:
{"ticket_num": 0}
```python
"""
产生的效果表明,所有的用户都可以抢票成功,这是因为在多个进程执行的同一刻,时间足够短所以多个进程可以同时改文件内的数据。for循环可以看做同一时间执行,图中执行效果并不是按照顺序来执行。要想避免此操作我们需要使用互斥锁。互斥锁建议加在数据处理的部分,因为互斥锁会拖慢代码执行的效率,只加在购票过程中,如果查票也要加锁name查票也需要排队,不符合逻辑。
"""
import random
from multiprocessing import Process,Lock
import json
import time
def search(name):
with open(r'data.json', 'r', encoding='utf8') as f:
data = json.load(f)
print(f'亲爱的{name}目前余票{data.get("ticket_num")}')
def buys(name):
# 查票
with open(r'data.json', 'r', encoding='utf8') as f:
data = json.load(f)
# 模拟网络延迟
time.sleep(random.randint(1,3))
if data.get('ticket_num') > 0:
with open(r'data.json', 'w', encoding='utf8') as f:
data['ticket_num'] -= 1
json.dump(data, f)
print(f'{name}购票成功')
else:
print('余票不足,无法购买')
def run(name,mutex):
search(name) # 看票大家都可以看
mutex.acquire() # 抢锁,抢到之后其他操作无法执行,CPU被下面的代码占用
buys(name)
mutex.release() # 释放锁
'''接下来我们模拟多个消费者同时抢票,模拟多进程'''
if name == 'main':
mutex = Lock()
for i in range(10):
p = Process(target=run, args=('用户%s号' % i, mutex))
p.start()
<img src="/i/l/?n=22&i=blog/2987409/202211/2987409-20221121230239181-40881578.png" height=450 width=800 />
## 3.线程理论
```python
1.进程:是资源单位,表示一块内存空间。
2.线程:是执行单位,表示真正的代码指令。
"""
每个进程内都至少有一个线程,线程才是真正跟CPU打交道的角色。如果把进程比作一个车间,那么车间内的生产线就是线程。
有了进城为什么还要有线程呢?
1.进程只能在一个时间干一件事，如果想同时干两件事或多件事，进程就无能为力了。
2.进程在执行的过程中如果阻塞，例如等待输入，整个进程就会挂起，即使进程中有些工作不依赖于输入的数据，也将无法执行。
"""
3.线程特性:
3.1一个进程号可以设置多个线程
3.2同一个进程下的多个线程是数据共享的
3.3创建进程和线程的区别:创建进程的资源远大于线程
4.创建线程的两种方式
方法1:
from threading import Thread
import time
def task(name):
print(f'{name} is running')
time.sleep(3)
print(f'{name} is over')
t = Thread(target=task, args=('max',))
t.start() # 创建一个线程
print('主线程')
'''
max is running主线程
max is over
'''
"""
创建进程时由于子进程加载需要时间,所以会先执行子进程下面的操作。而创建线程消耗资源极小,所以会先打印max is running
"""
方法2:s
class MyThread(Thread):
def run(self):
print('run is running')
time.sleep(1)
print('run is over')
obj = MyThread()
obj.start()
print('主线程')
创建进程CPU资源远大于线程
# 7 8
5.线程的诸多特性
1.join方法:
"""
正常情况下我们建立线程,会先执行子线程代码(IO状态除外),再执行主线程代码
"""
from threading import Thread
import time
def task(name):
print(f'{name} is running')
time.sleep(1)
print(f'{name} is over')
t = Thread(target=task, args=('max',))
t.start()
print('主线程')
'''
max is running主线程
max is over
'''
"""
加上join之后会先执行子线程代码,再执行主线程代码(和进程一样)
"""
from threading import Thread
import time
def task(name):
print(f'{name} is running')
time.sleep(1)
print(f'{name} is over')
t = Thread(target=task, args=('max',))
t.start()
t.join()
print('主线程')
'''
max is running
max is over
主线程
'''
2.同一个进程下的多个线程数据共享
from threading import Thread
money = 1000
def task():
global money
money = 666
t = Thread(target=task)
t.start()
t.join()
print(money)
'''
666 改变的是该线程名称空间内的money,子线程和主线程数据共享
'''
3.current_thread():查看是主线程还是子线程,主线程是MainThread,子线程是Thread-数字,数字从1开始排
from threading import Thread, current_thread
money = 1000
def task():
global money
money = 666
print(current_thread().name)
t = Thread(target=task)
t.start()
t.join()
print(money)
print(current_thread().name)
'''
Thread-1
666
MainThread
'''
3.线程也可以用os.getpid()查看进程号,它们都同属一个进程,所以进程号一致
from threading import Thread
import os
money = 1000
def task():
global money
money = 666
print('子线程进程号>>>:',os.getpid())
t = Thread(target=task)
t.start()
t.join()
print('主线程进程号>>>:',os.getpid())
'''
子线程进程号>>>: 34388
主线程进程号>>>: 34388
'''
4.active_count:用来查看存活的线程数量
"""
由于每次创建线程保留速度很短,线程创建之后就会消失,可以采用sleep的方法查看效果
"""
from threading import Thread, active_count
import os
import time
money = 1000
def task():
print('子线程进程号>>>:',os.getpid())
time.sleep(3)
for i in range(10):
t = Thread(target=task)
t.start()
print('存活的线程数:', active_count())
'''
11
'''
5.守护线程:守护线程和守护进程一样,会等待主进程(或线程)运行完毕后被销毁
from threading import Thread
import time
def task():
print('子线程运行task开始')
time.sleep(3)
print('子线程运行task结束')
t = Thread(target=task)
t.daemon = True
t.start()
print('主线程')
'''子线程运行task开始主线程''' # '子线程运行task结束'由于主线程运行结束没有执行
6.GIL全局解释器锁
"""
GIL是CPython解释器的特性,和Python语言无关系
"""
In CPython, the global interpreter lock, or GIL, is a mutex that prevents multiple native threads from executing Python bytecodes at once. This lock is necessary mainly because CPython’s memory management is not thread-safe. (However, since the GIL exists, other features have grown to depend on the guarantees that it enforces.
翻译:
1.在CPython解释器中存在全局解释器锁简称GIL
python解释器有很多类型
CPython JPython PyPython (常用的是CPython解释器)
2.GIL本质也是一把互斥锁,用来阻止同一个进程内多个线程同时执行(重要)
'''利用多道技术,同一时间同一个进程内,只能利用一个CPU'''
3.GIL的存在是因为CPython解释器中内存管理不是线程安全的(垃圾回收机制)
垃圾回收机制:引用计数、标记清除、分代回收
'''每一个进程都有一个垃圾回收机制的线程。为了防止每个数据值建立时就被垃圾回收机制回收,CPython中规定多个线程不能同时执行'''
4.所有解释型语言都逃不开上述的问题。
7.验证GIL的存在
from threading import Thread
num = 100
def task():
global num
num -= 1
t_list = []
for i in range(100):
t = Thread(target=task)
t.start()
t_list.append(t)
for t in t_list:
t.join()
print(num)
8.GIL与普通互斥锁
既然CPython解释器中有GIL 那么我们以后写代码是不是就不需要操作锁了!!!
"""
GIL只能够确保同进程内多线程数据不会被垃圾回收机制弄乱
并不能确保程序里面的数据是否安全
"""
import time
from threading import Thread,Lock
num = 100
def task(mutex):
global num
mutex.acquire()
count = num
time.sleep(0.1)
num = count - 1
mutex.release()
mutex = Lock()
t_list = []
for i in range(100):
t = Thread(target=task,args=(mutex,))
t.start()
t_list.append(t)
for t in t_list:
t.join()
print(num)
9.python多线程是否有用
需要分情况
情况1
单个CPU
多个CPU
情况2
IO密集型(代码有IO操作)
计算密集型(代码没有IO)
1.单个CPU
IO密集型
多进程
申请额外的空间 消耗更多的资源
多线程
消耗资源相对较少 通过多道技术
ps:多线程有优势!!!
计算密集型
多进程
申请额外的空间 消耗更多的资源(总耗时+申请空间+拷贝代码+切换)
多线程
消耗资源相对较少 通过多道技术(总耗时+切换)
ps:多线程有优势!!!
2.多个CPU
IO密集型
多进程
总耗时(单个进程的耗时+IO+申请空间+拷贝代码)
多线程
总耗时(单个进程的耗时+IO)
ps:多线程有优势!!!
计算密集型
多进程
总耗时(单个进程的耗时)
多线程
总耗时(多个进程的综合)
ps:多进程完胜!!!
from threading import Thread
from multiprocessing import Process
import os
import time
def work():
# 计算密集型
res = 1
for i in range(1, 100000):
res *= i
if __name__ == '__main__':
# print(os.cpu_count()) # 12 查看当前计算机CPU个数
start_time = time.time()
# p_list = []
# for i in range(12): # 一次性创建12个进程
# p = Process(target=work)
# p.start()
# p_list.append(p)
# for p in p_list: # 确保所有的进程全部运行完毕
# p.join()
t_list = []
for i in range(12):
t = Thread(target=work)
t.start()
t_list.append(t)
for t in t_list:
t.join()
print('总耗时:%s' % (time.time() - start_time)) # 获取总的耗时
"""
计算密集型
多进程:5.665567398071289
多线程:30.233906745910645
"""
def work():
time.sleep(2) # 模拟纯IO操作
if __name__ == '__main__':
start_time = time.time()
# t_list = []
# for i in range(100):
# t = Thread(target=work)
# t.start()
# for t in t_list:
# t.join()
p_list = []
for i in range(100):
p = Process(target=work)
p.start()
for p in p_list:
p.join()
print('总耗时:%s' % (time.time() - start_time))
"""
IO密集型
多线程:0.0149583816528320
多进程:0.6402878761291504
"""
10.死锁现象
acquire()
release()
from threading import Thread,Lock
import time
mutexA = Lock() # 产生一把锁
mutexB = Lock() # 产生一把锁
class MyThread(Thread):
def run(self):
self.func1()
self.func2()
def func1(self):
mutexA.acquire()
print(f'{self.name}抢到了A锁')
mutexB.acquire()
print(f'{self.name}抢到了B锁')
mutexB.release()
print(f'{self.name}释放了B锁')
mutexA.release()
print(f'{self.name}释放了A锁')
def func2(self):
mutexB.acquire()
print(f'{self.name}抢到了B锁')
time.sleep(1)
mutexA.acquire()
print(f'{self.name}抢到了A锁')
mutexA.release()
print(f'{self.name}释放了A锁')
mutexB.release()
print(f'{self.name}释放了B锁')
for i in range(10):
obj = MyThread()
obj.start()
11.信号量
在python并发编程中信号量相当于多把互斥锁(公共厕所)
from threading import Thread, Lock, Semaphore
import time
import random
sp = Semaphore(5) # 一次性产生五把锁
class MyThread(Thread):
def run(self):
sp.acquire()
print(self.name)
time.sleep(random.randint(1, 3))
sp.release()
for i in range(20):
t = MyThread()
t.start()
12.event事件
from threading import Thread, Event
import time
event = Event() # 类似于造了一个红绿灯
def light():
print('红灯亮着的 所有人都不能动')
time.sleep(3)
print('绿灯亮了 油门踩到底 给我冲!!!')
event.set()
def car(name):
print('%s正在等红灯' % name)
event.wait()
print('%s加油门 飙车了' % name)
t = Thread(target=light)
t.start()
for i in range(20):
t = Thread(target=car, args=('熊猫PRO%s' % i,))
t.start()
13.进程池与线程池
进程和线程能否无限制的创建 不可以
因为硬件的发展赶不上软件 有物理极限 如果我们在编写代码的过程中无限制的创建进程或者线程可能会导致计算机奔溃
池
降低程序的执行效率 但是保证了计算机硬件的安全
进程池
提前创建好固定数量的进程供后续程序的调用 超出则等待
线程池
提前创建好固定数量的线程供后续程序的调用 超出则等待
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import os
import time
import random
from threading import current_thread
# 1.产生含有固定数量线程的线程池
# pool = ThreadPoolExecutor(10)
pool = ProcessPoolExecutor(5)
def task(n):
print('task is running')
# time.sleep(random.randint(1, 3))
# print('task is over', n, current_thread().name)
# print('task is over', os.getpid())
return '我是task函数的返回值'
def func(*args, **kwargs):
print('from func')
if __name__ == '__main__':
# 2.将任务提交给线程池即可
for i in range(20):
# res = pool.submit(task, 123) # 朝线程池提交任务
# print(res.result()) # 不能直接获取
# pool.submit(task, 123).add_done_callback(func)
14.协程
"""
进程:资源单位
线程:执行单位
协程:单线程下实现并发(效率极高)
在代码层面欺骗CPU 让CPU觉得我们的代码里面没有IO操作
实际上IO操作被我们自己写的代码检测 一旦有 立刻让代码执行别的
(该技术完全是程序员自己弄出来的 名字也是程序员自己起的)
核心:自己写代码完成切换+保存状态
"""
import time
from gevent import monkey;
monkey.patch_all() # 固定编写 用于检测所有的IO操作(猴子补丁)
from gevent import spawn
def func1():
print('func1 running')
time.sleep(3)
print('func1 over')
def func2():
print('func2 running')
time.sleep(5)
print('func2 over')
if __name__ == '__main__':
start_time = time.time()
# func1()
# func2()
s1 = spawn(func1) # 检测代码 一旦有IO自动切换(执行没有io的操作 变向的等待io结束)
s2 = spawn(func2)
s1.join()
s2.join()
print(time.time() - start_time) # 8.01237154006958 协程 5.015487432479858
15.协程实现并发
import socket
from gevent import monkey;monkey.patch_all() # 固定编写 用于检测所有的IO操作(猴子补丁)
from gevent import spawn
def communication(sock):
while True:
data = sock.recv(1024)
print(data.decode('utf8'))
sock.send(data.upper())
def get_server():
server = socket.socket()
server.bind(('127.0.0.1', 8080))
server.listen(5)
while True:
sock, addr = server.accept() # IO操作
spawn(communication, sock)
s1 = spawn(get_server)
s1.join()
如何不断的提升程序的运行效率
多进程下开多线程 多线程下开协程