并发编程之多线程

Posted 2020-12-01 zrxu

线程
一、什么是线程?
在传统的操作系统当中，每一个进程有一个地址空间，默认就有一个控制线程
进程只是用来把资源集中到一起(进程只是一个资源单位,或者说是资源集合),而线程才是cpu上的执行单位
多线程:是指在一个进程当中，开启多个线程，多个线程共享该进程的地址空间。

二、线程与进程的区别
1、同一个进程内的多个线程共享该内存的地址资源
2、创建线程的开销远小于创建进程的开销

开启线程的两种方式
第一种方式

from threading import Thread
import time


class MyTread(Thread, object):

    def __init__(self, name):
        super(MyTread, self).__init__()
        self.name = name


    def run(self):
        print("%s is running" % self.name)
        time.sleep(1)
        print("%s is end" % self.name)

if __name__ == ‘__main__‘:
    t = MyTread("线程1")
    t.start()

    print("主线程")
‘‘‘
打印结果:
线程1 is running主线程

线程1 is end
‘‘‘

第二种方式

from threading import Thread
import time

def task(name):
    print("%s is running" % name)
    time.sleep(1)
    print("%s is end" % name)

if __name__ == ‘__main__‘:
    t = Thread(target=task, args=("线程1",))
    t.start()

    print("主线程")
‘‘‘
打印结果:
线程1 is running主线程

线程1 is end
‘‘‘

三、多线程与多进程的区别
多线程的打印结果与多进程的打印结果相比,多线程先打印的是"线程1 is running",而多进程先打印的是"主",可以看出,开启一个线程所需要的时间是小于多进程的

在来看一开pid

from threading import Thread
import time,os

def task(name):
    print("%s is running, pid:%s" % (name, os.getpid()))
    # time.sleep(1)
    # print("%s is end, pid:%s" % (name, os.getpid()))

if __name__ == ‘__main__‘:
    t = Thread(target=task, args=("线程1",))
    t1 = Thread(target=task, args=("线程2",))
    t.start()
    t1.start()

    print("主线程 pid:%s" % os.getpid())

‘‘‘
打印结果:
线程1 is running, pid:2157
线程2 is running, pid:2157
主线程 pid:2157
‘‘‘

可以看出其pid是一样的,一个进程下的多线程的pid与主线程是一样的

一个进程下的多线程共享该进程的资源

from threading import Thread
import os

def work():
    global n
    n=0

if __name__ == ‘__main__‘:
    n=100
    t=Thread(target=work)
    t.start()
    t.join()
    print(‘主‘,n)

‘‘‘
打印结果:
主 0
‘‘‘

四、threading的其他属性和方法
Thread对象的方法
isAlive(): 返回线程是否活动的。
getName(): 返回线程名。
setName(): 设置线程名

threading模块提供的一些方法：
threading.currentThread(): 返回当前的线程变量。
threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

from threading import Thread, current_thread
import threading
import time

def task():
    time.sleep(1)
    print("task内打印的子线程名:%s"% current_thread().getName())

if __name__ == ‘__main__‘:
    t = Thread(target=task)
    t.setName("线程1")
    t.start()

    print("t线程是否存活:%s" % t.is_alive())
    print("正在运行的线程的list:%s" % threading.enumerate())
    print("线程存活个数: %s" % threading.active_count())
    print("子线程名:%s" % t.getName())
    print("主线程名:%s" % current_thread().getName())

‘‘‘
打印结果:
t线程是否存活:True
正在运行的线程的list:[<_MainThread(MainThread, started 4642616768)>, <Thread(线程1, started 123145372258304)>]
线程存活个数: 2
子线程名:线程1
主线程名:MainThread
task内打印的子线程名:线程1
‘‘‘

join方法:等待子线程结束后才运行后面的程序

from threading import Thread, current_thread
import time

def task():
    print("子线程开始")
    time.sleep(1)
    print("task内打印的子线程名:%s"% current_thread().getName())

if __name__ == ‘__main__‘:
    t = Thread(target=task)
    t.setName("线程1")
    t.start()
    t.join()
    print("主线程")

‘‘‘
打印结果:
子线程开始
task内打印的子线程名:线程1
主线程
‘‘‘

五、守护线程
守护线程和守护进程类似:守护线程(进程)会等待主线程(进程)运行完毕后被销毁
ps:运行完毕与中止运行不同

1、主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，才会结束。
2、主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。

from threading import Thread, current_thread
import time

def task():
    print("子线程开始")
    time.sleep(1)
    print("task内打印的子线程名:%s"% current_thread().getName())

if __name__ == ‘__main__‘:
    t = Thread(target=task)
    t.setName("线程1")
    t.setDaemon(True)
    t.start()
    time.sleep(0.5)
    print("主线程")
‘‘‘
打印结果:
子线程开始
主线程
‘‘‘

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

在Cpython解释器中,同一个进程下开启的多线程,同一时刻只能有一个线程执行,无法利用多核优势
GIL全局解释器锁本质上是一个互斥锁，其与所有互斥锁的本质都一样,都是并发运行变成串行，以此来控制同一时间内共享数据只能被一个任务所修改，进而保证数据安全。

