Python多线程与多进程怎么选？详解GIL锁与多核CPU利用

import os

print('process (%s) start...' % os.getpid())
# only works on unix/linux/macos:
pid = os.fork()
if pid == 0:
    print('i am child process (%s) and my parent is %s.' % (os.getpid(), os.getppid()))
else:
    print('i (%s) just created a child process (%s).' % (os.getpid(), pid))

process (876) start...
i (876) just created a child process (877).
i am child process (877) and my parent is 876.

from multiprocessing import process
import os

# 子进程要执行的代码
def run_proc(name):
    print('run child process %s (%s)...' % (name, os.getpid()))

if __name__=='__main__':
    print('parent process %s.' % os.getpid())
    p = process(target=run_proc, args=('test',))
    print('child process will start.')
    p.start()
    p.join()
    print('child process end.')

parent process 928.
child process will start.
run child process test (929)...
process end.

from multiprocessing import pool
import os, time, random

def long_time_task(name):
    print('run task %s (%s)...' % (name, os.getpid()))
    start = time.time()
    time.sleep(random.random() * 3)
    end = time.time()
    print('task %s runs %0.2f seconds.' % (name, (end - start)))

if __name__=='__main__':
    print('parent process %s.' % os.getpid())
    p = pool(4)
    for i in range(5):
        p.apply_async(long_time_task, args=(i,))
    print('waiting for all subprocesses done...')
    p.close()
    p.join()
    print('all subprocesses done.')

parent process 669.
waiting for all subprocesses done...
run task 0 (671)...
run task 1 (672)...
run task 2 (673)...
run task 3 (674)...
task 2 runs 0.14 seconds.
run task 4 (673)...
task 1 runs 0.27 seconds.
task 3 runs 0.86 seconds.
task 0 runs 1.41 seconds.
task 4 runs 1.91 seconds.
all subprocesses done.

import subprocess

print('$ nslookup www.python.org')
r = subprocess.call(['nslookup', 'www.python.org'])
print('exit code:', r)

$ nslookup www.python.org
server:		192.168.19.4
address:	192.168.19.4#53

non-authoritative answer:
www.python.org	canonical name = python.map.fastly.net.
name:	python.map.fastly.net
address: 199.27.79.223

exit code: 0

import subprocess

print('$ nslookup')
p = subprocess.popen(['nslookup'], stdin=subprocess.pipe, stdout=subprocess.pipe, stderr=subprocess.pipe)
output, err = p.communicate(b'set q=mx\npython.org\nexit\n')
print(output.decode('utf-8'))
print('exit code:', p.returncode)

set q=mx       # 设置查询类型为 mx（邮件服务器）
python.org     # 查询 python.org 域名
exit           # 退出 nslookup

$ nslookup
server:		192.168.19.4
address:	192.168.19.4#53

non-authoritative answer:
python.org	mail exchanger = 50 mail.python.org.

authoritative answers can be found from:
mail.python.org	internet address = 82.94.164.166
mail.python.org	has aaaa address 2001:888:2000:d::a6


exit code: 0

from multiprocessing import process, queue
import os, time, random

# 写数据进程执行的代码:
def write(q):
    print('process to write: %s' % os.getpid())
    for value in ['a', 'b', 'c']:
        print('put %s to queue...' % value)
        q.put(value)
        time.sleep(random.random())

# 读数据进程执行的代码:
def read(q):
    print('process to read: %s' % os.getpid())
    while true:
        value = q.get(true)
        print('get %s from queue.' % value)

if __name__=='__main__':
    # 父进程创建queue，并传给各个子进程：
    q = queue()
    pw = process(target=write, args=(q,))
    pr = process(target=read, args=(q,))
    # 启动子进程pw，写入:
    pw.start()
    # 启动子进程pr，读取:
    pr.start()
    # 等待pw结束:
    pw.join()
    # pr进程里是死循环，无法等待其结束，只能强行终止:
    pr.terminate()

process to write: 50563
put a to queue...
process to read: 50564
get a from queue.
put b to queue...
get b from queue.
put c to queue...
get c from queue.

import time, threading

# 新线程执行的代码:
def loop():
    print('thread %s is running...' % threading.current_thread().name)
    n = 0
    while n < 5:
        n = n + 1
        print('thread %s >>> %s' % (threading.current_thread().name, n))
        time.sleep(1)
    print('thread %s ended.' % threading.current_thread().name)

print('thread %s is running...' % threading.current_thread().name)
t = threading.thread(target=loop, name='loopthread')
t.start()
t.join()
print('thread %s ended.' % threading.current_thread().name)

