Python升级之路( Lv15 ) 并发编程三剑客: 进程, 线程与协程

import time
from multiprocessing import Process


class MyProcess(Process):
    """进程的创建方式: 2.类包装"""
    def __init__(self, name):
        Process.__init__(self)
        self.name = name

    def run(self):
        print(f"进程{self.name} 启动")
        time.sleep(3)
        print(f"进程{self.name} 结束")


if __name__ == "__main__":
    print("创建进程")
    p1 = MyProcess("p1")
    p2 = MyProcess("p2")
    p1.start()
    p2.start()

import os
import time
from multiprocessing import Process


def function(name):
    """进程的创建方式: 1.方法包装"""
    print("当前进程ID:", os.getpid())
    print("父进程ID", os.getppid())
    print(f"Process:{name} start")
    time.sleep(3)
    print(f"Process:{name} end")


if __name__ == "__main__":
    print("当前main进程ID: ", os.getppid())
    # 创建进程
    p1 = Process(target=function, args=("p1",))
    p2 = Process(target=function, args=("p2",))
    p1.start()
    p2.start()

from multiprocessing import Process, Queue


class MyProcess(Process):
    def __init__(self, name, mq):
        Process.__init__(self)
        self.name = name
        self.mq = mq

    def run(self):
        print("Process {} started".format(self.name))
        print("===Queue", self.mq.get(), "===")
        self.mq.put(self.name)
        print("Process {} end".format(self.name))


if __name__ == "__main__":
    # 创建进程列表
    t_list = []
    mq = Queue()
    mq.put("1")
    mq.put("2")
    mq.put("3")
    # 利用range序列重复创建进程
    for i in range(3):
        t = MyProcess("p{}".format(i), mq)
        t.start()
        t_list.append(t)
    # 等待进程结束
    for t in t_list:
        t.join()
    print(mq.get())
    print(mq.get())
    print(mq.get())

import multiprocessing
import time


def fun1(conn1):
    """
    管道结构
    进程<==>conn1(管道头)==pipe==conn2(管道尾)<==>进程2
    """
    sub_info = "进程向conn1发送消息, 管道另一头conn2 可以接收到消息"
    print(f"进程1--{multiprocessing.current_process().pid}发送数据：{sub_info}")
    time.sleep(1)
    conn1.send(sub_info)        # 调用conn1.send发送消息, 发送的消息会被管道的另一头接收
    print(f"conn1接收消息:{conn1.recv()}")    # conn1.recv接收消息, 如果没有消息可接收, recv方法会一直阻塞. 如果管道已经被关闭，那么recv方法会抛出EOFError
    time.sleep(1)


def fun2(conn2):
    sub_info = "进程向conn2发送消息, 管道另一头conn1 可以接收到消息"
    print(f"进程2--{multiprocessing.current_process().pid}发送数据：{sub_info}")
    time.sleep(1)
    conn2.send(sub_info)
    print(f"conn2接收消息:{conn2.recv()}")
    time.sleep(1)


if __name__ == "__main__":
    # 创建管道
    # Pipe方法返回（conn1， conn2）代表一个管道的两个端.如果conn1带表头, conn2代表尾, conn1发送的消息只会被conn2接收, 同理conn2发送的消息也只会被conn1接收
    conn1, conn2 = multiprocessing.Pipe()
    # 创建子进程
    # Python中，圆括号意味着调用函数. 在没有圆括号的情况下，Python会把函数当做普通对象
    process1 = multiprocessing.Process(target=fun1, args=(conn1,))
    process2 = multiprocessing.Process(target=fun2, args=(conn2,))
    # 启动子进程
    process1.start()
    process2.start()

from multiprocessing import Manager, Process


def func1(name,m_list,m_dict):
     m_dict['area'] = '罗布泊'
     m_list.append('钱三强')


def func2(name, m_list, m_dict):
    m_dict['work'] = '造核弹'
    m_list.append('邓稼先')