解释器的代码是所有线程共享的，所以垃圾回收线程也可能访问到解释器的代码而去执行，这就导致了一个问题:对于同一个数据100，可能线程1执行x=100的同时，而垃圾回收执行的是回收100的操作，解决这种问题没有什么高明的方法，就是加锁处理，如下图的GIL，保证python解释器同一时间只能执行一个任务的代码
GIL 与Lock是两把锁，保护的数据不一样，前者是解释器级别的（当然保护的就是解释器级别的数据，比如垃圾回收的数据），后者是保护用户自己开发的应用程序的数据，很明显GIL不负责这件事，只能用户自定义加锁处理
ps:
1、cpu到底是用来做计算的，还是用来做I/O的？
2、多cpu，意味着可以有多个核并行完成计算，所以多核提升的是计算性能
3、每个cpu一旦遇到I/O阻塞，仍然需要等待，所以多核对I/O操作没什么用处
并发的任务是计算密集型:多进程效率高
并发的任务是I/O密集型:多线程效率高

七、死锁现象与递归锁
# 死锁现象

from threading import Thread,Lock
import time


class MyThread(Thread, object):

    def __init__(self, mutexA, mutexB):
        super(MyThread, self).__init__()
        self.mutexA = mutexA
        self.mutexB = mutexB

    def func1(self):
        self.mutexA.acquire()
        print("%s拿到A锁"% self.name)

        self.mutexB.acquire()
        print("%s拿到B锁"% self.name)
        self.mutexB.release()

        self.mutexA.release()

    def func2(self):
        self.mutexB.acquire()
        print("%s拿到B锁"% self.name)
        time.sleep(2)

        self.mutexA.acquire()
        print("%s拿到A锁"% self.name)
        self.mutexA.release()

        self.mutexB.release()

    def run(self):
        self.func1()
        self.func2()

if __name__ == ‘__main__‘:
    mutexA = Lock()
    mutexB = Lock()
    for i in range(5):
        t = MyThread(mutexA, mutexB)
        t.start()
‘‘‘
打印结果:
Thread-1拿到A锁
Thread-1拿到B锁
Thread-1拿到B锁
Thread-2拿到A锁
“然后一直阻塞住”
‘‘‘

死锁现象如何导致的呢?
因为当线程1先执行了self.func1(),然后拿到了A锁, 在拿到了B锁,self.func1()执行完毕后,执行self.func2(),然后拿到了B锁,然后执行time.sleep(2)进行2s的睡眠,在睡眠的过程中,另外一个线程就拿到了self.func1()中的A锁,当线程1睡了2s过后,准备去拿self.func2()中的A锁时,发现A锁被另外一个线程拿到了,就在那儿等待A锁被释放,而线程1中拿到的B锁一直未释放,另外一个线程无法拿到B锁,就无法释放A锁,这样就一直阻塞着,无法往下继续执行。这就导致了死锁现象。

而递归锁就很好的解决了这一个问题。
递归锁内部是通过count和Lock来实现的,通过count来计算Lock的acquire次数，当程序acquire一次,count便加1,当release一次,count便减1,当count为0时, 此锁才能被其他线程所拿到。
递归锁与互斥锁的最大区别便是, 递归锁能在锁未release()时也能acquire, 就是能连续acquire(),而互斥锁不能。

上面的程序就可以将锁变成 mutexA = mutexB = RLock()

from threading import Thread,RLock
import time


class MyThread(Thread, object):

    def __init__(self, mutexA, mutexB):
        super(MyThread, self).__init__()
        self.mutexA = mutexA
        self.mutexB = mutexB

    def func1(self):
        self.mutexA.acquire()
        print("%s拿到A锁"% self.name)

        self.mutexB.acquire()
        print("%s拿到B锁"% self.name)
        self.mutexB.release()

        self.mutexA.release()

    def func2(self):
        self.mutexB.acquire()
        print("%s拿到B锁"% self.name)
        time.sleep(2)

        self.mutexA.acquire()
        print("%s拿到A锁"% self.name)
        self.mutexA.release()

        self.mutexB.release()

    def run(self):
        self.func1()
        self.func2()

if __name__ == ‘__main__‘:
    mutexA = mutexB = RLock()

    for i in range(5):
        t = MyThread(mutexA, mutexB)
        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-4拿到A锁
Thread-4拿到B锁
Thread-4拿到B锁
Thread-4拿到A锁
Thread-3拿到A锁
Thread-3拿到B锁
Thread-3拿到B锁
Thread-3拿到A锁
Thread-5拿到A锁
Thread-5拿到B锁
Thread-5拿到B锁
Thread-5拿到A锁
‘‘‘

八、信号量
信号量本质上也是一把锁,与互斥锁不同的是，信号量可以规定多少个线程去拿到这把锁,如果将信号量设为5,那么就可以同时有5个线程去拿到这把锁进行使用,而这个设定的值便是信号量的大小
ps:
Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。

