内容详细

• 多进程实现TCP服务端并发

• 互斥锁代码实操

• 线程理论

• 创建线程的多种方式

• 线程诸多特性

• GIL全局解释器锁

• 验证GIL存在

• GIL与普通互斥锁的区别

• 验证多线程作用

• 死锁现象

• 信号量

• event事件

• 进程池与线程池(重要)

• 协程

• 协程实现并发

多进程实现TCP服务端并发

import socket
from multiprocessing import Process


def get_server():
    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_server()
    while True:
        sock, addr = server.accept()
        # 开设多进程去聊天
        p = Process(target=get_talk, args=(sock,))
        p.start()

互斥锁代码实操

锁:建议只加载操作数据的部分 否则整个程序的效率会极低
from multiprocessing import Process, Lock
import time
import json
import random


def search(name):
    with open(r'data.json', 'r', encoding='utf8') as f:
        data = json.load(f)
    print('%s查看票 目前剩余:%s' % (name, data.get('ticket_num')))


def buy(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('%s 买票成功' % name)
    else:
        print('%s 买票失败 非常可怜 没车回去了!!!' % name)


def run(name, mutex):
    search(name)
    mutex.acquire()  # 抢锁
    buy(name)
    mutex.release()  # 释放锁


if __name__ == '__main__':
    mutex = Lock()  # 产生一把锁
    for i in range(10):
        p = Process(target=run, args=('用户%s号' % i, mutex))
        p.start()
"""
锁有很多种 但是作用都一样
	行锁 表锁 ...
"""

线程理论

进程
	进程其实是资源单位 表示一块内存空间
线程
	线程才是执行单位 表示真正的代码指令
    
我们可以将进程比喻成车间 线程是车间里面的流水线
一个进程内部至少含有一个线程

# 为啥要有线程呢
  开设线程的消耗远远小于进程
  开进程的步骤:
    1.申请内存空间
    2.拷贝代码
  开线程的步骤:
    一个进程内可以有多个线程 无需申请内存空间、拷贝代码
    一个进程内的多个线程是共享的
    
# 开发一个文本编辑器
 1、获取用户输入并实时展示到屏幕上
 2、并实时保存到硬盘里面
多种功能应该开设多线程，而不是多进程

创建线程的两种方式

1.函数创建线程

# from threading import Thread 创建线程模块
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=('owen', ))
t.start()
print('主线程')

"""
创建线程无需考虑反复执行的问题
"""

2.用对象创建线程

from threading import Thread
import time
class MyThread(Thread):
    def __init__(self, username):
        super().__init__()
        self.username = username
    def run(self):
        print(f'{self.username}  is running')
        time.sleep(3)
        print(f'{self.username} is over')

t = MyThread('owen')
t.start()
print('主线程')

线程的诸多特性

1.join方法
2.同进程内多个线程数据共享
3.current_thread()
4.active_count()

GIL全局解释器锁

# 官方文档
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本质也是一把互斥锁 用来阻止同一个进程内多个线程同时执行(重要)

3.GIL的存在是因为CPython解释器中内存管理不是线程安全的(垃圾回收机制)
	垃圾回收机制
		引用计数、标记清除、分代回收
    
GIL只存在于CPython解释器中，不是python的特征
GIL是一把互斥锁用于阻止同一个进程下的多个线程同时执行
原因是因为CPython解释器中的垃圾回收机制不是线程安全的

反向验证GIL的存在 如果不存在会产生垃圾回收机制与正常线程之间数据错乱
GIL是加在CPython解释器上面的互斥锁
同一个进程下的多个线程要想执行必须先抢GIL锁 所以同一个进程下多个线程肯定不能同时运行 即无法利用多核优势

强调:同一个进程下的多个线程不能同时执行即不能利用多核优势
 python是垃圾 速度太慢 有多核都不能用
反怼:虽然用一个进程下的多个线程不能利用多核优势 但是还可以开设多进程!!!

再次强调:python的多线程就是垃圾!!!
反怼:要结合实际情况 
如果多个任务都是IO密集型的 那么多线程更有优势(消耗的资源更少)
		多道技术:切换+保存状态
如果多个任务都是计算密集型 那么多线程确实没有优势 但是可以用多进程
	CPU越多越好
以后用python就可以多进程下面开设多线程从而达到效率最大化
"""
1.所有的解释型语言都无法做到同一个进程下多个线程利用多核优势
2.GIL在实际编程中其实不用考虑

验证GIL的存在

from threading import Thread

num = 100 

def task():
    global num
    num -= 1
    
t_list = []
for i in rang(100):
    t = Thread(target=task)
    t.start()
    t_list.append(t)
for t in t_list:
    t.join()
print(num)

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) # 等待线程运行完毕再打印

"""
  GIL是一个纯理论知识， 在实际工作中不同考虑它的存在，
  GIL作用面很窄， 仅限于解释器级别
