python基础之并发编程(一)

2024-04-02 19:04:59 666人浏览独家记忆

Python 官方文档：入门教程 => 点击学习

摘要

目录一、进程（Process）二、线程（Thread）三、并发编程解决方案：四、多线程实现（两种）1、第一种函数方法2、第二种类方法包装五、守护线程与子线程1、线程在分法有：2

一、进程（Process）

是一个具有一定独立功能的程序关于某个数据集合的一次运行活动

二、线程（Thread）

是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位。

三、并发 编程解决方案：

1、多任务的实现有 3 种方式：

多进程模式；
多线程模式；
多进程+多线程模式

四、多线程实现（两种）

1、第一种函数方法


# 方法包装-启动多线程
from threading import Thread 
from time import sleep, time 
def func1(name): 
    print("Threading:{} start".fORMat(name)) 
    sleep(3) 
    print("Threading:{} end".format(name)) 
if __name__ == '__main__': 
    # 开始时间 
    start = time() 
    # 创建线程列表 
    t_list = [] 
    # 循环创建线程 
    for i in range(10): 
        t = Thread(target=func1, args=('t{}'.format(i),)) 
        t.start() 
        t_list.append(t) 
    # 等待线程结束 
    for t in t_list: 
        t.join() 
    # 计算使用时间 
    end = time() - start 
    print(end)

2、第二种类方法包装


# 类包装-启动多线程 
from threading import Thread 
from time import sleep, time 
class MyThread(Thread): 
    def __init__(self,name): 
        Thread.__init__(self) 
        self.name =name 
    def run(self): 
        print("Threading:{} start".format(self.name)) 
        sleep(3) 
        print("Threading:{} end".format(self.name)) 
if __name__ == '__main__': 
    # 开始时间 
    start = time() 
    # 创建线程列表 
    t_list = [] 
    # 循环创建线程 
    for i in range(10): 
        t = MyThread('t{}'.format(i)) 
        t.start() 
        t_list.append(t) 
    # 等待线程结束 
    for t in t_list: 
        t.join() 
    # 计算使用时间 
    end = time() - start 
    print(end)

注意：

主线程不会等待子线程运行结束，如果需要等待可使用 join()方法不要启动线程后立即 join()，很容易造成串行运行，导致并发失效

五、守护线程与子线程

1、线程在分法有：

主线程：程序的本身

子线程：在程序另开起的线程

2、守护线程

主要的特征是它的生命周期。主线程死亡，它也就随之死亡


# 类包装-启动多线程 
from threading import Thread 
from time import sleep, time 
class MyThread(Thread): 
    def __init__(self,name): 
        Thread.__init__(self) 
        self.name =name 
    def run(self): 
        print("Threading:{} start".format(self.name)) 
        sleep(3) 
        print("Threading:{} end".format(self.name)) 
if __name__ == '__main__': 
    # 开始时间 
    start = time() 
    # 循环创建线程 
    for i in range(10): 
        t = MyThread('t{}'.format(i)) 
        t.setDaemon(True)
        t.start() 
    # 计算使用时间 
    end = time() - start 
    print(end)

六、锁


from threading import Thread 
def func1(name): 
    print('Threading:{} start'.format(name)) 
    global num 
    for i in range(50000000): # 有问题 
    #for i in range(5000): # 无问题 
        num += 1 
    print('Threading:{} end num={}'.format(name, num))
if __name__ == '__main__': 
    num =0 
    # 创建线程列表 
    t_list = [] 
    # 循环创建线程 
    for i in range(5): 
        t = Thread(target=func1, args=('t{}'.format(i),)) 
        t.start() 
        t_list.append(t) 
    # 等待线程结束 
    for t in t_list: 
        t.join()

python 使用线程的时候，会定时释放 GIL 锁，这时会 sleep,所以才会出现上面的问题。面对这个问题，如果要解决此问题，我们可以使用 Lock 锁解决此问题（加锁的目的是：保证数据安全）


from threading import Thread,Lock 
def func1(name):
    # 获取锁
    lock.acquire()
    with lock:
        global count
        for i in range(100000):
            count += 1
    # 释放锁 
    lock.release()
if __name__ == "__main__":
    count = 0
    t_list = []
    # 创建锁对象
    lock = Lock()
    for i in range(10):
        t = Thread(target=func1,args=(f't{i+1}',))
        t.start()
        t_list.append(t)
    for t in t_list:
        t.join()
    print(count)

七、死锁


