Python 多线程、多进程

旧城等待， 2021-12-01 19:06 401阅读 0赞

## 什么是线程？ ##

py文件在执行程序中，他会根据程序的编写来区分，假如没有创建子进程，整个程序就是主进程。

那程序中，有主线程而且还有子线程，那他就是一个多线程。

使用多线程可以提升I/O密集型的效率。

## 什么是进程？ ##

py文件就是一个进程，比如：QQ，360，浏览器。

使用多进程，会消耗很大的资源问题。

## GIL锁 ##

GIL锁又称，全局解释器锁。

GIL锁的作用：在同一时刻，只能有一个线程进入解释器。

站在开发Python语言的那一端时，他就是一个神器，而站在使用这门语言这一端时，他就是一个BUG般的存在，而这BUG仅存在于CPython中。

为什么说BUG呢，因为有了GIL锁，我们使用多线程在进行计算密集型中，计算机的核数再多，他也只能使用一核。

## I/O密集型，计算密集型 ##

什么是I/O密集型？

说白了，他就是一个堵塞，当我们创建多线程（A、B），A线程在执行，遇到了堵塞，在CPU空闲时，切换到了B线程。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import threading
    import time
    
    start  = time.time()
    def music():
        print('I listening Music')
        time.sleep(2)
    
    def movie():
        print('I watching TV')
        time.sleep(3)
    
    t1 = threading.Thread(target=music)
    t2 = threading.Thread(target=movie)
    t1.start()
    t2.start()
    end = time.time()
    result = end - start
    print(result)
    print('主线程结束')

什么时计算密集型？

线程在计算过程中，没有遇到堵塞，而是一直在执行计算。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

def add():
        num = 0
        for i in range(1000000):
            num += i
        print(num)

## 如何创建多线程？ ##

Threading模块

函数创建

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import threading
    import time
    
    start  = time.time()
    def music():
        for i in range(3):
            print('I listening Music')
            time.sleep(1)
    
    def movie():
        for i in range(2):
            print('I watching TV')
            time.sleep(5)
    
    t1 = threading.Thread(target=music) #创建子线程
    t2 = threading.Thread(target=movie) #创建子线程
    threads = [t1,t2]
    
    for t in threads:
        t.start() #启动子线程

类创建

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import threading
    
    class MyThread(threading.Thread): # 首先要继承这个方法
        def __init__(self,count):
            super().__init__()
            self.count = count
        def current_thread_count(self):
            print(self.count)
        def run(self): #定义每个线程要运行的内容
            self.current_thread_count()
    
    t1 = MyThread(threading.active_count())
    t2 = MyThread(threading.active_count())
    t1.start() #开启线程
    t2.start()

join ()方法

主线程A中，创建了子线程B，并且在主线程A中调用了B.join()，那么，主线程A会在调用的地方等待，直到子线程B完成操作后，

才可以接着往下执行，那么在调用这个线程时可以使用被调用线程的join方法。join(\[timeout\]) 里面的参数时可选的，代表线程运行的最大时

间，即如果超过这个时间，不管这个此线程有没有执行完毕都会被回收，然后主线程或函数都会接着执行的，如果线程执行时间小于参数表示的

时间，则接着执行，不用一定要等待到参数表示的时间。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

setDaemon()方法

主线程A中，创建了子线程B，并且在主线程A中调用了B.setDaemon(),这个的意思是，把主线程A设置为守护线程，这

时候，要是主线程A执行结束了，就不管子线程B是否完成,一并和主线程A退出.这就是setDaemon方法的含义，这基本和join是相反的。此外，还有

个要特别注意的：必须在start() 方法调用之前设置，如果不设置为守护线程，程序会被无限挂起，只有等待了所有线程结束它才结束。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import threading
    import time
    
    start  = time.time()
    def music():
        for i in range(3):
            print('I listening Music')
            time.sleep(1)
    
    def movie():
        for i in range(2):
            print('I watching TV')
            time.sleep(5)
    
    t1 = threading.Thread(target=music) #创建子线程
    t2 = threading.Thread(target=movie) #创建子线程
    threads = [t1,t2]
    t2.setDaemon(1) #守护线程
    
    for t in threads:
        t.start() #启动子线程
    print('主线程结束')

## 同步锁 ##

**为什么会有同步锁？**

当我们创建多线程时，并且有一个全局变量，而多线程操作这个全局变量。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import threading
    import time
    
    def sub():
        global number
        num = number
        time.sleep(0.1)
        number = num - 1
    
    number = 10
    threads = []
    
    for i in range(10):
        t = threading.Thread(target=sub)
        t.start()
        threads.append(t)
    for i in threads:
        i.join()
    print(number) # 9

