线程、携程

并发编程

3.进程

16)互斥锁

(1)什么是互斥锁

将并发/并行改为串行(类似单道)，牺牲了效率保障了数据的安全

除了有互斥锁，还有：行锁、表锁、乐观锁、悲观锁..等

(2)互斥锁如何使用

from multiprocessing import Lock  # 导入互斥锁模块

mutex = Lock()  # 产生锁
mutex.acquire()  # 抢锁
# (操作数据代码)
mutex.release()  # 释放锁

"""
注：
  互斥锁之应该放在程序操作的上下,其他位置不要加 会让整个程序的效率特别低
"""

(3)互斥锁解决模拟抢票问题

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(f'{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:
        data['ticket_num'] -= 1
        with open(r'data.json', 'w', encoding='utf8') as f:
            json.dump(data, f)
        print(f'{name}买票成功')
    else:
        print(f'{name}很倒霉 没有抢到票')

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

if __name__ == '__main__':
    mutex = Lock()  # 产生一把锁
    for i in range(3):
        p = Process(target=run, args=(f'用户{i}', mutex))
        p.start()
        """
        执行结果：
          用户0在查票 当前余票为:1
          用户1在查票 当前余票为:1
          用户2在查票 当前余票为:1
          用户0买票成功
          用户1很倒霉 没有抢到票
          用户2很倒霉 没有抢到票
        """

4.线程

1)线程理论

进程：进程是资源单位，表示一块内存空间
线程：线程是执行单位，表示真正的代码指令

进程可以理解为车间 线程可以理解为流水线
一个进程内【至少会含有一个线程】！！！！

🍉1.一个进程内可以开设多个线程

🍉2.同一个进程下的多个线程数据是共享的

🍉3.创建进程的消耗远远大于创建线程
多进程：需要申请内存空间,拷贝全部代码,资源消耗大
多线程：不需要申请内存空间,不需要拷贝全部代码,资源消耗小

2)创建线程的多种方式

创建线程由于不需要拷贝全部代码,所以无论什么系统都不会出现反复操作的情况,也不需要在启动脚本中执行,不过为了兼容和统一,建议加上

由于创建线程需要的资源小,所以以下会发现子线程运行与主线程运行几乎是同时起来的

(1)利用函数方法创建

from threading import Thread
import time

def task(name):
    print(f'{name}子线程正在运行')
    time.sleep(3)
    print(f'{name}子线程运行结束')

if __name__ == '__main__':
    # 创建一个对象t
    t = Thread(target=task, args=('jason',))  # 注意元组
    # 用对象t去创建子线程 去执行task函数
    t.start()
    print('主线程')
    """
    执行结果：
      jason子线程正在运行
      主线程
      jason子线程运行结束
    """

(2)利用面向对象方法创建

from threading import Thread
import time

class MyThread(Thread):
    def run(self):
        print('子线程正在运行')
        time.sleep(3)
        print('子线程运行结束')
        
if __name__ == '__main__':
    obj = MyThread()
    obj.start()
    print('主线程')
    """
    执行结果：
      子线程正在运行
      主线程
      子线程运行结束
    """

如果有参数如何传：

from threading import Thread
import time

class MyThread(Thread):
    def __init__(self,name):
        super().__init__() # 由于重写了__init__方法 但是本身的name默认是有一个子线程名字的,所以需调用super()方法 然后再修改里面的name (如果super()放在下一行那么name就会变成子线程的名字)
# 子线程的名字：MyThread-1
        self.name = name
        
    def run(self):
        print(f'{self.name}子线程正在运行')
        time.sleep(3)
        print(f'{self.name}子线程运行结束')
        
if __name__ == '__main__':
    obj = MyThread('jason')
    obj.start()
    print('主线程')
    """
    执行结果：
      jason子线程正在运行
      主线程
      jason子线程运行结束
    """

3)线程join方法

调用join方法会让主线程变为同步操作,子线程还是异步 (主线程等子线程运行结束后再运行)