thread mainthread is running...
thread loopthread is running...
thread loopthread >>> 1
thread loopthread >>> 2
thread loopthread >>> 3
thread loopthread >>> 4
thread loopthread >>> 5
thread loopthread ended.
thread mainthread ended.

# multithread
import time, threading

# 假定这是你的银行存款:
balance = 0

def change_it(n):
    # 先存后取，结果应该为0:
    global balance
    balance = balance + n
    balance = balance - n

def run_thread(n):
    for i in range(10000000):
        change_it(n)

t1 = threading.thread(target=run_thread, args=(5,))
t2 = threading.thread(target=run_thread, args=(8,))
t1.start()
t2.start()
t1.join()
t2.join()
print(balance)

balance = balance + n

初始值 balance = 0

t1: x1 = balance + 5 # x1 = 0 + 5 = 5
t1: balance = x1     # balance = 5
t1: x1 = balance - 5 # x1 = 5 - 5 = 0
t1: balance = x1     # balance = 0

t2: x2 = balance + 8 # x2 = 0 + 8 = 8
t2: balance = x2     # balance = 8
t2: x2 = balance - 8 # x2 = 8 - 8 = 0
t2: balance = x2     # balance = 0
    
结果 balance = 0

初始值 balance = 0

t1: x1 = balance + 5  # x1 = 0 + 5 = 5

t2: x2 = balance + 8  # x2 = 0 + 8 = 8
t2: balance = x2      # balance = 8

t1: balance = x1      # balance = 5
t1: x1 = balance - 5  # x1 = 5 - 5 = 0
t1: balance = x1      # balance = 0

t2: x2 = balance - 8  # x2 = 0 - 8 = -8
t2: balance = x2      # balance = -8

结果 balance = -8

balance = 0
lock = threading.lock()

def run_thread(n):
    for i in range(100000):
        # 先要获取锁:
        lock.acquire()
        try:
            # 放心地改吧:
            change_it(n)
        finally:
            # 改完了一定要释放锁:
            lock.release()

import threading

class singleton:
    _instance = none
    _lock = threading.lock()  # 类级别锁，保证线程安全

    def __new__(cls, *args, **kwargs):
        if not cls._instance:  # 第一层检查，提高效率
            with cls._lock:    # 加锁，防止多线程同时创建
                if not cls._instance:  # 第二层检查，防止重复创建
                    cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self, value=none):
        self.value = value

# 测试多线程
def task(name):
    obj = singleton(name)
    print(f"{name} -> {obj.value}, id: {id(obj)}")

threads = []
for i in range(5):
    t = threading.thread(target=task, args=(f"thread-{i}",))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

import threading, multiprocessing

def loop():
    x = 0
    while true:
        x = x ^ 1

for i in range(multiprocessing.cpu_count()):
    t = threading.thread(target=loop)
    t.start()

def process_student(name):
    std = student(name)
    # std是局部变量，但是每个函数都要用它，因此必须传进去：
    do_task_1(std)
    do_task_2(std)

def do_task_1(std):
    do_subtask_1(std)
    do_subtask_2(std)

def do_task_2(std):
    do_subtask_2(std)
    do_subtask_2(std)

global_dict = {}

def std_thread(name):
    std = student(name)
    # 把std放到全局变量global_dict中：
    global_dict[threading.current_thread()] = std
    do_task_1()
    do_task_2()

def do_task_1():
    # 不传入std，而是根据当前线程查找：
    std = global_dict[threading.current_thread()]
    ...

def do_task_2():
    # 任何函数都可以查找出当前线程的std变量：
    std = global_dict[threading.current_thread()]
    ...

import threading
    
# 创建全局threadlocal对象:
local_school = threading.local()

def process_student():
    # 获取当前线程关联的student:
    std = local_school.student
    print('hello, %s (in %s)' % (std, threading.current_thread().name))

def process_thread(name):
    # 绑定threadlocal的student:
    local_school.student = name
    process_student()

t1 = threading.thread(target= process_thread, args=('alice',), name='thread-a')
t2 = threading.thread(target= process_thread, args=('bob',), name='thread-b')
t1.start()
t2.start()
t1.join()
t2.join()

import threading
    
# 创建全局threadlocal对象:
local_school = threading.local()

def process_student():
    # 获取当前线程关联的student:
    std = local_school.student
    print('hello, %s (in %s)' % (std, threading.current_thread().name))