from threading import Thread, Lock #Lock 锁 同步锁 互斥锁
from time import sleep 
def fun1(): 
    lock1.acquire() 
    print('fun1 拿到键盘') 
    sleep(2) 
    lock2.acquire() 
    print('fun1 拿到鼠标') 
    lock2.release() 
    print('fun1 释放鼠标') 
    lock1.release() 
    print('fun1 释放键盘') 
def fun2(): 
    lock2.acquire() 
    print('fun2 拿到鼠标') 
    lock1.acquire() 
    print('fun2 拿到键盘') 
    lock1.release() 
    print('fun2 释放键盘') 
    lock2.release() 
    print('fun2 释放鼠标') 
if __name__ == '__main__':
    lock1 = Lock() 
    lock2 = Lock() 
    t1 = Thread(target=fun1) 
    t2 = Thread(target=fun2) 
    t1.start() 
    t2.start()


from threading import RLock
'''
Lock 锁 同步锁 互斥锁
RLock 递归锁
'''
def func1():
    lock.acquire()
    print('func1获取锁')
    func2()
    lock.release()
    print('func1释放锁')
def func2():
    lock.acquire()
    print('func2获取锁')
    lock.release()
    print('func2释放锁')
def func3():
    func1()
    func2()
if __name__ == "__main__":
    #lock = Lock()  会产生错误 
    lock = RLock()
    func3()

八、信号量（Semaphore）

我们都知道在加锁的情况下，程序就变成了串行，也就是单线程，而有时，我们在不用考虑数据安全时，为了避免业务开启过多的线程时。我们就可以通过信号量（Semaphore）来设置指定个数的线程。（比如：电梯每次只能承载三个人，那么同时只能有三个人乘坐，其他人只能等别人做完才能乘坐）


from time import sleep
from threading import Thread
from threading import BoundedSemaphore
def index(num):
    lock.acquire()
    print(f'第{num}个人乘坐！！')
    sleep(2)
    lock.release()
if __name__ == "__main__":
    lock = BoundedSemaphore(3)
    for i in range(10):
        t = Thread(target=index,args=(f'{i+1}',))
        t.start()

九、事件(Event)

Event()可以创建一个事件管理标志，该标志（event）默认为 False，event 对象主要有四种方法可以调用：

1、 event.wait(timeout=None)：调用该方法的线程会被阻塞，如果设置了 timeout 参数，超时后，线程会停止阻塞继续执行；

2、event.set()：将 event 的标志设置为 True，调用 wait 方法的所有线程将被唤醒；

3、event.clear()：将 event 的标志设置为 False，调用 wait 方法的所有线程将被阻塞；

4、event.is_set()：判断 event 的标志是否为 True。

十、线程通信-队列

线程安全是多线程编程时的计算机程序代码中的一个概念。在拥有共享数据的多条线程并行执行的程序中，线程安全的代码会通过同步机制保证各个线程都可以正常且正确的执行，不会出现数据污染等意外情况

1使用的队列的好处：

1. 安全

2. 解耦

3. 提高效率

2Queue模块中的常用方法:

Python的Queue模块中提供了同步的、线程安全的队列类，包括FIFO（先入先出)队列Queue，LIFO（后入先出）队列LifoQueue，和优先级队列PriorityQueue。这些队列都实现了锁原语，能够在多线程中直接使用。可以使用队列来实现线程间的同步

Queue.qsize() 返回队列的大小
Queue.empty() 如果队列为空，返回True,反之False
Queue.full() 如果队列满了，返回True,反之False
Queue.full 与 maxsize 大小对应
Queue.get([block[, timeout]])获取队列，timeout等待时间
Queue.get_nowait() 相当Queue.get(False)
Queue.put(item) 写入队列，timeout等待时间
Queue.put_nowait(item) 相当Queue.put(item, False)
Queue.task_done() 在完成一项工作之后，Queue.task_done()函数向任务已经完成的队列发送一个信号
Queue.join() 实际上意味着等到队列为空，再执行别的操作

十一、生产者和消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。


from threading import Thread
from queue import Queue
from time import sleep
def producer():
    num = 1
    while True:
        print(f'生产了{num}号皮卡丘')
        qe.put(f'{num}号皮卡丘')
        num += 1 
        sleep(1)
def consumer():
    print('购买了{}'.format(qe.get()))
    sleep(2)
if __name__ == "__main__":
    # 共享数据的容器
    qe= Queue(maxsize=5)
    # 创建生产者线程
    t1 = Thread(target = producer)
    # 创建消费者线程
    t2 = Thread(target = consumer)
    # 创建消费者线程
    t3 = Thread(target = consumer)
    # 开始工作
    t1.start()
    t2.start()
    t3.start()