if __name__ == "__main__":
    with Manager() as mgr:
        m_list = mgr.list()
        m_dict = mgr.dict()
        m_list.append("钱学森")
        # 两个进程不能直接互相使用对象，需要互相传递
        p1 = Process(target=func1, args=('p1', m_list, m_dict))
        p1.start()
        p1.join()   # 等p1进程结束，主进程继续执行
        print(m_list)
        print(m_dict)
        p2 = Process(target=func2, args=('p1', m_list, m_dict))
        p2.start()
        p2.join()   # 等p2进程结束，主进程继续执行
        print(m_list)
        print(m_dict)

from multiprocessing import Pool
import os
from time import sleep


def func1(name):
    print(f"方法1输出: 当前进程的ID:{os.getpid()},{name}")
    sleep(2)
    return name


def func2(args):
    print("方法2输出: ", args)


if __name__ == "__main__":
    pool = Pool(5)
    pool.apply_async(func=func1, args=('进程1',), callback=func2)
    pool.apply_async(func=func1, args=('进程2',), callback=func2)
    pool.apply_async(func=func1, args=('进程3',), callback=func2)
    pool.apply_async(func=func1, args=('进程4',))
    pool.apply_async(func=func1, args=('进程5',))
    pool.apply_async(func=func1, args=('进程6',))
    pool.apply_async(func=func1, args=('进程7',))
    pool.close()
    pool.join()

from multiprocessing import Pool
import os
from time import sleep


def func1(name):
    print(f"方法1输出: 当前进程的ID:{os.getpid()},{name}")
    sleep(2)
    return name


if __name__ == "__main__":
    with Pool(5) as pool:
        args = pool.map(func1, ('进程1,', '进程2,', '进程3,', '进程4,', '进程5,', '进程6,', '进程7,', '进程8,'))
    for a in args:
        print(a)

from threading import Thread
from time import sleep


def func1(name):
    """线程创建方式(方法包装)"""
    for i in range(3):
        print(f"thread:{name} : {i}")
        sleep(1)


if __name__ == "__main__":
    """新建两个线程, 每个线程每隔1秒打印一次消息, 总共打印3次"""
    print("主线程, start")
    # 创建线程
    t1 = Thread(target=func1, args=("t1",))
    t2 = Thread(target=func1, args=("t2",))
    # 启动线程
    t1.start()
    t2.start()
    print("主线程, end")
"""
注意: 运行结果可能会出现换行问题，是因为多个线程抢夺控制台输出的IO流
"""

from threading import Thread
from time import sleep


class MyThread(Thread):
    """
    类的方式创建线程(类包装)
    """

    def __init__(self, name):
        Thread.__init__(self)
        self.name = name

    def run(self):
        for i in range(3):
            print(f"thread : {self.name} : {i}")
            sleep(1)


if __name__ == "__main__":
    """新建两个线程, 每个线程每隔1秒打印一次消息, 总共打印3次"""
    print("主线程, start")
    # 创建线程(类创建的方式)
    t1 = MyThread("t1")
    t2 = MyThread("t2")
    # 启动线程
    t1.start()
    t2.start()
    print("主线程, end")

if __name__ == "__main__":
    """新建两个线程, 每个线程每隔1秒打印一次消息, 总共打印3次"""
    print("主线程, start")
    # 创建线程(类创建的方式)
    t1 = MyThread("t1")
    t2 = MyThread("t2")
    # 启动线程
    t1.start()
    t2.start()
    # 主线程等t1, t2结束之后, 再往下执行
    t1.join()
    t2.join()
    print("主线程, end")

from threading import Thread
from time import sleep


class MyThread(Thread):
    """守护线程"""
    def __init__(self, name):
        Thread.__init__(self)
        self.name = name

    def run(self):
        for i in range(3):
            print(f"thread : {self.name} : {i}")
            sleep(1)


if __name__ == "__main__":
    print("主线程, start")
    # 创建线程(类创建的方式)
    t1 = MyThread("t1")
    # 设置为守护线程
    # t1.setDaemon(True)  # 该方法3.10后被废弃, 可以直接按照下面的方法设置
    t1.daemon = True
    # 启动线程
    t1.start()
    print("主线程, end")

