多线程(二)

多线程(二)

1.死锁现象与递归锁

死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程，如下就是死锁

from threading import Thread,Lock
import time

lock_A = Lock()
lock_B = Lock()

class MyThread(Thread):
    def run(self):
        self.f1()
        self.f2()
        
    def f1(self):
        lock_A.acquire()
        print(f'{self.name}拿到了A锁')
        lock_B.acquire()
        print(f'{self.name}拿到了B锁')
        lock_B.release()
        lock_A.release()
        
    def f2(self):
        lock_B.acquire()
        print(f'{self.name}拿到了B锁')
        time.sleep(0.1)
        lock_A.acquire()
        print(f'{self.name}拿到了A锁')
        lock_A.release()
        lock_B.release()
        
if __name__ == '__main__':
    for i in range(3):
        t = MyThread()
        t.start()

#Thread-1拿到了A锁
#Thread-1拿到了B锁
#Thread-1拿到了B锁
#Thread-2拿到了A锁
解决方法，递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。

这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁：
        
        
from threading import Thread,RLock
import time

lock_A = lock_B = RLock()
#递归锁有一个计数的功能,原数字为0,上一次锁,计数+1,释放一次锁,计数-1
#只要递归锁上面的数字不为0,其他线程就不能抢锁

class MyThread(Thread):
    def run(self):
        self.f1()
        self.f2()
        
    def f1(self):
        lock_A.acquire()
        print(f'{self.name}拿到了A锁')
        lock_B.acquire()
        print(f'{self.name}拿到了B锁')
        lock_B.release()
        lock_A.release()
        
    def f2(self):
        lock_B.acquire()
        print(f'{self.name}拿到了B锁')
        time.sleep(0.1)
        lock_A.acquire()
        print(f'{self.name}拿到了A锁')
        lock_A.release()
        lock_B.release()
        
if __name__ == '__main__':
    for i in range(3):
        t = MyThread()
        t.start()        
        
#Thread-1拿到了A锁
#Thread-1拿到了B锁
#Thread-1拿到了B锁
#Thread-1拿到了A锁
#Thread-2拿到了A锁
#Thread-2拿到了B锁
#Thread-2拿到了B锁
#Thread-2拿到了A锁
#Thread-3拿到了A锁
#Thread-3拿到了B锁
#Thread-3拿到了B锁
#Thread-3拿到了A锁

2.信号量

Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。

实例：(同时只有5个线程可以获得semaphore,即可以限制最大连接数为5)：

from threading import Thread, Semaphore, current_thread
import time
import random
sem = Semaphore(5)

def task():
    sem.acquire()

    print(f'{current_thread().name} 厕所ing')
    time.sleep(random.randint(1,3))

    sem.release()


if __name__ == '__main__':
    for i in range(20):
        t = Thread(target=task,)
        t.start()

3.GIL全局解释器锁

理论上:单个进程的多线程可以利用多核,但是Cpython解释器给进入解释器的线程加了锁

加锁的原因:

1.当时处于单核时代,且cpu价格昂贵

2.如果不加全局解释器锁,则程序员需要在源码内部各种主动加锁,解锁,出现死锁现象,则直接在进入解释器时给线程加了锁

优点:保证了Cpython解释器的数据资源的安全

缺点:单个进程的多线程不能利用多核

结论：在Cpython解释器中，同一个进程下开启的多线程，同一时刻只能有一个线程执行，无法利用多核优势

首先需要明确的一点是GIL并不是Python的特性，它是在实现Python解析器(CPython)时所引入的一个概念。就好比C++是一套语言（语法）标准，但是可以用不同的编译器来编译成可执行代码。有名的编译器例如GCC，INTEL C++，Visual C++等。Python也一样，同样一段代码可以通过CPython，PyPy，Psyco等不同的Python执行环境来执行。像其中的JPython,PyPy就没有GIL。然而因为CPython是大部分环境下默认的Python执行环境。所以在很多人的概念里CPython就是Python，也就想当然的把GIL归结为Python语言的缺陷。所以这里要先明确一点：GIL并不是Python的特性，Python完全可以不依赖于GIL


GIL介绍:
    GIL本质就是一把互斥锁，既然是互斥锁，所有互斥锁的本质都一样，都是将并发运行变成串行，以此来控制同一时间内共享数据只能被一个任务所修改，进而保证数据安全。

　　可以肯定的一点是：保护不同的数据的安全，就应该加不同的锁。
	要想了解GIL，首先确定一点：每次执行python程序，都会产生一个独立的进程。例如python test.py，python aaa.py，python bbb.py会产生3个不同的python进程

#验证python test.py只会产生一个进程
#test.py内容
import os,time
print(os.getpid())
time.sleep(1000)

python3 test.py 
#在windows下
tasklist |findstr python
#在linux下
ps aux |grep python

在一个python的进程内，不仅有test.py的主线程或者由该主线程开启的其他线程，还有解释器开启的垃圾回收等解释器级别的线程，总之，所有线程都运行在这一个进程内，毫无疑问

#1 所有数据都是共享的，这其中，代码作为一种数据也是被所有线程共享的（test.py的所有代码以及Cpython解释器的所有代码）
例如：test.py定义一个函数work（代码内容如下图），在进程内所有线程都能访问到work的代码，于是我们可以开启三个线程然后target都指向该代码，能访问到意味着就是可以执行。

