Python学习笔记十三（多任务、进程、进程池）

2018-04-29 本文已影响166人 DragonFangQy

多任务之进程

什么是进程

进程是CPU资源分配的基本单位，进程包含线程，线程依赖于进程。

进程的状态

进程的状态

就绪态：运行的条件都已经慢去，正在等在cpu执行
执行态：cpu正在执行其功能
等待态：等待某些条件满足，例如一个程序sleep了，此时就处于等待态

进程可以理解为一个团队，而线程就是团队的成员。当团队要同时完成多个项目，缺少人手时需要招人，既开辟子线程，当然也可以使用另外的方式完成多任务，比如再找一个团队，既多线程完成多任务。

多进程多任务

Python 中可以使用multiprocessing 开启多进程


import time

def look():
    """看风景"""
    for i in range(10):
        print("这里风景真好...")
        time.sleep(0.1)


def walk():
    """徒步旅行"""
    for i in range(10):
        print("我在徒步旅行...")
        time.sleep(0.2)


def main():
    """入口函数"""
    look()
    walk()


if __name__ == '__main__':
    main()

# 运行结果:
# 这里风景真好...
# 这里风景真好...
# 这里风景真好...
# 这里风景真好...
# 这里风景真好...
# 这里风景真好...
# 这里风景真好...
# 这里风景真好...
# 这里风景真好...
# 这里风景真好...
# 我在徒步旅行...
# 我在徒步旅行...
# 我在徒步旅行...
# 我在徒步旅行...
# 我在徒步旅行...
# 我在徒步旅行...
# 我在徒步旅行...
# 我在徒步旅行...
# 我在徒步旅行...
# 我在徒步旅行...

将上面的单进程单任务改为多进程多任务


import multiprocessing
import time


def look():
    # 获取当前进程
    curr_process = multiprocessing.current_process()
    print("look_process", curr_process)

    # 打印当前进程的id
    print("look_pid", curr_process.pid)

    """看风景"""
    for i in range(10):
        print("这里风景真好...")
        time.sleep(0.1)


def walk():
    # 获取当前进程
    curr_process = multiprocessing.current_process()
    print("walk_process", curr_process)

    # 打印当前进程的id
    print("walk_pid", curr_process.pid)

    """徒步旅行"""
    for i in range(10):
        print("我在徒步旅行...")
        time.sleep(0.2)


def main():
    """入口函数"""
    # look()
    # walk()

    # 获取当前进程
    curr_process = multiprocessing.current_process()
    print("current_process", curr_process)

    # 打印当前进程的id
    print("process_pid", curr_process.pid)

    # 获取当前进程中活动的子进程
    print("multiprocessing.Process前 ", multiprocessing.active_children())

    look_process = multiprocessing.Process(target=look)
    walk_process = multiprocessing.Process(target=walk)

    # 获取当前进程中活动的子进程
    print("multiprocessing.Process后 ", multiprocessing.active_children())

    look_process.start()
    walk_process.start()

    # 获取当前进程中活动的子进程
    print("start后 ", multiprocessing.active_children())


if __name__ == '__main__':
    main()

    # 运行结果:
    # current_process < _MainProcess(MainProcess, started) >
    # process_pid 14734
    # multiprocessing.Process前[]
    # multiprocessing.Process后[]
    # start后[ < Process(Process - 2, started) >, < Process(Process - 1, started) >]
    # look_process < Process(Process - 1, started) >
    # look_pid 14735
    # 这里风景真好...
    # walk_process < Process(Process - 2, started) >
    # walk_pid 14736
    # 我在徒步旅行...
    # 这里风景真好...
    # 这里风景真好...
    # 我在徒步旅行...
    # 这里风景真好...
    # 这里风景真好...
    # 我在徒步旅行...
    # 这里风景真好...
    # 我在徒步旅行...
    # 这里风景真好...
    # 这里风景真好...
    # 我在徒步旅行...
    # 这里风景真好...
    # 这里风景真好...
    # 我在徒步旅行...
    # 我在徒步旅行...
    # 我在徒步旅行...
    # 我在徒步旅行...
    # 我在徒步旅行...