import time
import threading


def chihuoguo(name):
    """利用事件实现就餐, 吃饭"""
    # 等待时间, 进入阻塞等待状态
    print(f'{name} 已经到餐厅')
    print(f'小伙伴{name} 已经进入就餐状态！')
    event.wait()
    print(f'{name} 收到通知了.')
    print(f'小伙伴{name} 开始吃咯！')


if __name__ == "__main__":
    event = threading.Event()
    # 创建新线程
    threa1 = threading.Thread(target=chihuoguo, args=("玛奇玛",))
    threa2 = threading.Thread(target=chihuoguo, args=("电次",))
    # 开启线程
    threa1.start()
    threa2.start()
    time.sleep(10)
    # 发送时间通知
    print('---->>>主线程(波吉塔)通知小伙伴开吃咯！')
    event.set()

from threading import Thread
from time import sleep


class Account:
    def __init__(self, money, name):
        self.money = money
        self.name = name


# 模拟提款操作
class Drawing(Thread):
    def __init__(self, drawingNum, account):
        Thread.__init__(self)
        self.drawingNum = drawingNum
        self.account = account
        self.expenseTotal = 0

    def run(self):
        if self.account.money - self.drawingNum < 0:
            return
        sleep(1)  # 判断完后阻塞。其他线程开始运行
        self.account.money -= self.drawingNum
        self.expenseTotal += self.drawingNum
        print(f"账户：{self.account.name},余额是：{self.account.money}")
        print(f"账户：{self.account.name},总共取了：{self.expenseTotal}")


if __name__ == "__main__":
    a1 = Account(100, "timepause")
    # 定义取钱线程对象
    draw1 = Drawing(80, a1)
    # 定义取钱线程对象
    draw2 = Drawing(80, a1)
    draw1.start()
    draw2.start()

from threading import Thread, Lock
from time import sleep


class Account:
    def __init__(self, money, name):
        self.money = money
        self.name = name


# 模拟提款操作
class Drawing(Thread):
    def __init__(self, drawingNum, account):
        Thread.__init__(self)
        self.drawingNum = drawingNum
        self.account = account
        self.expenseTotal = 0

    def run(self):
        # 获得锁
        lock1.acquire()
        if self.account.money - self.drawingNum < 0:
            return
        sleep(1)  # 判断完后阻塞。其他线程开始运行
        self.account.money -= self.drawingNum
        self.expenseTotal += self.drawingNum
        # 释放锁
        lock1.release()
        print(f"账户：{self.account.name},余额是：{self.account.money}")
        print(f"账户：{self.account.name},总共取了：{self.expenseTotal}")


if __name__ == "__main__":
    a1 = Account(100, "timepause")
    # 创建互斥锁
    lock1 = Lock()
    # 定义取钱线程对象
    draw1 = Drawing(80, a1)
    # 定义取钱线程对象
    draw2 = Drawing(80, a1)
    draw1.start()
    draw2.start()

from threading import Lock, Semaphore, Thread
from time import sleep


def home(name, se):
    """
    信号量控制同时访问资源的数量。信号量和锁相似，锁同一时间只允许一个对象(进程)通过，信号量同一时间允许多个对象(进程)通过
    """
    # 拿到信号量
    se.acquire()
    print(f"{name}进入了厕所")
    sleep(3)
    print(f'******************{name}走出了厕所')
    # 归还信号量
    se.release()


if __name__ == "__main__":
    """
    一个房间一次只允许两个人使用: 若不使用信号量，会造成所有人都进入这个房子; 若只允许一人通过可以用锁Lock()
    """
    # 创建信号量的对象，有两个坑位
    se = Semaphore(2)
    print("欢迎光临本厕所, 目前只有两个坑位(联系商务即可扩增~~~)")
    for i in range(7):
        p = Thread(target=home, args=(f'tom{i}', se))  # 前面加f的作用是令{}里面的变量引用可以生效
        p.start()