from threading import Thread,Semaphore,current_thread
import time


def task():
    smlock.acquire()
    print("%s get smlock" %current_thread().getName())
    time.sleep(0.5)
    smlock.release()

if __name__ == ‘__main__‘:
    smlock = Semaphore(5)
    for i in range(20):
        t = Thread(target=task, )
        t.start()
‘‘‘
打印结果:
Thread-1 get smlock
Thread-2 get smlock
Thread-3 get smlock
Thread-4 get smlock
Thread-5 get smlock
Thread-7 get smlock
Thread-8 get smlock
Thread-10 get smlock
Thread-9 get smlock
Thread-6 get smlock
Thread-11 get smlock
Thread-13 get smlock
Thread-15 get smlock
Thread-14 get smlock
Thread-12 get smlock
Thread-16 get smlock
Thread-18 get smlock
Thread-17 get smlock
Thread-19 get smlock
Thread-20 get smlock
‘‘‘

Eevent事件
event.isSet()：返回event的状态值；
event.wait()：如果 event.isSet()==False将阻塞线程；
event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
event.clear()：恢复event的状态值为False
定时器 Timer
指定某个任务n秒后执行

from threading import Timer

def hello():
    print("hello, world")

t = Timer(1, hello)
t.start()

九、线程池与进程池

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


def task(name):
    print("%s is running pid:%s" %(name, os.getpid()))
    time.sleep(random.randint(1,4))


if __name__ == ‘__main__‘:
    pool = ThreadPoolExecutor(4)
    for i in range(10):
        pool.submit(task, "xu{}".format(i))
    pool.shutdown(wait=True) # shutdown操作过后,就不能再向线程池中提交任务。
    # pool.submit(task, "sy")
    # 如果提交了便会报出错误:RuntimeError: cannot schedule new futures after shutdown
    print("主")

‘‘‘
打印结果:
xu0 is running pid:1306
xu1 is running pid:1306
xu2 is running pid:1306
xu3 is running pid:1306
xu4 is running pid:1306
xu5 is running pid:1306
xu6 is running pid:1306
xu7 is running pid:1306
xu8 is running pid:1306
xu9 is running pid:1306
主
‘‘‘

创建进程池与线程池相同,只是将pool = ThreadPoolExecutor(4)改成pool=ProcessPoolExecutor(4)便行了
练习1:多线程练习(利用queue+Thread模块实现线程池的套接字通信)

# server端
from threading import Thread
import queue
‘‘‘
利用queue+Thread模块实现线程池的思路：
1、创建队列, 然后设置队列的大小，此队列的大小就是线程池的大小
2、然后通过for循环，将需要开启线程的函数名加入队列中
3、然后每进来一个客户端,便从队列中取出一个函数名,然后将取出的函数名赋值给target
4、当一个客户端断开连接时,再向队列中put一个函数名
‘‘‘


class MyServer(object,):

    def __init__(self, nums):
        self.nums = nums
        self.q = self.__add_q()

    def __add_q(self):
        q = queue.Queue()
        for i in range(self.nums):
            q.put(self.multy)
        return q

    def run(self):
        self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server.bind(("127.0.0.1", 8080))
        self.server.listen(5)
        while True:
            client, addr = self.server.accept()
            print(client)
            execute_func = self.q.get()
            t = Thread(target=execute_func, args=(client, addr))
            t.start()

    def multy(self, client, addr):
        global n
        n = 100
        while True:
            data = client.recv(1024).decode("utf-8")
            if not data:
                self.q.put(self.multy)
                client.close()
                break
            send_data = data.upper()
            print(addr, send_data)
            n -= 1
            print(n)
            client.send(send_data.encode("utf-8"))

if __name__ == ‘__main__‘:
    obj = MyServer(3)
    obj.run()
    
# 客户端
from socket import *


conn = socket(AF_INET, SOCK_STREAM)

conn.connect(("127.0.0.1", 8080))

while True:
    inp = input(">>>:")
    conn.send(inp.encode("utf-8"))
    res = conn.recv(1024)
    print(res.decode("utf-8"))

练习2:使用线程池爬取多个网站的数据(通过此练习引入线程池下的回调函数)

import requests
import time
from concurrent.futures import ThreadPoolExecutor


def recv_data(url):
    response = requests.get(url)
    time.sleep(1)
    return {"url":url, "result":response.text}

def len_data(res):
    result = res.result()  # 获取recv_data的传输过来的数据
    print("url:%s len_data:%s" %(result["url"], len(result["result"])))


if __name__ == ‘__main__‘:
    urls = [
        "https://home.firefoxchina.cn/",
        "https://www.baidu.com/",
        "https://www.cnblogs.com/"
    ]
    thread_pool = ThreadPoolExecutor(2)
    for url in urls:
        thread_pool.submit(recv_data, url).add_done_callback(len_data)
        ‘‘‘
        当线程1将recv_data执行完毕后,会自动调用len_data函数,将recv_data返回值的对象传给res，
        然后在调用result方法获取传输过来的数据。
        ‘‘‘