结果并不是我们想要的。

**为什么出现这种问题？**

程序在sleep的一瞬间，cpu来回切换，还没等着修改全局变量，所有的线程已经被创建，而且也已经被赋值。

**如何解决？**

那就是加锁了。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import threading
    import time
    
    def sub():
        global number
    
        r.acquire() # 获得锁
        num = number
        time.sleep(0.1)
        number = num - 1
        r.release() # 释放锁
    
    number = 10
    threads = []
    r = threading.Lock()
    
    for i in range(10):
        t = threading.Thread(target=sub)
        t.start()
        threads.append(t)
    for i in threads:
        i.join()
    print(number) # 0

加锁，其实就是不让cup进行线程切换，直到锁被释放。

如果锁没被释放，不会让其他线程进入，也不会影响不进入线程的执行。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import threading
    import time
    
    number = 10
    threads = []
    r = threading.Lock()
    
    def sub():
        global number
        r.acquire()
        num = number
        time.sleep(0.1)
        number = num - 1
        r.release()
    
    def music():
        time.sleep(0.5)
        print('Music')
    
    t = threading.Thread(target=music)
    t.start()
    for i in range(10):
        t = threading.Thread(target=sub)
        t.start()
        threads.append(t)
    for i in threads:
        i.join()
    print(number)

## 递归锁(Rlock) ##

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import threading
    
    r = threading.Lock()
    
    class MyThread(threading.Thread):
        def Thread_1(self):
            r.acquire()
            print('第一层',self.name)
            r.acquire()
            print('第二层',self.name)
            r.release()
            r.release()
        def run(self):
            self.Thread_1()
    
    for i in range(5):
        t = MyThread()
        t.start()

死锁

递归锁，与Lock很相似，但是他有一个计数的功能，能解决死锁

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import threading
    
    r = threading.RLock()
    
    class MyThread(threading.Thread):
        def Thread_1(self):
            r.acquire()
            print('第一层',self.name)
            r.acquire()
            print('第二层',self.name)
            r.release()
            r.release()
        def Thread_2(self):
            r.acquire()
            print('第一层',self.name)
            r.acquire()
            print('第二层',self.name)
            r.release()
            r.release()
        def run(self):
            self.Thread_1()
            self.Thread_2()
    
    for i in range(5):
        t = MyThread()
        t.start()

## 信号量(Semaphore) ##

信号量相当于，可以限制最大进入的线程数量。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import threading
    import time
    r = threading.Semaphore(2) # 创建信号量，最大进入的线程数量
    
    class MyThread(threading.Thread):
        def Thread_1(self):
    
            r.acquire() # 每次进入线程+1，但不能超过信号量设定的值
            print(self.name)
            time.sleep(2)
            r.release() # -1
        def run(self):
            self.Thread_1()
    
    for i in range(5):
        t = MyThread()
        t.start()

## 条件变量(Conditon) ##

wait()：条件不满足时调用，线程会释放锁并进入等待阻塞。

notify()：条件创造后调用，通知等待池激活一个线程。

notifyAll()：条件创造后调用，通知等待池激活所有线程。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import threading
    import time
    import random
    
    def producer():
        time.sleep(0.2)
        global F
        while True:
            if con_Lock.acquire():
                r = random.randrange(0,100)
                F.append(r)
                print(str(threading.current_thread())+ '--->' + str(r))
                con_Lock.notify()
                con_Lock.release()
            time.sleep(3)
    
    def consumer():
        global F
        while True:
            con_Lock.acquire()
            if not F:
                print("老板，快点，没有包子了")
                con_Lock.wait()
            a = F.pop()
            print('包子%s已经被吃'%a)
            time.sleep(0.5)
    
    con_Lock = threading.Condition()
    threads = []
    F = []
    
    for i in range(5):
        threads.append(producer)
    threads.append(consumer)
    for i in threads:
        t = threading.Thread(target=i)
        t.start()

## Event ##

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from threading import Event
    Event.isSet() #返回event的状态值
    Event.wait() #如果 event.isSet()==False将阻塞线程；
    Event.set() #设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
    Event.clear() #恢复

