python3全栈开发-并发编程，多线程

Posted 2020-11-01 明天的希望，让我们忘了今天的痛苦。

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了python3全栈开发-并发编程，多线程相关的知识，希望对你有一定的参考价值。

一、什么是线程　　

　　在传统操作系统中，每个进程有一个地址空间，而且默认就有一个控制线程

　　线程顾名思义，就是一条流水线工作的过程，一条流水线必须属于一个车间，一个车间的工作过程是一个进程

　　车间负责把资源整合到一起，是一个资源单位，而一个车间内至少有一个流水线

　　流水线的工作需要电源，电源就相当于cpu

　　所以，进程只是用来把资源集中到一起（进程只是一个资源单位，或者说资源集合），而线程才是cpu上的执行单位。

　　多线程（即多个控制线程）的概念是，在一个进程中存在多个控制线程，多个控制线程共享该进程的地址空间，相当于一个车间内有多条流水线，都共用一个车间的资源。

　　例如，北京地铁与上海地铁是不同的进程，而北京地铁里的13号线是一个线程，北京地铁所有的线路共享北京地铁所有的资源，

二、线程的创建开销小

创建进程的开销要远大于线程？

　　如果我们的软件是一个工厂，该工厂有多条流水线，流水线工作需要电源，电源只有一个即cpu（单核cpu）

　　一个车间就是一个进程，一个车间至少一条流水线（一个进程至少一个线程）

　　创建一个进程，就是创建一个车间（申请空间，在该空间内建至少一条流水线）

　　而建线程，就只是在一个车间内造一条流水线，无需申请空间，所以创建开销小

进程之间是竞争关系，线程之间是协作关系？

车间直接是竞争/抢电源的关系，竞争（不同的进程直接是竞争关系，是不同的程序员写的程序运行的，迅雷抢占其他进程的网速，360把其他进程当做病毒干死）
一个车间的不同流水线式协同工作的关系（同一个进程的线程之间是合作关系，是同一个程序写的程序内开启动，迅雷内的线程是合作关系，不会自己干自己）

三、线程与进程的区别

线程共享创建它的进程的地址空间;进程有自己的地址空间。
线程可以直接访问其进程的数据段;进程有自己的父进程数据段的副本。
线程可以直接与进程的其他线程通信;进程必须使用进程间通信来与同胞进程通信。
新线程很容易创建;新进程需要父进程的重复。线程可以对相同进程的线程进行相当大的控制;
进程只能对子进程进行控制。对主线程的更改(取消、优先级更改等)可能会影响进程的其他线程的行为;
对父进程的更改不会影响子进程。

四、为何要用多线程

多线程指的是，在一个进程中开启多个线程，简单的讲：如果多个任务共用一块地址空间，那么必须在一个进程内开启多个线程。详细的讲分为4点：

　 1. 多线程共享一个进程的地址空间

2. 线程比进程更轻量级，线程比进程更容易创建可撤销，在许多操作系统中，创建一个线程比创建一个进程要快10-100倍，在有大量线程需要动态和快速修改时，这一特性很有用

3. 若多个线程都是cpu密集型的，那么并不能获得性能上的增强，但是如果存在大量的计算和大量的I/O处理，拥有多个线程允许这些活动彼此重叠运行，从而会加快程序执行的速度。

应用举例

　　开启一个字处理软件进程，该进程肯定需要办不止一件事情，比如监听键盘输入，处理文字，定时自动将文字保存到硬盘，这三个任务操作的都是同一块数据，因而不能用多进程。只能在一个进程里并发地开启三个线程,如果是单线程，那就只能是，键盘输入时，不能处理文字和自动保存，自动保存时又不能输入和处理文字。

五、threading模块介绍

multiprocess模块的完全模仿了threading模块的接口，二者在使用层面，有很大的相似性，因而不再详细介绍

官网链接：https://docs.python.org/3/library/threading.html?highlight=threading#

六、开启线程的两种方式