def process_thread(name):
    # 绑定threadlocal的student:
    local_school.student = name
    process_student()

t1 = threading.thread(target= process_thread, args=('alice',), name='thread-a')
t2 = threading.thread(target= process_thread, args=('bob',), name='thread-b')
t1.start()
t2.start()
t1.join()
t2.join()

async def coroutine():
    print("hello world")

async def coroutine_add_one(number):
    return number + 1
def add_one(number):
    return number + 1
function_result = add_one(1)
coroutine_result = coroutine_add_one(1)
print(function_result)
print(type(function_result))
"""
2
<class 'int'>
"""
print(coroutine_result)
print(type(coroutine_result))
"""
<coroutine object coroutine_add_one at 0x000002977045bac0>
<class 'coroutine'>
"""

import asyncio
async def coroutine_add_one(number):
    return number + 1
coroutine_result = asyncio.run(coroutine_add_one(1))
print(coroutine_result)  # 2

import asyncio
async def add_one(number):
    return number + 1
async def main():
    # main() 协程将暂停执行，直到 add_one(1) 运行完毕
    one_plus_one = await add_one(1)
    # main() 协程将暂停执行，直到 add_one(2) 运行完毕
    two_plus_one = await add_one(2)
    print(one_plus_one)
    print(two_plus_one)
asyncio.run(main())
"""
2
3
"""

import asyncio
async def hello_world():
    # 暂停 hello_world 协程一秒钟
    await asyncio.sleep(1)
    return "hello world"
async def main():
    # 暂停 main 协程，直到 hello_world 协程运行完毕
    message = await hello_world()
    print(message)
asyncio.run(main())
"""
hello world
"""

import asyncio

async def delay(seconds):
    print(f"开始休眠 {seconds} 秒")
    await asyncio.sleep(seconds)
    print(f"休眠完成")
    return seconds

async def add_one(number):
    return number + 1

async def hello_world():
    await delay(1)
    return "hello world"

async def main():
    # 暂停 main()，直到 add_one(1) 返回    
    one_plus_one = await add_one(1)
    # 暂停 main()，直到 hello_world() 返回
    message = await hello_world()
    print(one_plus_one)
    print(message)

asyncio.run(main())
"""
开始休眠 1 秒
休眠完成
2
hello world
"""

import asyncio

async def delay(seconds):
    print(f"开始休眠 {seconds} 秒")
    await asyncio.sleep(seconds)
    print(f"休眠完成")
    return seconds

async def main():
    # 将 delay(3) 包装成任务，注：包装完之后直接就丢到事件循环里面运行了
    # 因此这里会立即返回，而返回值是一个 asyncio.task 对象
    sleep_for_three = asyncio.create_task(delay(3))
    print("sleep_for_three:", sleep_for_three.__class__)
    # 至于协程究竟有没有运行完毕，我们可以通过 task 对象来查看
    # 当协程运行完毕或者报错，都看做是运行完毕了，那么调用 task 对象的 done 方法会返回 true
    # 否则返回 false，由于代码是立即执行，还没有到 3 秒钟，因此打印结果为 false
    print("协程(任务)是否执行完毕:", sleep_for_three.done())
    # 这里则保证必须等到 task 对象里面的协程运行完毕后，才能往下执行
    result = await sleep_for_three
    print("协程(任务)是否执行完毕:", sleep_for_three.done())
    print("返回值:", result)

asyncio.run(main())
"""
开始休眠 3 秒
sleep_for_three: <class '_asyncio.task'>
协程(任务)是否执行完毕: false
休眠完成
协程(任务)是否执行完毕: true
返回值: 3
"""

import asyncio

async def delay(seconds):
    print(f"开始休眠 {seconds} 秒")
    await asyncio.sleep(seconds)
    print(f"休眠完成")
    return seconds

async def main():
    sleep_for_three = asyncio.create_task(delay(3))
    sleep_again = asyncio.create_task(delay(3))
    sleep_once_more = asyncio.create_task(delay(3))

    await sleep_for_three
    await sleep_again
    await sleep_once_more

asyncio.run(main())
"""
开始休眠 3 秒
开始休眠 3 秒
开始休眠 3 秒
休眠完成
休眠完成
休眠完成
"""

import asyncio

async def delay(seconds):
    print(f"开始休眠 {seconds} 秒")
    await asyncio.sleep(seconds)
    print(f"休眠完成")
    return seconds

async def hello_from_second():
    for i in range(10):
        await asyncio.sleep(1)
        print("你好，我每秒钟负责打印一次")