同进程,join()的位置不同,执行的结果也不同

from threading import Thread
import time

def task():
    print('子线程正在运行')
    time.sleep(3)
    print('子线程运行结束')

if __name__ == '__main__':
    t = Thread(target=task)
    t.start()
    t.join()  # 主线程等子线程结束后再执行
    print('主线程')
    """
    执行结果：
      子线程正在运行
      子线程运行结束
      主线程
    """

4)同进程内多个线程数据共享

from threading import Thread

money = 1000

def task():
    global money
    money = 666

if __name__ == '__main__':
    t = Thread(target=task)
    t.start()
    t.join()  # 为了确保子线程运行完毕再查找money
    print(money)  # 666

以上得出：同一个进程中,不同线程中数据是共享的 线程修改了进程中的money

5)线程对象的多种方法

(1)查看线程名

print(current_thread().name) 查看自己的线程名

from threading import Thread, current_thread

def task(name):
    print(f'子线程{name}查看了自己的线程名字：', current_thread().name)

# 利用for循环打开2个线程
for i in range(2):
    t = Thread(target=task, args=(i,))
    t.start()
print('主线程查看了自己的线程名字', current_thread().name)
"""
执行结果：
  子线程0查看了自己的线程名字： Thread-1
  子线程1查看了自己的线程名字： Thread-2
  主线程查看了自己的线程名字： MainThread
"""

(2)查看进程号

os.getpid()查看自己的进程号

from threading import Thread, current_thread
import os

def task(name):
    print(f'子线程{name}查看了自己的进程号：', os.getpid())

# 利用for循环打开2个线程
for i in range(2):
    t = Thread(target=task, args=(i,))
    t.start()
print('主线程查看了自己的进程号：', os.getpid())
"""
执行结果：
  子线程0查看了自己的进程号： 6068
  子线程1查看了自己的进程号： 6068
  主线程查看了自己的进程号： 6068
"""

(3)统计进程下的线程数

from threading import Thread, active_count
import time

def task(name):
    print(f'子线程{name}正在运行')
    time.sleep(1)
    print(f'子线程{name}运行结束')

for i in range(2):
    t = Thread(target=task,args=(i,))
    t.start()
    t.join()
print('统计当前进程下存活的线程数：', active_count())
print('主线程')
"""
执行结果：
  子线程0正在运行
  子线程0运行结束
  子线程1正在运行
  子线程1运行结束
  统计当前进程下存活的线程数： 1
  主线程
"""

以上验证了当子线程运行结束后,进程中还是会保留一个线程！

from threading import Thread, active_count
import time

def task(name):
    time.sleep(2)
    print(f'子线程{name}正在运行')
    time.sleep(1)
    print(f'子线程{name}运行结束')

for i in range(2):
    t = Thread(target=task,args=(i,))
    t.start()
print('统计当前进程下存活的线程数：', active_count())
print('主线程')
"""
执行结果：
统计当前进程下存活的线程数： 3
主线程
子线程0正在运行
子线程1正在运行
子线程1运行结束
子线程0运行结束
"""

以上当去掉join，且在子线程前加上睡眠时间即可查看当前线程数

6)GIL全局解释器锁

# 预备知识：
"""
python解释器也是由编程语言写出来的：
  Cpython 是用C写出来的
  Jpython 是用Java写出来的
  Pypython 是用python写出来的
ps:最常用的就是Cpython
"""

# 官方文档对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。
2.GIL本质也是一把互斥锁 让'同一个进程内多个线程无法同时执行'('关键')。
3.GIL的存在主要是因为Cpython解释器中内存管理(垃圾回收机制)不是安全的。

(1)GIL的特点

1.本质也是互斥锁
2.是Cpython解释器的特点 不是python语法的特点
3.GIL的存在让'同一个进程内多个线程无法同时执行',无法发挥多核优势
4.GIL的存在主要是因为Cpython解释器中垃圾回收机制不是安全的
5.GIL两种情况下会释放锁：
    1)线程正常结束
    2)线程进入IO操作