#方式一
from threading import Thread
import time
def run(name):
    time.sleep(2)
    print(\'%s say hello\' %name)

if __name__ == \'__main__\':
    t=Thread(target=run,args=(\'duoduo\',))
    t.start()
    print(\'主线程\')  #在python中没有主线程和子线程的区别，只是这样叫来好区分

方式一

#方式二
from threading import Thread
import time
class Fun(Thread):
    def run(self):               #这里的run可执行
        time.sleep(2)
        print(\'%s say hello\' % self.name)


if __name__ == \'__main__\':
    t = Fun()                  
    t.start()
    print(\'主线程\')

方式二

七、在一个进程下开启多个线程与在一个进程下开启多个子进程的区别

from threading import Thread
from multiprocessing import Process
import os

def work():
    print(\'hello\')

if __name__ == \'__main__\':
    #在主进程下开启线程
    t=Thread(target=work)
    t.start()
    print(\'主线程/主进程\')
    \'\'\'
    打印结果:
    hello
    主线程/主进程
    \'\'\'

    #在主进程下开启子进程
    t=Process(target=work)
    t.start()
    print(\'主线程/主进程\')
    \'\'\'
    打印结果:
    主线程/主进程
    hello
    \'\'\'

1 谁开启的速度快

from threading import Thread
from multiprocessing import Process
import os

def work():
    print(\'hello\',os.getpid())

if __name__ == \'__main__\':
    #part1:在主进程下开启多个线程,每个线程都跟主进程的pid一样
    t1=Thread(target=work)
    t2=Thread(target=work)
    t1.start()
    t2.start()
    print(\'主线程/主进程pid\',os.getpid())

    #part2:开多个进程,每个进程都有不同的pid
    p1=Process(target=work)
    p2=Process(target=work)
    p1.start()
    p2.start()
    print(\'主线程/主进程pid\',os.getpid())

2 看一看pid

from  threading import Thread
from multiprocessing import Process
import os
def work():
    global n
    n=0

if __name__ == \'__main__\':
    # n=100
    # p=Process(target=work)
    # p.start()
    # p.join()
    # print(\'主\',n) #毫无疑问子进程p已经将自己的全局的n改成了0,但改的仅仅是它自己的,查看父进程的n仍然为100


    n=1
    t=Thread(target=work)
    t.start()
    t.join()
    print(\'主\',n) #查看结果为0,因为同一进程内的线程之间共享进程内的数据

3 同一进程内的线程共享该进程的数据？

八、线程相关的其他方法

Thread实例对象的方法
  # isAlive(): 返回线程是否活动的。
  # getName(): 返回线程名。
  # setName(): 设置线程名。

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

一些方法

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

def work():
    import time
    time.sleep(3)
    print(threading.current_thread().getName())


if __name__ == \'__main__\':
    #在主进程下开启线程
    t=Thread(target=work)
    t.start()

    print(threading.current_thread().getName())
    print(threading.current_thread()) #主线程
    print(threading.enumerate()) #连同主线程在内有两个运行的线程
    print(threading.active_count())
    print(\'主线程/主进程\')

    \'\'\'
    打印结果:
    MainThread
    <_MainThread(MainThread, started 140735268892672)>
    [<_MainThread(MainThread, started 140735268892672)>, <Thread(Thread-1, started 123145307557888)>]
    主线程/主进程
    Thread-1
    \'\'\'

方法的应用

主线程等待子线程结束

from threading import Thread
import time
def func(name):
    time.sleep(2)
    print(\'%s say hello\' %name)

if __name__ == \'__main__\':
    t=Thread(target=func,args=(\'duoduo\',))
    t.start()
    t.join()
    print(\'主线程\')
    print(t.is_alive())     #join串行
    \'\'\'
    duoduo say hello
    主线程
    False
    \'\'\'

join串行的结果

九、守护线程

无论是进程还是线程，都遵循：守护xxx会等待主xxx运行完毕后被销毁

需要强调的是：运行完毕并非终止运行

1.对主进程来说，运行完毕指的是主进程代码运行完毕
2.对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕

详细解释：

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

from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print(\'%s say hello\' %name)

if __name__ == \'__main__\':
    t=Thread(target=sayhi,args=(\'duoduo\',))
    t.daemon=True#必须在t.start()之前设置
    t.start()

    print(\'主线程\')
    print(t.is_alive())
    \'\'\'
    主线程
    True
    \'\'\'

守护线程

十、线程互斥锁

#因为Python解释器帮你自动定期进行内存回收，你可以理解为python解释器里有一个独立的线程，每过一段时间它起wake up做一次全局轮询看看哪些内存数据是可以被清空的，此时你自己的程序 里的线程和 py解释器自己的线程是并发运行的，假设你的线程删除了一个变量，py解释器的垃圾回收线程在清空这个变量的过程中的clearing时刻，可能一个其它线程正好又重新给这个还没来及得清空的内存空间赋值了，结果就有可能新赋值的数据被删除了，为了解决类似的问题，python解释器简单粗暴的加了锁，即当一个线程运行时，其它人都不能动，这样就解决了上述的问题，这可以说是Python早期版本的遗留问题。

引子

from threading import Thread
import os,time
def work():
    global n
    temp=n
    time.sleep(0.1)
    n=temp-1
if __name__ == \'__main__\':
    n=100
    l=[]
    for i in range(100):
        p=Thread(target=work)
        l.append(p)
        p.start()
    for p in l:
        p.join()

    print(n) #结果可能为99

引子例子

锁通常被用来实现对共享资源的同步访问。为每一个共享资源创建一个Lock对象，当你需要访问该资源时，调用acquire方法来获取锁对象（如果其它线程已经获得了该锁，则当前线程需等待其被释放），待资源访问完后，再调用release方法释放锁：

import threading
R=threading.Lock()
R.acquire()
\'\'\'
对公共数据的操作
\'\'\'
R.release()

格式

from threading import Thread,Lock
import os,time
def work():
    global n
    lock.acquire()
    temp=n
    time.sleep(0.1)
    n=temp-1
    lock.release()
if __name__ == \'__main__\':
    lock=Lock()
    n=100
    l=[]
    for i in range(100):
        p=Thread(target=work)
        l.append(p)
        p.start()
    for p in l:
        p.join()

    print(n) #结果肯定为0，由原来的并发执行变成串行，牺牲了执行效率保证了数据安全

加锁的例子

#不加锁:并发执行,速度快,数据不安全
from threading import current_thread,Thread,Lock
import os,time
def task():
    global n
    print(\'%s is running\' %current_thread().getName())
    temp=n
    time.sleep(0.5)
    n=temp-1


if __name__ == \'__main__\':
    n=100
    lock=Lock()
    threads=[]
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()

    stop_time=time.time()
    print(\'主:%s n:%s\' %(stop_time-start_time,n))

\'\'\'
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:0.5216062068939209 n:99
\'\'\'


#不加锁:未加锁部分并发执行,加锁部分串行执行,速度慢,数据安全
from threading import current_thread,Thread,Lock
import os,time
def task():
    #未加锁的代码并发运行
    time.sleep(3)
    print(\'%s start to run\' %current_thread().getName())
    global n
    #加锁的代码串行运行
    lock.acquire()
    temp=n
    time.sleep(0.5)
    n=temp-1
    lock.release()

if __name__ == \'__main__\':
    n=100
    lock=Lock()
    threads=[]
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    stop_time=time.time()
    print(\'主:%s n:%s\' %(stop_time-start_time,n))

\'\'\'
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:53.294203758239746 n:0
\'\'\'

#有的同学可能有疑问:既然加锁会让运行变成串行,那么我在start之后立即使用join,就不用加锁了啊,也是串行的效果啊
#没错:在start之后立刻使用jion,肯定会将100个任务的执行变成串行,毫无疑问,最终n的结果也肯定是0,是安全的,但问题是
#start后立即join:任务内的所有代码都是串行执行的,而加锁,只是加锁的部分即修改共享数据的部分是串行的
#单从保证数据安全方面,二者都可以实现,但很明显是加锁的效率更高.
from threading import current_thread,Thread,Lock
import os,time
def task():
    time.sleep(3)
    print(\'%s start to run\' %current_thread().getName())
    global n
    temp=n
    time.sleep(0.5)
    n=temp-1


if __name__ == \'__main__\':
    n=100
    lock=Lock()
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        t.start()
        t.join()
    stop_time=time.time()
    print(\'主:%s n:%s\' %(stop_time-start_time,n))

\'\'\'
Thread-1 start to run
Thread-2 start to run
......
Thread-100 start to run
主:350.6937336921692 n:0 #耗时是多么的恐怖
\'\'\'

互斥锁与join的区别（重点！！！）

十一、死锁现象与递归锁

进程也有死锁与递归锁，在进程那里忘记说了，放到这里一切说了额

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

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(\'\\033[41m%s 拿到A锁\\033[0m\' %self.name)

        mutexB.acquire()
        print(\'\\033[42m%s 拿到B锁\\033[0m\' %self.name)
        mutexB.release()

        mutexA.release()

    def func2(self):
        mutexB.acquire()
        print(\'\\033[43m%s 拿到B锁\\033[0m\' %self.name)
        time.sleep(2)

        mutexA.acquire()
        print(\'\\033[44m%s 拿到A锁\\033[0m\' %self.name)
        mutexA.release()

        mutexB.release()

if __name__ == \'__main__\':
    for i in range(10):
        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，则不会发生死锁：

mutexA=mutexB=threading.RLock() #一个线程拿到锁，counter加1,该线程内又碰到加锁的情况，则counter继续加1，
这期间所有其他线程都只能等待，等待该线程释放所有锁，即counter递减到0为止

from threading import Thread,Lock,RLock
import time

# mutexA=Lock()
# mutexB=Lock()

mutexA=mutexB=RLock()

class MyThread(Thread):
    def run(self):
        self.f1()
        self.f2()

    def f1(self):
        mutexA.acquire()
        print(\'%s 拿到了A锁\' %self.name)
        mutexB.acquire()
        print(\'%s 拿到了B锁\' %self.name)
        mutexB.release()
        mutexA.release()

    def f2(self):
        mutexB.acquire()
        print(\'%s 拿到了B锁\' %self.name)
        time.sleep(0.1)
        mutexA.acquire()
        print(\'%s 拿到了A锁\' %self.name)
        mutexA.release()
        mutexB.release()


if __name__ == \'__main__\':
    for i in range(10):
        t=MyThread()
        t.start()

如何解死锁

最牛逼的机器抢锁的速度：

十二、信号量Semaphore

同进程的一样

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

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

from threading import Thread,Semaphore,current_thread
import time,random

sm=Semaphore(5)

def go_wc():
    sm.acquire()
    print(\'%s 上厕所ing\' %current_thread().getName())
    time.sleep(random.randint(1,3))
    sm.release()

if __name__ == \'__main__\':
    for i in range(23):
        t=Thread(target=go_wc)
        t.start()