async def main():
    sleep_for_three = asyncio.create_task(delay(3))
    sleep_again = asyncio.create_task(delay(3))

    await hello_from_second()

asyncio.run(main())
"""
开始休眠 3 秒
开始休眠 3 秒
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
休眠完成
休眠完成
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
你好，我每秒钟负责打印一次
"""

import asyncio

async def delay(seconds):
    print(f"开始休眠 {seconds} 秒")
    await asyncio.sleep(seconds)
    print(f"休眠完成")
    return seconds

async def main():
    long_task = asyncio.create_task(delay(10))
    seconds_elapsed = 0

    while not long_task.done():
        print("检测到任务尚未完成，一秒钟之后继续检测")
        await asyncio.sleep(1)
        seconds_elapsed += 1
        # 时间超过 5 秒，取消任务
        if seconds_elapsed == 5:
            long_task.cancel()

    try:
        # 等待 long_task 完成，显然执行到这里的时候，任务已经被取消
        # 不管是 await 一个已经取消的任务，还是 await 的时候任务被取消
        # 都会引发 asyncio.cancellederror
        await long_task
    except asyncio.cancellederror:
        print("任务被取消")

asyncio.run(main())
"""
检测到任务尚未完成，一秒钟之后继续检测
开始休眠 10 秒
检测到任务尚未完成，一秒钟之后继续检测
检测到任务尚未完成，一秒钟之后继续检测
检测到任务尚未完成，一秒钟之后继续检测
检测到任务尚未完成，一秒钟之后继续检测
检测到任务尚未完成，一秒钟之后继续检测
任务被取消
"""

import asyncio

async def delay(seconds):
    print(f"开始休眠 {seconds} 秒")
    await asyncio.sleep(seconds)
    print(f"休眠完成")
    return seconds

async def main():
    long_task = asyncio.create_task(delay(3))
    # 立刻取消
    long_task.cancel()
    # 但 cancellederror 只有在 await 取消的协程时才会触发
    # 所以下面的语句会正常执行
    print("我会正常执行")
    print("hello world")
    print(list(range(10)))
    await asyncio.sleep(5)
    try:
        # 引发 cancellederror
        await long_task
    except asyncio.cancellederror:
        print("任务被取消")

asyncio.run(main())
"""
我会正常执行
hello world
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
任务被取消
"""

import asyncio

async def delay(seconds):
    print(f"开始休眠 {seconds} 秒")
    await asyncio.sleep(seconds)
    print(f"休眠完成")
    return seconds

async def main():
    long_task = asyncio.create_task(delay(3))
    await asyncio.sleep(5)
    # 显然执行到这里，任务已经结束了
    long_task.cancel()
    try:
        await long_task
        print("任务执行完毕")
    except asyncio.cancellederror:
        print("任务被取消")

asyncio.run(main())
"""
开始休眠 3 秒
休眠完成
任务执行完毕
"""

import asyncio

async def delay(seconds):
    print(f"开始休眠 {seconds} 秒")
    await asyncio.sleep(seconds)
    print(f"休眠完成")
    return seconds

async def main():
    delay_task = asyncio.create_task(delay(2))
    try:
        result = await asyncio.wait_for(delay_task, 1)
        print("返回值:", result)
    except asyncio.timeouterror:
        print("超时啦")
        # delay_task.cancelled() 用于判断任务是否被取消
        # 任务被取消：返回 true，没有被取消：返回 false
        print("任务是否被取消:", delay_task.cancelled())

asyncio.run(main())
"""
开始休眠 2 秒
超时啦
任务是否被取消: true
"""

import asyncio

async def delay(seconds):
    print(f"开始休眠 {seconds} 秒")
    await asyncio.sleep(seconds)
    print(f"休眠完成")
    return seconds

async def main():
    delay_task = asyncio.create_task(delay(2))
    try:
        # 通过 asyncio.shield 将 delay_task 保护起来
        result = await asyncio.wait_for(asyncio.shield(delay_task), 1)
        print("返回值:", result)
    except asyncio.timeouterror:
        print("超时啦")
        # 如果超时依旧会引发 timeouterror，但和之前不同的是
        # 此时任务不会被取消了，因为 asyncio.shield 会将取消请求忽略掉
        print("任务是否被取消:", delay_task.cancelled())
        # 从出现超时的地方，继续执行，并等待它完成
        result = await delay_task
        print("返回值:", result)

asyncio.run(main())
"""
开始休眠 2 秒
超时啦
任务是否被取消: false
休眠完成
返回值: 2
"""