分析结果：

默认有一条进程MainProcess ，叫做主进程
multiprocessing.Process() 只是准备了一条进程，并没有创建子进程（multiprocessing.Process前[]）
start 将准备好的子线程运行起来（multiprocessing.Process前[]）
进程之间执行是无序的(输出结果无序)
主进程默认会等待子进程结束

设置守护进程


import multiprocessing
import time


def look():
    """看风景"""
    for i in range(10):
        print("这里风景真好...")
        time.sleep(0.1)


def walk():
    """徒步旅行"""
    for i in range(10):
        print("我在徒步旅行...")
        time.sleep(0.2)


def main():
    """入口函数"""
    # look()
    # walk()

    look_process = multiprocessing.Process(target=look)
    walk_process = multiprocessing.Process(target=walk)

    # 设置守护进程
    look_process.daemon = True
    walk_process.daemon = True

    look_process.start()
    walk_process.start()

    time.sleep(1)
    print("over")


if __name__ == '__main__':
    main()
# 运行结果:
# 我在徒步旅行...
# 这里风景真好...
# 这里风景真好...
# 我在徒步旅行...
# 这里风景真好...
# 这里风景真好...
# 我在徒步旅行...
# 这里风景真好...
# 这里风景真好...
# 我在徒步旅行...
# 这里风景真好...
# 这里风景真好...
# 我在徒步旅行...
# 这里风景真好...
# 这里风景真好...
# over

从结果上看，主进程结束，子进程被结束或者说被销毁。

小结：

获取当前进程
multiprocessing.current_process()
打印当前进程的id
- print("process_pid", curr_process.pid)
获取活动子进程
multiprocessing.active_children()
进程之间执行是无序的
默认情况下，主进程会等待子进程执行结束，程序再退出，既主进程结束。
设置守护进程，主进程结束，子进程被销毁。
注意点：主进程里的子进程如果有一个没有设置守护，那么主进程会等待这个子进程执行结束，因为主进程没有结束，所以守护进程不会被销毁。
注意点：设置守护进程( daemon )，要在start() 之前设置，否则会报错（RuntimeError: cannot set daemon status of active thread）。

讨论：为什么进程执行无序？首先，线程是CPU调度的基本单位，线程执行是无序的；其次，线程依赖于进程，一个进程默认有一条线程，所以进程之间的执行顺序也是无序的。

记录：线程和进程对CPU的使用情况，线程对CPU的使用率达不到100%，仅有50%左右；进程对CPU的使用率可以达到100%。

多进程与全局变量^[1]


import multiprocessing
import time

tmp_list = []


def look():
    """看风景"""
    for i in range(10):
        tmp_list.append("这里风景真好...")
        print("这里风景真好...")
        time.sleep(0.1)

    print(id(tmp_list)) 
    print(tmp_list)

def walk():
    """徒步旅行"""
    for i in range(10):
        tmp_list.append("我在徒步旅行...")

        print("我在徒步旅行...")
        time.sleep(0.2)

    print(id(tmp_list))
    print(tmp_list)


def main():
    """入口函数"""
    # look()
    # walk()

    look_process = multiprocessing.Process(target=look)
    walk_process = multiprocessing.Process(target=walk)

    # 设置守护进程
    look_process.daemon = True
    walk_process.daemon = True

    look_process.start()
    walk_process.start()

    time.sleep(3)
    print("over", tmp_list)
    print(id(tmp_list))