#2 所有线程的任务，都需要将任务的代码当做参数传给解释器的代码去执行，即所有的线程要想运行自己的任务，首先需要解决的是能够访问到解释器的代码。

综上：

　　如果多个线程的target=work，那么执行流程是

　　多个线程先访问到解释器的代码，即拿到执行权限，然后将target的代码交给解释器的代码去执行

　　解释器的代码是所有线程共享的，所以垃圾回收线程也可能访问到解释器的代码而去执行，这就导致了一个问题:对于同一个数据100，可能线程1执行x=100的同时，而垃圾回收执行的是回收100的操作，解决这种问题没有什么高明的方法，就是加锁处理，如下图的GIL，保证python解释器同一时间只能执行一个任务的代码
        
IO密集型:适合单个进程的多线程,并发执行
计算密集型:适合多进程的并行

4.GIL锁与lock锁的区别

GIL保护的是解释器级的数据，保护用户自己的数据则需要自己加锁处理

相同点:都是同种锁(互斥锁)
不同点:
    1.GIL全局解释器锁保护解释器内部的资源数据安全
    2.GIL锁的上锁释放无需手动操作
    3.自己代码中定义的互斥锁保护进程中的资源数据的安全
    4.自己定义的互斥锁必须手动上锁和释放

5.验证计算密集型与IO密集型的效率

计算密集型:

from threading import Thread
from multiprocessing import Process
import time,random

def task():
    count = 0
    for i in range(100000000):
        count += 1

#多进程的并发,并行
if __name__ == '__main__':
    start_time = time.time()
    l1 = []
    for i in range(4):
        p = Process(target = task)
        l1.append(p)
        p.start()
    for i in l1:
        p.join()
    print(f"执行效率:{time.time() - start_time}")
    
#执行效率:6.620556354522705
    
#多线程的并发
if __name__ == '__main__':
    start_time = time.time()
    l1 = []
    for i in range(4):
        p = Thread(target = task)
        l1.append(p)
        p.start()
    for i in l1:
        p.join()
    print(f"执行效率:{time.time() - start_time}")
    
#执行效率:17.61694884300232

总结:计算密集型:多进程的并发并行执行效率高
        
        
        
IO密集型:
    
from threading import Thread
from multiprocessing import Process
import time,random
def task():
    count = 0
    time.sleep(random.randint(1,3))
    count += 1

#多进程的并发,并行
if __name__ == '__main__':
    start_time = time.time()
    l1 = []
    for i in range(50):
        p =Process(target = task)
        l1.append(p)
        p.start()
    for i in l1:
        p.join()
    print(f"执行效率:{time.time() - start_time}")
    
#执行效率:3.101606607437134

if __name__ == '__main__':
    start_time = time.time()
    l1 = []
    for i in range(50):
        p =Thread(target = task)
        l1.append(p)
        p.start()
    for i in l1:
        p.join()
    print(f"执行效率:{time.time() - start_time}")
    
#执行效率:2.010244846343994

总结:IO密集型:单个进程的多线程的并发执行效率高
        
        
        

6.多线程实现socket通信

server端:
    
import socket
from threading import Thread

def communicate(conn,addr):
    while 1:
        try:
            from_client_data = conn.recv(1024)
            print(f'来自客户端{addr[1]}的消息: {from_client_data.decode("utf-8")}')
            to_client_data = input('>>>').strip()
            conn.send(to_client_data.encode('utf-8'))
        except Exception:
            break
    conn.close()



def _accept():
    server = socket.socket()

    server.bind(('127.0.0.1', 8848))

    server.listen(5)

    while 1:
        conn, addr = server.accept()
        t = Thread(target=communicate,args=(conn,addr))
        t.start()

if __name__ == '__main__':
    _accept()    

client端:
    
import socket

client = socket.socket()

client.connect(('127.0.0.1',8848))

while 1:
    try:
        to_server_data = input('>>>').strip()
        client.send(to_server_data.encode('utf-8'))

        from_server_data = client.recv(1024)
        print(f'来自服务端的消息: {from_server_data.decode("utf-8")}')

    except Exception:
        break
client.close()

7.进程池和线程池

创建进程池的类：如果指定numprocess为3，则进程池会从无到有创建三个进程，然后自始至终使用这三个进程去执行所有任务（高级一些的进程池可以根据你的并发量，搞成动态增加或减少进程池中的进程数量的操作），不会开启其他进程，提高操作系统效率，减少空间的占用等。

线程池: 一个容器,这个容器限制住你开启线程的数量,比如4个,第一次肯定只能并发的处理4个任务,只要有任务完成,线程马上就会接下一个任务.
#以时间换空间

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import os,time,random

def task():
    print(f"{os.getpid()}工作")
    time.sleep(random.randint(1,3))

#开启进程池(并发或并行)
if __name__ == '__main__':
    p = ProcessPoolExecutor() #默认不写,进程池里面的进程数与本机cpu数量相等
    for i in range(20):
        p.submit(task)
        
        
#开启线程池(并发)
if __name__ == '__main__':
    t = ThreadPoolExecutor() #默认不写,线程池里的线程数为本机cpu个数*5
    for i in range(20):
        t.submit(task)