实例

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#首先定义两个函数，一个是连接数据库
    # 一个是检测数据库
    from threading import Thread,Event,currentThread
    import time
    e = Event()
    def conn_mysql():
        '''链接数据库'''
        count = 1
        while not e.is_set():  #当没有检测到时候
            if count >3: #如果尝试次数大于3，就主动抛异常
                raise ConnectionError('尝试链接的次数过多')
            print('\033[45m%s 第%s次尝试'%(currentThread(),count))
            e.wait(timeout=1) #等待检测（里面的参数是超时1秒）
            count+=1
        print('\033[44m%s 开始链接...'%(currentThread().getName()))
    def check_mysql():
        '''检测数据库'''
        print('\033[42m%s 检测mysql...' % (currentThread().getName()))
        time.sleep(5)
        e.set()
    if __name__ == '__main__':
        for i  in range(3):  #三个去链接
            t = Thread(target=conn_mysql)
            t.start()
        t = Thread(target=check_mysql)
        t.start()

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from  threading import Thread,Event,currentThread
    import time
    e = Event()
    def traffic_lights():
        '''红绿灯'''
        time.sleep(5)
        e.set()
    def car():
        '''车'''
        print('\033[42m %s 等绿灯\033[0m'%currentThread().getName())
        e.wait()
        print('\033[44m %s 车开始通行' % currentThread().getName())
    if __name__ == '__main__':
        for i in range(10):
            t = Thread(target=car)  #10辆车
            t.start()
        traffic_thread = Thread(target=traffic_lights)  #一个红绿灯
        traffic_thread.start()
    
    红绿灯

## 队列(Queue) ##

**Python Queue模块有三种队列及构造函数:**  
1、Python Queue模块的FIFO队列先进先出。 class queue.Queue(maxsize)  
2、LIFO类似于堆，即先进后出。 class queue.LifoQueue(maxsize)  
3、还有一种是优先级队列级别越低越先出来。 class queue.PriorityQueue(maxsize)

**此包中的常用方法(q = Queue.Queue()):**  
q.qsize() 返回队列的大小  
q.empty() 如果队列为空，返回True,反之False  
q.full() 如果队列满了，返回True,反之False  
q.full 与 maxsize 大小对应  
q.get(\[block\[, timeout\]\]) 获取队列，timeout等待时间  
q.get\_nowait() 相当q.get(False)  
非阻塞 q.put(item) 写入队列，timeout等待时间  
q.put\_nowait(item) 相当q.put(item, False)  
q.task\_done() 在完成一项工作之后，q.task\_done() 函数向任务已经完成的队列发送一个信号  
q.join() 实际上意味着等到队列为空，再执行别的操作

![ContractedBlock.gif][]

##  ##

## 如何创建多进程？ ##

multiprocessing 模块

Process

**构造方法：**

Process(\[group \[, target \[, name \[, args \[, kwargs\]\]\]\]\])

group: 线程组，目前还没有实现，库引用中提示必须是None；  
　　target: 要执行的方法；  
　　name: 进程名；  
　　args/kwargs: 要传入方法的参数。

**实例方法：**

is\_alive()：返回进程是否在运行。

join(\[timeout\])：阻塞当前上下文环境的进程程，直到调用此方法的进程终止或到达指定的timeout（可选参数）。

start()：进程准备就绪，等待CPU调度

run()：strat()调用run方法，如果实例进程时未制定传入target，这star执行t默认run()方法。

terminate()：不管任务是否完成，立即停止工作进程

**属性：**

authkey

daemon：和线程的setDeamon功能一样

