进程，线程和协程

Process, Thread and Coroutine

实现篇

分类: 编程 / 字数: 8071 / 标签: 进程, Process, 线程, Thread, 协程, Coroutine, CPU 密集型, IO 密集型, 磁盘 IO 密集型, 网络 IO 密集型, 模拟 IO 密集型, 互斥锁, 可重入锁, 嵌套锁, 死锁, 全局解释器锁, Global Interpreter Lock, GIL, 事件, Event, 队列, Queue, 池, 进程池, 线程池, 容器, 可迭代对象, 生成器, 迭代器, yield, 双向通道, asyncio, 异步 IO, 事件循环, 回调, Callback

import requests

# CPU 密集型
def cpu_bound_task(x=1, y=1):
    c = 0

    while c < 1500000:
        c += 1
        x += x
        y += y

# 磁盘 IO 密集型
def disk_io_bound_task():
    with open('tmp.log', 'w') as f:
        for idx in range(5000000):
            f.write('{}\n'.format(idx))

# 网络 IO 密集型
def web_io_bound_task():
    try:
        requests.get('https://www.baidu.com')
    except Exception as e:
        pass

# 模拟 IO 密集型
def simulation_io_bound_task():
    time.sleep(2)

import time

def timer(task_mode):
    def wrapper(func):
        def decorator(*args, **kwargs):
            task_type = kwargs.setdefault('task_type', None)
            start_time = time.time()
            func(*args, **kwargs)
            end_time = time.time()
            print('耗时（{} - {}）: {}'.format(
                task_mode, task_type, end_time - start_time))
        return decorator
    return wrapper

from threading import Thread
from multiprocessing import Process

@timer('单线程')
def single_thread(func, task_type='', n=10):
    for idx in range(n):
        func()


@timer('多线程')
def multi_threads(func, task_type='', n=10):
    threads = {}

    for idx in range(n):
        t = Thread(target=func)
        threads[idx] = t
        t.start()

    for thread in threads.values():
        thread.join()


@timer('多进程')
def multi_processes(func, task_type='', n=10):
    processes = {}

    for idx in range(n):
        p = Process(target=func)
        processes[idx] = p
        p.start()

    for process in processes.values():
        process.join()

# 单线程
single_thread(cpu_bound_task, task_type='CPU 密集型任务')
single_thread(disk_io_bound_task, task_type='磁盘 IO 密集型任务')
single_thread(web_io_bound_task, task_type='网络 IO 密集型任务')
single_thread(simulation_io_bound_task, task_type='模拟 IO 密集型任务')