if __name__ == '__main__':
    main()

    # 运行结果:
    # 这里风景真好...
    # 我在徒步旅行...
    # 这里风景真好...
    # 这里风景真好...
    # 我在徒步旅行...
    # 这里风景真好...
    # 这里风景真好...
    # 我在徒步旅行...
    # 这里风景真好...
    # 这里风景真好...
    # 我在徒步旅行...
    # 这里风景真好...
    # 我在徒步旅行...
    # 这里风景真好...
    # 这里风景真好...
    # 139881809190792
    # ['这里风景真好...', '这里风景真好...', '这里风景真好...', '这里风景真好...', '这里风景真好...', '这里风景真好...', '这里风景真好...', '这里风景真好...', '这里风景真好...', '这里风景真好...']
    # 我在徒步旅行...
    # 我在徒步旅行...
    # 我在徒步旅行...
    # 我在徒步旅行...
    # 我在徒步旅行...
    # 139881809190792
    # ['我在徒步旅行...', '我在徒步旅行...', '我在徒步旅行...', '我在徒步旅行...', '我在徒步旅行...', '我在徒步旅行...', '我在徒步旅行...', '我在徒步旅行...', '我在徒步旅行...', '我在徒步旅行...']
    # over[]
    # 139881809190792

从结果可以看出，进程之间不共享全局变量

使用Queue，完成进程之间数据共享


import multiprocessing
import time

tmp_list = []


def look(qun):
    """看风景"""
    for i in range(10):
        tmp = qun.get()

        tmp.append("这里风景真好...")
        qun.put(tmp)
        print("这里风景真好...")
        time.sleep(0.1)


def walk(qun):
    """徒步旅行"""
    for i in range(10):
        tmp = qun.get()
        tmp.append("我在徒步旅行...")

        print("我在徒步旅行...")

        qun.put(tmp)
        time.sleep(0.2)


def main():
    """入口函数"""
    # look()
    # walk()

    qun = multiprocessing.Queue(3)
    qun.put(tmp_list)

    look_process = multiprocessing.Process(target=look, args=(qun,))
    walk_process = multiprocessing.Process(target=walk, args=(qun,))

    # 设置守护进程
    look_process.daemon = True
    walk_process.daemon = True

    look_process.start()
    walk_process.start()

    time.sleep(3)
    print("over", qun.get())


if __name__ == '__main__':
    main()

    # 运行结果:
    # 这里风景真好...
    # 我在徒步旅行...
    # 这里风景真好...
    # 这里风景真好...
    # 我在徒步旅行...
    # 这里风景真好...
    # 这里风景真好...
    # 我在徒步旅行...
    # 这里风景真好...
    # 这里风景真好...
    # 我在徒步旅行...
    # 这里风景真好...
    # 这里风景真好...
    # 我在徒步旅行...
    # 这里风景真好...
    # 我在徒步旅行...
    # 我在徒步旅行...
    # 我在徒步旅行...
    # 我在徒步旅行...
    # 我在徒步旅行...
    # over[
    #     '这里风景真好...', '我在徒步旅行...', '这里风景真好...', '这里风景真好...', '我在徒步旅行...', '这里风景真好...', '这里风景真好...', '我在徒步旅行...', '这里风景真好...', '这里风景真好...', '我在徒步旅行...', '这里风景真好...', '这里风景真好...', '我在徒步旅行...', '这里风景真好...', '我在徒步旅行...', '我在徒步旅行...', '我在徒步旅行...', '我在徒步旅行...', '我在徒步旅行...']

通过Queue 实现了数据共享。

进程之间的资源竞争


import multiprocessing
import time

a = 0


def look(qun):
    """看风景"""
    for i in range(1000000):
        qun.value += 1

def walk(qun):
    """徒步旅行"""
    for i in range(1000000):
        qun.value += 1


def main():
    """入口函数"""
    # look()
    # walk()

    qun = multiprocessing.Value("d", a)

    look_process = multiprocessing.Process(target=look, args=(qun,))
    walk_process = multiprocessing.Process(target=walk, args=(qun,))

    # 设置守护进程
    look_process.daemon = True
    walk_process.daemon = True

    look_process.start()
    walk_process.start()

    time.sleep(10)
    print("over", qun.value)