"""

验证多线程作用

需要分情况
	情况1
    	单个CPU
    	多个CPU
	情况2
    	IO密集型(代码有IO操作)
       计算密集型(代码没有IO)
 
# 1.单个CPU
	IO密集型
    	多进程
        申请额外的空间 消耗更多的资源
		多线程
        消耗资源相对较少 通过多道技术
      ps:多线程有优势!!!
        
 	计算密集型
    	多进程
        申请额外的空间 消耗更多的资源(总耗时+申请空间+拷贝代码+切换)
 		多线程
        消耗资源相对较少 通过多道技术(总耗时+切换)
      ps:多线程有优势!!!
        
# 2.多个CPU
	IO密集型
   		多进程
         总耗时(单个进程的耗时+IO+申请空间+拷贝代码)
       多线程
    	  总耗时(单个进程的耗时+IO)
       ps:多线程有优势!!!
        
	计算密集型
    	多进程
       	  总耗时(单个进程的耗时)
    	多线程
          总耗时(多个进程的综合)
       ps:多进程完胜!!!

计算密集型

 # 多进程   
from multiprocessing import Process
import time

def work():
    res = 1
    for i in range(1, 10000):
        res *= i

if __name__ == '__main__':
    # print(os.cpu_count())  # 8 可以获取计算机的CPU个数
    start_time = time.time()
    p_list = []
    for i in range(8):
        p = Process(target=work)
        p.start()
        p_list.append(p)
    for p in p_list:
        p.join()
    print('总耗时:%s' % (time.time() - start_time))
    总耗时:0.19758033752441406
        
# 多线程       
from threading import Thread
from multiprocessing import Process
import os
import time

def work():
    res = 1
    for i in range(1, 10000):
        res *= i

if __name__ == '__main__':
    # print(os.cpu_count())  # 8 可以获取计算机的CPU个数
    start_time = time.time()
    p_list = []
    for i in range(8):
        p = Process(target=work)
        p.start()
        p_list.append(p)
    for p in p_list:
        p.join()
    t_list = []
    for i in range(8):
        t = Thread(target=work)
        t.start()
        t_list.append(t)
    for t in t_list:
        t.join()
    print('总耗时:%s' % (time.time() - start_time))  
    总耗时:0.322171688079834
 """
   计算密集型
     多进程
       0.19758033752441406
     多线程
        0.322171688079834
   所以多进程好
 """

IO密集型

# 多线程
from threading import Thread
import time
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()
    print('总耗时:%s' % (time.time() - start_time))
 # 总耗时:0.011942148208618164

# 多进程
def work():
    time.sleep(2)

if __name__ == '__main__':
    start_time = time.time()
    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))
    # 总耗时:0.5508136749267578
"""
 IO密集型
   多线程
     0.011942148208618164
   多进程
     0.5508136749267578
  结论:
    多线程更好
"""

死锁现象

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()
    # 出现死锁现象

"""
锁不能轻易使用并且我们也不会自己去处理锁，因为都是拿别人封装的工具
"""

信号量

# 信号量在不同的体系里面， 展现出来的功能是不一样
在并发编程中信号量意思就是多把互斥锁
在django框架中信号量意思就是达到某个条件自动触发特定的功能
"""
 将自定义互斥锁比喻成豪华个人卫生间
 信号量就相当于公共厕所，里面有多个位子
"""
from threading import Thread, Semaphore
import time
import random

sp = Semaphore(5)  # 自定义五个带门锁的坑位的厕所

def task(name):
    sp.acquire()  # 抢锁
    print('%s正在蹲坑' % name)
    time.sleep(random.randint(1, 5))
    sp.release()  # 放锁

for i in range(20):
    t = Thread(target=task, args=('小赤佬%s号'% i, ))
    t.start()

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()
 # 结果显示，前面先在等红灯， 然后时间一到就开始跑了

进程池与线程池(重要)

"""
  服务端必备的三要素
      1、24小时不间断提供服务
      2、固定的IP和port
      3、支持高并发
"""
TCP服务端实现并发
  多进程: 来一个人客户端开一个进程
  多线程: 来一个客户端开一个线程
    
进程和线程能否无限制的创建 不可以
因为硬件的发展赶不上软件 有物理极限 如果我们在编写代码的过程中无限制的创建进程或者线程可能会导致计算机奔溃
    
计算机硬件是有物理极限的，我们不可能无限制的创建进程和线程
解决方法:
   """ 池：
       降低程序的执行效率 但是保证了计算机硬件的安全
    进程池：
        提前创建好固定数量的进程供后续程序的调用 超出则等待
    线程池:
        提前创建好固定数量的线程供后续程序的调用 超出则等待
        
 当任务超出了池子里面的最大进程或线程数， 就原地等待
 进程池和线程池其实降低了程序的运行效率， 但是它保证了硬件的安全
    """
# 代码
# 线程池
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time
from threading import current_thread

pool = ThreadPoolExecutor(5)  # 线程池线程数默认是CPU的五倍，也可以自定义

def task(n):
    time.sleep(3)
    print(n)
    print(current_thread().name)  # 线程最大数为4
    

def func(*args, **kwargs):
    print(args, kwargs)


'''让异步提交自动提醒>>>:异步回调机制'''
if __name__ = '__main__':
    for i in range(10):
        pool.submit(task, i).add_done_callback(func)
"""
 add_done_callback当任务有结果的时候， 自动调用括号内的函数处理
"""

# 进程池
pool = ProcessPoolExecutor(5)  # 进程池进程数默认是CPU个数 也可以自定义
'''代码执行之后就会立刻创建五个等待工作的进程'''
pool.submit(task, i).add_done_callback(func)

协程

"""
进程:资源单位
线程:执行单位
协程:单线程下实现并发(效率极高)
	在代码层面欺骗CPU 让CPU觉得我们的代码里面没有IO操作
	实际上IO操作被我们自己写的代码检测 一旦有 立即让代码执行别的
	从而最大幅度的占用CPU
	(该技术)
"""
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()
    # print(time.time() - start_time) 
    # 8.01237154006958
    s1 = spawn(func1)  # 检测代码 一旦有IO自动切换(执行没有io的操作 变向的等待io结束)
    s2 = spawn(func2)
    s1.join()
    s2.join()
    print(time.time() - start_time)  
    # 协程 5.015487432479858

协程实现并发

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

如何不断的提升程序的运行效率
	多进程下开多线程 多线程下开协程