# 多线程
multi_threads(cpu_bound_task, task_type='CPU 密集型任务')
multi_threads(disk_io_bound_task, task_type='磁盘 IO 密集型任务')
multi_threads(web_io_bound_task, task_type='网络 IO 密集型任务')
multi_threads(simulation_io_bound_task, task_type='模拟 IO 密集型任务'

# 多进程
multi_processes(cpu_bound_task, task_type='CPU 密集型任务')
multi_processes(disk_io_bound_task, task_type='磁盘 IO 密集型任务')
multi_processes(web_io_bound_task, task_type='网络 IO 密集型任务')
multi_processes(simulation_io_bound_task, task_type='模拟 IO 密集型任务')

import time
from threading import Thread

# 自定义函数
def func(name='Python'):
    for idx in range(2):
        print('Hello, {}'.format(name))
        time.sleep(1)

# 创建线程
thread_1 = Thread(target=func)
thread_2 = Thread(target=func, args=('Leo', ))

# 启动线程
thread_1.start()
thread_2.start()

# 自定义类
class MyThread(Thread):
    def __init__(self, name='Python'):
        super(MyThread, self).__init__()
        self.name = name

    def run(self):
        for idx in range(2):
            print('Hello, {}'.format(self.name))
            time.sleep(1)

# 创建线程
thread_1 = MyThread()
thread_2 = MyThread('Leo')

# 启动线程
thread_1.start()
thread_2.start()

# 创建线程
t = Thread(target=func)

# 启动线程
t.start()

# 阻塞线程
t.join()

# 判断线程是否处于执行状态
# True: 执行中，False: 其他
t.is_alive()

# 这是线程是否随主线程退出而退出
# 默认为 False
t.daemon = True

# 设置线程名称
t.name = 'My Thread'

import time
import random

from threading import Thread

# 共用变量
C = 0

def job1(n=10):
    global C

    for idx in range(n):
        C += 1
        print('Job1: {}'.format(C))

def job2(n=10):
    global C

    for idx in range(n):
        C += 10
        print('Job2: {}'.format(C))

t1 = Thread(target=job1)
t2 = Thread(target=job2)

t1.start()
t2.start()

import time
import random

from threading import Lock

# 全局唯一锁
LOCK = Lock()

# 共用变量
C = 0

def job1_with_lock(n=10):
    global C, LOCK

    LOCK.acquire()

    for idx in range(n):
        C += 1
        print('Job1: {}'.format(C))
        time.sleep(random.random())

    LOCK.release()


def job2_with_lock(n=10):
    global C, LOCK

    LOCK.acquire()

    for idx in range(n):
        C += 10
        print('Job2: {}'.format(C))
        time.sleep(random.random())

    LOCK.release()

t1 = Thread(target=job1_with_lock)
t2 = Thread(target=job2_with_lock)

t1.start()
t2.start()

from threading import Lock

def lock_with_lock(n=10):
    c = 0
    lock = Lock()

    with lock:
        for idx in range(n):
            c += 1
            with lock:
                print(c)

t = Thread(target=lock_with_lock)
t.start()

from threading import RLock

def rlock_with_lock(n=10):
    c = 0
    lock = RLock()

    with lock:
        for idx in range(n):
            c += 1
            with lock:
                print(c)
  
t = Thread(target=rlock_with_lock)
t.start()

event = threading.Event()

# 重置 event，使得所有该 event 事件都处于待命状态
event.clear()

# 等待接收 event 的指令，决定是否阻塞程序执行
event.wait()

# 发送 event 指令，使所有设置该 event 事件的线程执行
event.set()

import time

from threading import Thread, Event

class EventThread(Thread):
    def __init__(self, name, event):
        super(EventThread, self).__init__()

        self.name = name
        self.event = event

    def run(self):
        print('线程 {} 启动于 {}'.format(self.name, time.ctime(time.time())))
        self.event.wait()
        print('线程 {} 结束于 {}'.format(self.name, time.ctime(time.time())))

threads = {}
event = Event()

for tid in range(3):
    threads[tid] = EventThread(str(tid), event)

event.clear()

for thread in threads.values():
    thread.start()

print('等待 3 秒钟 ...')
time.sleep(3)

print('唤醒所有线程 ...')
event.set() 

cond = threading.Condition()

# 类似 lock.acquire()
cond.acquire()

# 类似 lock.release()
cond.release()

# 等待指定触发，同时会释放对锁的获取,直到被 notify 才重新占有琐。
cond.wait()

# 发送指定，触发执行
cond.notify()

import time

from threading import Thread, Condition

class Seeker(Thread):
    def __init__(self, condition, name):
        super(Seeker, self).__init__()

        self.condition = condition
        self.name = name

    def run(self):
        time.sleep(1) # 确保先运行 Hider 中的方法

        self.condition.acquire()

        print('{}: 我把眼睛蒙好了'.format(self.name))
        self.condition.notify()
        self.condition.wait()
        print('{}: 我找到你了'.format(self.name))
        self.condition.notify()

        self.condition.release()
        print('{}: 我赢了'.format(self.name))


class Hider(Thread):
    def __init__(self, condition, name):
        super(Hider, self).__init__()

        self.condition = condition
        self.name = name

    def run(self):
        self.condition.acquire()

        self.condition.wait()
        print('{}: 我藏好了'.format(self.name))
        self.condition.notify()
        self.condition.wait()
        self.condition.release()
        print('{}: 被你找到了'.format(self.name))

condition = Condition()

seeker = Seeker(condition, 'Seeker')
hider = Hider(condition, 'Hider')

seeker.start()
hider.start()

from queue import Queue

# maxsize=0 表示不限大小
# maxsize>0 且消息数达到限制时，put() 方法会阻塞
q = Queue(maxsize=0)

# 默认阻塞程序，等待队列消息，可设置超时时间
q.get(block=True, timeout=None)

# 发送消息，默认会阻塞程序至队列中有空闲位置放入数据
q.put(item, block=True, timeout=None)

# 等待所有的消息都被消费完
q.join()

# 通知队列任务处理已经完成，当所有任务都处理完成时，join() 阻塞将会解除
q.task_done()

# 查询当前队列的消息个数
q.qsize()

# 队列消息是否都被消费完，返回 True/False
q.empty()

# 检测队列里消息是否已满
q.full()

import time

from queue import Queue
from threading import Thread

class Student(object):
    def __init__(self, name):
        super(Student, self).__init__()

        self.name = name

    def speak(self):
        print('{}: 到'.format(self.name))


class Teacher(object):
    def __init__(self, queue):
        super(Teacher, self).__init__()

        self.queue = queue

    def call(self, student_name):
        if student_name == 'exit':
            print('老师: 点名结束，开始上课')
        else:
            print('老师: {}'.format(student_name))

        self.queue.put(student_name)


class CallManager(Thread):
    def __init__(self, queue):
        super(CallManager, self).__init__()

        self.students = {}
        self.queue = queue

    def put(self, student):
        self.students.setdefault(student.name, student)

    def run(self):
        while True:
            student_name = self.queue.get()

            if student_name == 'exit':
                break
            elif student_name in self.students:
                self.students[student_name].speak()
            else:
                print('学生: 老师，没有 {} 这个人'.format(student_name))

queue = Queue()

teacher = Teacher(queue=queue)
s1 = Student(name='张三')
s2 = Student(name='李四')

cm = CallManager(queue)
cm.put(s1)
cm.put(s2)

cm.start()

print('开始点名')
teacher.call('张三')
time.sleep(1)
teacher.call('李四')
time.sleep(1)
teacher.call('王五')
time.sleep(1)
teacher.call('exit')

import time
import threading

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed


def print_func(n=3):
    for idx in range(n):
        print('运行 {}-{}'.format(threading.get_ident(), idx))
        time.sleep(1)


def return_func(n=3):
    res = []

    for idx in range(n):
        res.append('{}-{}'.format(threading.get_ident(), idx))
        time.sleep(1)

    return res


def test_thread_pool_print(n=3, m=12):
    with ThreadPoolExecutor(max_workers=n) as executor:
        for _ in range(m):
            executor.submit(print_func)


def test_process_pool_print(n=3, m=12):
    with ProcessPoolExecutor(max_workers=n) as executor:
        for _ in range(m):
            executor.submit(print_func)


def test_thread_pool_return(n=3, m=12):
    with ThreadPoolExecutor(max_workers=n) as executor:
        futures = [executor.submit(return_func) for _ in range(m)]

        for future in as_completed(futures):
            print(future.result())


def test_process_pool_return(n=3, m=12):
    with ProcessPoolExecutor(max_workers=n) as executor:
        futures = [executor.submit(return_func) for _ in range(m)]

        for future in as_completed(futures):
            print(future.result())


line_sep = '-' * 60

print(line_sep)
print('测试线程池')
print(line_sep)
test_thread_pool_print()
print(line_sep)

print(line_sep)
print('测试进程池')
print(line_sep)
test_process_pool_print()
print(line_sep)

print(line_sep)
print('测试线程池')
print(line_sep)
test_thread_pool_return()
print(line_sep)

print(line_sep)
print('测试进程池')
print(line_sep)
test_process_pool_return()
print(line_sep)

from collections import deque
from collections.abc import Iterable

print(isinstance('Leo Van', Iterable))
print(isinstance([1, 2, 3], Iterable))
print(isinstance({'k1': 'v1', 'k2': 'v2'}, Iterable))
print(isinstance(deque('abc'), Iterable))

class MyList(object):
    def __init__(self, end):
        super(MyList, self).__init__()

        self.end = end

    def __iter__(self):
        return MyListIterator(self.end)

    def __repr__(self):
        return '[{}]'.format(', '.join([str(ele) for ele in self]))


class MyListIterator(object):
    def __init__(self, end):
        super(MyListIterator, self).__init__()

        self.data = end
        self.start = 0

    def __iter__(self):
        return self

    def __next__(self):
        while self.start < self.data:
            self.start += 1
            return self.start - 1

        raise StopIteration

my_list = MyList(3)

print('MyList: {}'.format(my_list))
print(isinstance(my_list, Iterable))
print(isinstance(my_list, Iterator))

for ele in my_list:
    print(ele)

my_list_iterator = MyListIterator(3)

print('MyListIterator: {}'.format(my_list_iterator))
print(isinstance(my_list_iterator, Iterable))
print(isinstance(my_list_iterator, Iterator))

my_iterator = iter(my_list)

print('MyIterator: {}'.format(my_iterator))
print(isinstance(my_iterator, Iterable))
print(isinstance(my_iterator, Iterator))

while True:
    try:
        print(next(my_iterator))
    except StopIteration as e:
        return

from collections.abc import Generator
from inspect import getgeneratorstate

a_list = [x for x in range(10)]
print(a_list)
print(isinstance(a_list, Generator))

a_generator = (x for x in range(10))
print(a_generator)
print(isinstance(a_generator, Generator))

def my_yield(n):
    now = 0

    while now < n:
        yield now
        now += 1

    raise StopIteration

gen = my_yield(4)
print(gen)
print(isinstance(gen, Generator))

def my_yield(n):
    now = 0

    while now < n:
        yield now
        now += 1

    raise StopIteration

gen = my_yield(4)
print(gen.send(None))
print(next(gen))
print(gen.send(None))
print(next(gen))

from collections.abc import Generator
from inspect import getgeneratorstate

def my_yield(n):
    now = 0

    while now < n:
        yield now
        now += 1

    raise StopIteration

gen = my_yield(4)
print(gen)
print(getgeneratorstate(gen))

print(gen.send(None))
print(next(gen))
print(getgeneratorstate(gen))

print(gen.send(None))
print(next(gen))
print(getgeneratorstate(gen))

gen.close()
print(getgeneratorstate(gen))

def jump_range(n):
    idx = 0

    while idx < n:
        jump = yield idx
        print('[idx: {}, jump: {}]'.format(idx, jump))

        if jump is None:
            jump = 1

        idx += jump

itr = jump_range(6)
print(next(itr))
print(itr.send(2))
print(next(itr))
print(itr.send(-1))
print(next(itr))
print(next(itr))

a_str = 'Leo'
a_list = [1, 2, 3]
a_dict = {'name': 'Leo', 'gender': 'Male'}
a_gen = (idx for idx in range(4, 8))

def gen(*args, **kwargs):
    for item in args:
        for ele in item:
            yield ele

new_gen = gen(a_str, a_list, a_dict, a_gen)
print(list(new_gen))

a_gen = (idx for idx in range(4, 8))

def gen_from(*args, **kwargs):
    for item in args:
        yield from item

new_gen = gen_from(a_str, a_list, a_dict, a_gen)
print(list(new_gen))

# 子生成器
def average_gen():
    total = 0
    count = 0
    average = 0

    while True:
        num = yield average

        if num is None:
            break

        count += 1
        total += num
        average = total / count

    return total, count, average

# 委托生成器
def proxy_gen():
    while True:
        total, count, average = yield from average_gen()
        print('计算完毕，共输入 {} 个数值，总和 {}，平均值 {}'.format(
            count, total, average))

# 调用方
calc_average = proxy_gen()
next(calc_average)
print(calc_average.send(10))
print(calc_average.send(20))
print(calc_average.send(30))
calc_average.send(None)

from collections.abc import Coroutine

async def async_func(name):
    print('Hello, ', name)

coroutine = async_func('World')
print(isinstance(coroutine, Coroutine))

import asyncio

from collections.abc import Generator, Coroutine

@asyncio.coroutine
def coroutine_func(name):
    print('Hello,', name)
    yield from asyncio.sleep(1)
    print('Bye,', name)


coroutine = coroutine_func('World')
print(isinstance(coroutine, Generator))
print(isinstance(coroutine, Coroutine))

import asyncio

async def hello(name):
    print('Hello,', name)
    
# 定义协程
coroutine = hello('World')

# 定义事件循环
loop = asyncio.get_event_loop()

# 创建任务
task = loop.create_task(coroutine)

# 将任务交由时间循环并执行
loop.run_until_complete(task)

import asyncio
from asyncio.futures import Future

async def hello(name):
    await asyncio.sleep(2)
    print('Hello, ', name)

coroutine = hello("World")

# 将协程转为 task 对象
task = asyncio.ensure_future(coroutine)

print(isinstance(task, Future))

import asyncio
import time

async def _sleep(x):
    time.sleep(x)
    return '暂停了 {} 秒'.format(x)

coroutine = _sleep(2)
loop = asyncio.get_event_loop()
task = asyncio.ensure_future(coroutine)

loop.run_until_complete(task)
print('返回结果：{}'.format(task.result()))

import asyncio
import time

async def _sleep(x):
    time.sleep(x)
    return '暂停了 {} 秒'.format(x)

def callback(future):
    print('回调返回结果：{}'.format(future.result()))

coroutine = _sleep(2)
loop = asyncio.get_event_loop()
task = asyncio.ensure_future(coroutine)

task.add_done_callback(callback)
loop.run_until_complete(task)

import asyncio

async def do_some_work(x):
    print('等待中 ...')
    await asyncio.sleep(x)
    print('{} 秒后结束'.format(x))
    return x

# 协程对象
coroutine1 = do_some_work(1)
coroutine2 = do_some_work(2)
coroutine3 = do_some_work(4)

# 任务列表
tasks = [
    asyncio.ensure_future(coroutine1),
    asyncio.ensure_future(coroutine2),
    asyncio.ensure_future(coroutine3)
]

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

for task in tasks:
    print('任务结果：{}'.format(task.result()))

import asyncio

async def do_some_work(x):
    print('等待中 ...')
    await asyncio.sleep(x)
    print('{} 秒后结束'.format(x))
    return x

async def out_do_some_work():
    coroutine1 = do_some_work(1)
    coroutine2 = do_some_work(2)
    coroutine3 = do_some_work(4)

    tasks = [
        asyncio.ensure_future(coroutine1),
        asyncio.ensure_future(coroutine2),
        asyncio.ensure_future(coroutine3)
    ]

    dones, pendings = await asyncio.wait(tasks)

    for task in dones:
        print('任务结果：{}'.format(task.result()))

loop = asyncio.get_event_loop()
loop.run_until_complete(out_do_some_work())

results = await asyncio.gather(*tasks)
for result in results:
    print('任务结果：{}'.format(result))

coroutine = _sleep(10)
loop = asyncio.get_event_loop()
task = loop.create_task(coroutine)

print('Pending')

try:
    t = Thread(target=loop.run_until_complete, args=(task, ))
    t.start()
    print('Running')
    t.join()
except KeyboardInterrupt as e:
    task.cancel()
    print('Cancel')
finally:
    print('Done')

import random
import asyncio

async def random_sleep():
    await asyncio.sleep(random.uniform(0.5, 6))

loop = asyncio.get_event_loop()
tasks = [random_sleep() for _ in range(1, 10)]

dones, pendings = loop.run_until_complete(asyncio.wait(
    tasks, return_when=asyncio.FIRST_COMPLETED))
print('第一次完成的任务数：{}'.format(len(dones)))

dones, pendings = loop.run_until_complete(asyncio.wait(
    pendings, timeout=2))
print('第二次完成的任务数: {}'.format(len(dones)))

dones, pendings = loop.run_until_complete(asyncio.wait(pendings))
print('第三次完成的任务数：{}'.format(len(dones)))