exitcode(进程在运行时为None、如果为–N，表示被信号N结束）

name：进程名字。

pid：进程号。

多进程与多线程的创建完全一致，也有两种方式。

函数创建

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from multiprocessing import Process
    import os
    
    def My_process():
        print(os.getpid()) # 获取进程和PID
    
    process_num = []
    if __name__ == "__main__":
        for i in range(3):
            p = Process(target=My_process,args=())
            p.start()
            process_num.append(p)
        for p in process_num:
            p.join()
        print('主进程结束')

类创建

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from multiprocessing import Process
    import time
    
    class My_process(Process):# 继承
        def __init__(self,name):
            super().__init__() # 调用父类的__init__方法
            self.name = name
        def run(self):
            print(self.name)
    
    if __name__ == '__main__':
        for i in range(3):
            p = My_process(str(i))
            p.daemon = True # 启动守护进程
            p.start()
        time.sleep(0.5)
        print('主进程结束')

注意：多进程开启时

windows必须在，if \_\_name\_\_ == "\_\_main\_\_": 下开启

linux下不用

## 进程之间通讯 ##

Queue（队列）

这里的Queue跟threading的用法类似:

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from multiprocessing import Process,Queue
    
    q = Queue() # 创建队列
    
    def multi_process(i,q):
        q.put(i)
    
    process_obj = []
    
    def run():
        for i in range(3):
            p = Process(target=multi_process,args=(i,q))
            p.start()
            process_obj.append(p)
        for p in process_obj:
            p.join()
    if __name__ == "__main__":
        run()
        for i in range(3):
            print(q.get())

Pipe（管道）

The [`Pipe()`][Pipe] function returns a pair of connection objects connected by a pipe which by default is duplex (two-way). For example:

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import os
    from multiprocessing import Process,Pipe
    
    parent_port,child_port = Pipe()
    
    def multi_process(port):
        Pid = os.getpid()
        print(port.recv())
        port.send('我的Pid是：%s'%Pid)
        port.close()
    
    process_obj = []
    
    def run():
        for i in range(3):
            p = Process(target=multi_process,args=(child_port,))
            p.start()
            process_obj.append(p)
        for p in process_obj:
            parent_port.send('你的Pid是多少?')
            print(parent_port.recv())
            p.join()
    
    if __name__ == "__main__":
        run()

The two connection objects returned by [`Pipe()`][Pipe] represent the two ends of the pipe. Each connection object has [`send()`][send] and [`recv()`][recv] methods (among others). Note that data in a pipe may become corrupted if two processes (or threads) try to read from or write to the *same* end of the pipe at the same time. Of course there is no risk of corruption from processes using different ends of the pipe at the same time.

Manager

A manager object returned by `Manager()` controls a server process which holds Python objects and allows other processes to manipulate them using proxies.

A manager returned by `Manager()` will support types [`list`][list], [`dict`][dict], [`Namespace`][Namespace], [`Lock`][Lock], [`RLock`][RLock], [`Semaphore`][Semaphore], [`BoundedSemaphore`][BoundedSemaphore], [`Condition`][Condition], [`Event`][Event], [`Barrier`][Barrier], [`Queue`][Queue], [`Value`][Value] and [`Array`][Array]. For example,

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import os
    from multiprocessing import Process,Manager
    
    def multi_process(*args):
       for arg in args:
           try:
               arg[os.getpid()] = os.getpid()
           except:
               arg.append(os.getpid())
    
    def start(*args):
        for i in range(3):
            p = Process(target=multi_process,args=(args))
            p.start()
        return p
    if __name__ == "__main__":
        with Manager() as manager:
            l = manager.list()
            d = manager.dict()
            p = start(l,d)
            p.join()
            print(l)
            print(d)

##  ##

## 进程池 ##

用multiprocess中的Pool起进程池  
进程池中开启的个数：默认是cpu个数  
提交任务（不能传队列作为子进程的参数，只能传管道）  
\- apply 同步提交，直接返回结果  
\- apply\_async 异步提交，返回对象，通过对象获取返回值

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from multiprocessing import Pool
    import time
    
    def Foo(i):
        time.sleep(2)
        return i + 100
    
    def Back(arg):
        print('--->:', arg)
    
    if __name__ == '__main__':
        pool = Pool(5)
    
        for i in range(10):
            pool.apply_async(func=Foo, args=(i,), callback=Back)
            # pool.apply(func=Foo, args=(i,))
    
        print('end')
        pool.close()
        pool.join()

转载于:https://www.cnblogs.com/py-peng/p/11068296.html

[ContractedBlock.gif]: /images/20211201/bc0d3891711540a7aa62cc214411d60a.png
[ExpandedBlockStart.gif]: /images/20211201/85a10ea19d1f48c3af49409065344f7d.png
[Pipe]: https://docs.python.org/3.5/library/multiprocessing.html#multiprocessing.Pipe
[send]: https://docs.python.org/3.5/library/multiprocessing.html#multiprocessing.Connection.send
[recv]: https://docs.python.org/3.5/library/multiprocessing.html#multiprocessing.Connection.recv
[list]: https://docs.python.org/3.5/library/stdtypes.html#list
[dict]: https://docs.python.org/3.5/library/stdtypes.html#dict
[Namespace]: https://docs.python.org/3.5/library/multiprocessing.html#multiprocessing.managers.Namespace
[Lock]: https://docs.python.org/3.5/library/multiprocessing.html#multiprocessing.Lock
[RLock]: https://docs.python.org/3.5/library/multiprocessing.html#multiprocessing.RLock
[Semaphore]: https://docs.python.org/3.5/library/multiprocessing.html#multiprocessing.Semaphore
[BoundedSemaphore]: https://docs.python.org/3.5/library/multiprocessing.html#multiprocessing.BoundedSemaphore
[Condition]: https://docs.python.org/3.5/library/multiprocessing.html#multiprocessing.Condition
[Event]: https://docs.python.org/3.5/library/multiprocessing.html#multiprocessing.Event
[Barrier]: https://docs.python.org/3.5/library/multiprocessing.html#multiprocessing.Barrier
[Queue]: https://docs.python.org/3.5/library/multiprocessing.html#multiprocessing.Queue
[Value]: https://docs.python.org/3.5/library/multiprocessing.html#multiprocessing.Value
[Array]: https://docs.python.org/3.5/library/multiprocessing.html#multiprocessing.Array