(2)验证GIL的存在

from threading import Thread

num = 10

def task():
    global num
    num -= 1

# 因为下面线程是异步操作,所以用列表存放每一个线程对象
t_list = []
for i in range(10):
    t = Thread(target=task)
    t.start()
    t_list.append(t)
# 循环线程对象列表,同时在最后执行join方法
for t in t_list:
    t.join()
# 等待所有线程运行结束后再查看num是多少
print(num) # 0

以下举例：由于GIL遇到IO操作就会释放锁

from threading import Thread
import time

num = 10

def task():
    global num
    count = num
    time.sleep(0.5)
    num = count - 1

# 因为下面线程是异步操作,所以用列表存放每一个线程对象
t_list = []
for i in range(10):
    t = Thread(target=task)
    t.start()
    t_list.append(t)
# 循环线程对象列表,同时在最后执行join方法
for t in t_list:
    t.join()
# 等待所有线程运行结束后再查看num是多少
print(num)  # 9

(3)验证：既然有GIL是否还要自己加锁

提问：既然Cpython解释器中有GIL的存在,那么以后写代码是不是就不用操作锁了？
回答：错！GIL只能确保同进程内多线程数据不会被垃圾回收机制乱删,并不能确保程序里的数据是否安全。上面第二个例子验证了如果不自己加锁则会让数据错乱,本来应该每一个数值都基于上一个线程减的数再减1.

from threading import Thread, Lock
import time

num = 10

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(10):
    t = Thread(target=task, args=(mutex,))
    t.start()
    t_list.append(t)
# 循环线程对象列表,同时在最后执行join方法
for t in t_list:
    t.join()
# 等待所有线程运行结束后再查看num是多少
print(num)  # 0

(4)python多线程是否有用

需分以下4种情况来判断
"""
单CPU / 多CPU      
IO密集型(有IO操作) / 计算密集型(无IO操作)
"""

1.单CPU 且 IO密集型：
    多进程：申请额外的内存空间,消耗更多资源
    多线程：消耗资源相对较少,利用多道技术实现并发
    # 多线程有优势
2.单CPU 且 计算密集型：
    多进程：申请额外的内存空间,消耗更多资源(总耗时+申请空间+拷贝代码+切换)
    多线程：消耗资源相对较少,利用多道技术实现并发(总耗时+切换)
    # 多线程有优势
3.多CPU 且 IO密集型：
    多进程：总耗时(单个进程的耗时+申请内存空间+拷贝代码)
    多线程：总耗时(单个进程的耗时+IO操作时间)
    # 多线程有优势
4.多CPU 且 计算密集型：
    多进程：总耗时(单个进程的耗时)
    多线程：总耗时(多个进程的综合)
    # 多进程完胜

#计算密集型：多进程 与 多线程

from multiprocessing import Process
from threading import Thread
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()
    # print('总耗时:%s' % (time.time() - start_time))
    '计算密集型-多进程 运行结果:总耗时:6.665070295333862'
    
    t_list = []
    for i in range(12):#一次性创建12个线程
        t = Thread(target=work)
        t.start()
        t_list.append(t)
    for t in t_list:#确保所有的线程全部运行完毕
        t.join()
    print('总耗时:%s' % (time.time() - start_time))
    '计算密集型-多线程 运行结果:总耗时:28.145029306411743'

#IO密集型：多进程 与 多线程

from multiprocessing import Process
from threading import Thread
import os
import time

def work():
    time.sleep(2) #模拟纯IO操作
    
if __name__ == '__main__':
    print(os.cpu_count())  # 12  查看当前计算机CPU个数
    start_time = time.time()
    # p_list = []
    # for i in range(100):  # 一次性创建100个进程
    #     p = Process(target=work)
    #     p.start()
    # for p in p_list:  # 确保所有的进程全部运行完毕
    #     p.join()
    #     p_list.append(p)
    # print('总耗时:%s' % (time.time() - start_time))
    'IO密集型-多进程 运行结果:总耗时:2.979950428009033'
    
    t_list = []
    for i in range(100):  # 一次性创建100个线程
        t = Thread(target=work)
        t.start()
        t_list.append(t)
    for t in t_list:
        t.join()
    print('总耗时:%s' % (time.time() - start_time))
    'IO密集型-多线程 运行结果:总耗时:2.0314550399780273'