if __name__ == '__main__':
    main()

    # 运行结果:
    # over 1052801.0

同步解决资源竞争问题


import multiprocessing
import time

a = 0


def look(qun):
    """看风景"""
    for i in range(1000000):
        qun.value += 1

def walk(qun):
    """徒步旅行"""
    for i in range(1000000):
        qun.value += 1


def main():
    """入口函数"""
    # look()
    # walk()

    qun = multiprocessing.Value("d", a)

    look_process = multiprocessing.Process(target=look, args=(qun,))
    walk_process = multiprocessing.Process(target=walk, args=(qun,))

    # 设置守护进程
    look_process.daemon = True
    walk_process.daemon = True

    look_process.start()
    
    # 进程同步
    look_process.join()
    
    walk_process.start()

    time.sleep(10)
    print("over", qun.value)


if __name__ == '__main__':
    main()

    # 运行结果:
    # over 2000000.0

类比线程资源竞争问题，进程同样可以使用同步与互斥锁解决资源竞争问题。

小结

多进程共享全局变量，但共享全局变量会产生资源竞争问题
进程同步（join()）和互斥锁都能解决资源竞争问题，但是都是将多任务变成了单任务。
使用锁要避免死锁
注意点：如果两个子进程共享全局变量，只对一个子进程加锁，可以理解为和没加锁一样。
注意点：效率问题，同样是累加200,0000次，直接遍历累加200,0000次所用时间小于进程同步所用时间小于互斥锁所用时间

进程池


import multiprocessing
import time


# 向指定队列写入数据
def write_data(queue):
    for i in range(5):
        queue.put(i)
        print(i)
        time.sleep(0.2)

    print("数据写入完成")

# 向指定队列读取数据
def read_data(queue):
    while True:
        # 判断队列是否为空
        if queue.empty():
            print("队列为空")
            break
        else:
            value = queue.get()
            print(value)

if __name__ == '__main__':
    # 创建进程池中的queue
    queue = multiprocessing.Manager().Queue()

    # 创建进程池
    pool = multiprocessing.Pool(2)

    # 通过进程池执行写入数据的任务
    # pool.apply(write_data, (queue,))
    # # 通过进程池执行读取数据的任务
    # pool.apply(read_data, (queue,))

    # 使用异步执行对应的任务
    # 执行异步异步任务会返回一个应用结果对象，这个对象里面有一个wait()方法可以等待异步任务执行完成以后代码再继续往下执行
    # <multiprocessing.pool.ApplyResult object at 0x7f22d1795e48>
    result = pool.apply_async(write_data, (queue,))
    # 等待写入这个异步任务执行完成以后代码再继续执行
    result.wait()
    # time.sleep(1.1)
    pool.apply_async(read_data, (queue,))

    # 关闭进程池
    pool.close()
    # 主进程等待进程池执行完任务以后程序再退出
    pool.join()

进程池: 池子里面放的都是进程，进程池可以根据任务自动创建进程，会合理利用进程池中的进程完成多任务，能够节省资源的开销
默认情况下，主进程不会等待进程池结束
使用组合拳，pool.close() 和 pool.join() 让主进程等待进程池任务执行结束
进程池分为同步执行pool.apply 和异步执行pool.apply_async

到此结 DragonFangQy 2018.4.29

共享数据的方式 ↩

Python学习笔记十三（多任务、进程、进程池）

多任务之进程

多进程多任务

设置守护进程

多进程与全局变量^[1]

进程之间的资源竞争

进程池

猜你喜欢

热点阅读

Python学习笔记十三（多任务、进程、进程池）

多任务之进程

多进程多任务

设置守护进程

多进程与全局变量[1]

进程之间的资源竞争

进程池

猜你喜欢

热点阅读

多进程与全局变量^[1]