from threading import Lock, Thread
from time import sleep


def run1():
    # 获取锁
    lock1.acquire()
    print("幸平创真拿到菜刀")
    # lock1.release()
    # print("幸平创真释放菜刀")
    lock2.acquire()
    print("幸平创真拿到锅")
    # lock2.release()
    # print("幸平创真释放锅")



def run2():
    lock2.acquire()
    print("卫宫士郎拿到锅")
    # lock2.release()
    # print("卫宫士郎释放锅")
    lock1.acquire()
    print("卫宫士郎拿到菜刀")
    # lock1.release()
    # print("卫宫士郎释放菜刀")



if __name__ == "__main__":
    """
    死锁是由于“同步块需要同时持有多个锁造成”的，要解决这个问题思路很简单，就是：同一个代码块，不要同时持有两个对象锁
    """
    lock1 = Lock()
    lock2 = Lock()

    t1 = Thread(target=run1)
    t2 = Thread(target=run2)

    t1.start()
    t2.start()

import queue
from threading import Thread
from time import sleep


def producer():
    num = 1
    while True:
        if queue.qsize() < 5:
            print(f"生产:{num}号消息")
            queue.put(f"消费: {num}号消息")
            num += 1
        else:
            print("消息队列已满, 等待被消费")
        sleep(1)


def consumer():
    while True:
        print(f"获取消息:{queue.get()}")
        sleep(1)


if __name__ == "__main__":
    queue = queue.Queue()
    # 注意这里 target=函数名, 而不是 target=函数名()
    t = Thread(target=producer)
    t.start()
    c = Thread(target=consumer)
    c.start()
    c2 = Thread(target=consumer)
    c2.start()

# 不使用协程执行多个任务
import time


def fun1():
    for i in range(3):
        print(f'原子弹：第{i}次爆炸啦')
        time.sleep(1)
    return "fun1执行完毕"


def fun2():
    for k in range(3):
        print(f'氢弹：第{k}次爆炸了')
        time.sleep(1)
    return "fun2执行完毕"


def main():
    fun1()
    fun2()


if __name__ == "__main__":
    start_time = time.time()
    main()
    end_time = time.time()
    print(f"耗时{end_time - start_time}")  # 不使用协程，耗时6秒

# 不使用协程执行多个任务
import time


def fun1():
    for i in range(3):
        print(f'原子弹：第{i}次爆炸啦')
        yield   # 只要方法包含了yield，就变成一个生成器
        time.sleep(1)
    return "fun1执行完毕"


def fun2():
    g = fun1()   # fun1是一个生成器，fun1()就不会直接调用，需要通过next()或for循环调用
    print(type(g))
    for k in range(3):
        print(f'氢弹：第{k}次爆炸了')
        next(g)   # 继续执行fun1的代码
        time.sleep(1)
    return "fun2执行完毕"


def main():
    fun1()
    fun2()


if __name__ == "__main__":
    start_time = time.time()
    main()
    end_time = time.time()
    print(f"耗时{end_time - start_time}")  # 耗时5.0秒，效率差别不大

# 不使用协程执行多个任务
import asyncio
import time


async def fun1():  # async表示方法是异步的
    for i in range(3):
        print(f'原子弹：第{i}次爆炸啦')
        # await异步执行func1方法
        await asyncio.sleep(1)
    return "fun1执行完毕"


async def fun2():
    for k in range(3):
        print(f'氢弹：第{k}次爆炸了')
        # await异步执行func2方法
        await asyncio.sleep(1)
    return "fun2执行完毕"


async def main():
    res = await asyncio.gather(fun1(), fun2())
    # 返回值为函数的返回值列表,本例为["func1执行完毕", "func2执行完毕"]
    print(res)

if __name__ == "__main__":
    start_time = time.time()
    asyncio.run(main())
    end_time = time.time()
    print(f"耗时{end_time - start_time}")  # 耗时3秒,效率极大提高