结论：

IO密集型：
多进程 :总耗时:2.979950428009033′
多线程 :总耗时:2.0314550399780273′

计算密集型：
​ 多进程 :总耗时:6.665070295333862′
​ 多线程 :总耗时:28.145029306411743′

(5)死锁现象

#死锁就是我想要你手上的锁,你想要我手上的锁,程序卡在这里动不了了

"""
就算掌握了互斥锁的使用
    先抢锁 后释放锁
但是在实际项目尽量少用,一没用好就变成了死锁(项目中领导也不允许使用)
"""
from threading import Thread, Lock
import time
#产生AB两把锁
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):
    t = MyThread()
    t.start()

(6)信号量

#信号量本质也是互斥锁 只不过它是一次性产生多把锁

"""
强调:
    信号量在不同的知识体系中意思不同：
        1.在并发编程中 信号量就是多把互斥锁
        2.在django中 信号量指的是达到某个条件自动触发(中间件)

之前使用Lock产生的是单把锁
	类似于单间厕所,一个人走了另一个才能进
信号量是一次性创建多把锁
	类似于公共厕所,其中一个人走了另一个人才能进
"""

from threading import Thread, 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))  # 随机睡眠1~3秒
        sp.release()  # 释放锁

# 开10个线程
for i in range(10):
    t = MyThread()
    t.start()

(7)event事件

子进程\子线程之间可以彼此等待彼此
	eg:
      子A运行到某一个代码位置后发信号告诉子B开始运行
    
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(5):
    t = Thread(target=car, args=('赛车%s' % i,))
    t.start()
    """
    执行结果：
      红灯 所有人都不能动
      赛车0正在等红灯
      赛车1正在等红灯
      赛车2正在等红灯
      赛车3正在等红灯
      赛车4正在等红灯
      绿灯 踩油门冲啊!!!
      赛车0加油门 飙车了
      赛车4加油门 飙车了
      赛车1加油门 飙车了
      赛车2加油门 飙车了
      赛车3加油门 飙车了
    """

5.进程池与线程池

# 问：在实际应用中是否可以无限制的开进程和线程

 答：不可以! 因为硬件的发展赶不上软件，有'物理极限' 如果在编写代码的过程中无限制的创建进程或者线程可能会导致计算机崩溃

池：降低程序的执行效率 保证计算机硬件的安全
进程池：提前创建好固定个数的进程供程序使用 后续不会再创建
线程池：提前创建好固定个数的线程供程序使用 后续不会再创建

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from threading import current_thread
import os
import time

# pool = ThreadPoolExecutor(5)  # 固定产生五个线程池
pool = ProcessPoolExecutor(5)  # 固定产生五个进程池
def task(n):
    # print(current_thread().name)
    print(os.getpid())#获取当前进程号
    # print(n)
    time.sleep(1)
    return '返回的结果'
def func(*args, **kwargs):
    print('func', args, kwargs)
    print(args[0].result())
if __name__ == '__main__':
    for i in range(20):
        # res = pool.submit(task,123)  # 朝池子中提交任务(异步)
        # print(res.result())  # 同步
        # pool.submit(task, 123).add_done_callback(func)
        """
        add_done_callback
        异步回调:异步任务执行完成后有结果就会自动触发该机制
        """
        pool.submit(task, 123).add_done_callback(func)

6.携程

进程:资源单位
线程:执行单位
并发:多个进程看上去像同时执行的就是并发
协程:单线程下实现并发(效率极高)
在代码层面欺骗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

1)协程实现TCP服务端并发

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

原文地址：http://www.cnblogs.com/lvqingmei/p/16913607.html