【Python】Python多进程、多线程和协程简介

# -*- coding:utf-8 -*-
from multiprocessing import Process
from time import ctime, sleep

# 定义一个函数,用于多个进程执行
def job(n, sec):
    print("start job", n, "at:", ctime())
    sleep(sec)
    print("job", n, "done at:", ctime())


if __name__ == "__main__":
    p1 = Process(target=job, args=(1, 2)) #创建进程P1,实例化进程对象。target 指定要执行的函数,args 用元组（Tuple）形式向函数传递参数。
    p2 = Process(target=job, args=(2, 3)) #创建进程P2,实例化进程对象。target 指定要执行的函数,args 用元组（Tuple）形式向函数传递参数。
    p1.start() #异步启动进程。这行代码只负责向操作系统发出“启动进程”的指令，不会等待子进程执行完毕，主进程会立刻向下执行。
    p2.start() #异步启动进程。这行代码只负责向操作系统发出“启动进程”的指令，不会等待子进程执行完毕，主进程会立刻向下执行。

    print("main job") #主进程继续,打印信息
    
    p1.join()  #告诉主进程：“你必须在这里等着，直到 p1 进程完全运行结束，才能向下走。”
    print("main wait")  # 由于p1.join()阻塞,这里主进程会在2秒后打印,就是p1进程执行完后主进程才会往后执行
    p2.join() #告诉主进程：“你必须在这里等着，直到 p2 进程完全运行结束，才能向下走。”

    print("finished") #因为有 join() 的保护，这行代码百分之百是全程序最后一行打印的内容。它保证了主进程在所有子进程收尾后才退出。

# -*- coding:utf-8 -*-
from multiprocessing import Pool
from time import ctime, sleep
import time

def job(n, sec):
    print("start job", n, "at:", ctime())
    sleep(sec)
    print("job", n, "done at:", ctime())


if __name__ == "__main__":
    pool = Pool(processes=3)  #进程池容量3
    START_TIME = time.time()
    for i, j in zip([1, 2, 3, 4], [4, 3, 2, 1]):
        # 维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
        pool.apply_async(job, args=(i, j))
    pool.close()
    # 调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool，join函数等待所有子进程结束
    pool.join()
    end_offset = time.time() - START_TIME
    print(f"任务耗时: {end_offset:.4f} 秒")
    print("finished")

from concurrent.futures import ProcessPoolExecutor
from time import ctime, sleep
import time


def job(n, sec):
    print("start job", n, "at:", ctime())
    sleep(sec)
    print("job", n, "done at:", ctime())


if __name__ == '__main__':
    START_TIME = time.time()
    with ProcessPoolExecutor(max_workers=3) as executor:
        all_jobs = [executor.submit(job, i, j) for i, j in zip([1, 2, 3, 4], [4, 3, 2, 1])]
    end_offset = time.time() - START_TIME
    print(f"任务耗时: {end_offset:.4f} 秒")
    print("finished")

import threading
from time import ctime, sleep
import time


def job(n, sec):
    print("start job", n, "at:", ctime())
    sleep(sec)
    print("job", n, "done at:", ctime())


if __name__ == '__main__':
    threads = []
    for i, j in zip([1, 2, 3, 4], [4, 3, 2, 1]):
        t = threading.Thread(target=job, args=(i, j))
        threads.append(t)
    start_time = time.time()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    end_offset = time.time() - start_time
    print(f"任务耗时: {end_offset:.4f} 秒")
    print("finished")

import threading
from time import ctime, sleep
import time

# 创建一把全局线程锁
print_lock = threading.Lock()

def job(n, sec):
    # 使用 lock 保护启动打印
    with print_lock:
        print("start job", n, "at:", ctime())
    sleep(sec)
    # 使用 lock 保护启动打印
    with print_lock:
        print("job", n, "done at:", ctime())


if __name__ == '__main__':
    threads = []
    for i, j in zip([1, 2, 3, 4], [4, 3, 2, 1]):
        t = threading.Thread(target=job, args=(i, j))
        threads.append(t)
    start_time = time.time()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    end_offset = time.time() - start_time
    print(f"任务耗时: {end_offset:.4f} 秒")
    print("finished")

start job 1 at: Fri Jul 31 20:17:49 2026
start job 2 at: Fri Jul 31 20:17:49 2026
start job 3 at: Fri Jul 31 20:17:49 2026
start job 4 at: Fri Jul 31 20:17:49 2026
job 4 done at: Fri Jul 31 20:17:50 2026
job 3 done at: Fri Jul 31 20:17:51 2026
job 2 done at: Fri Jul 31 20:17:52 2026
job 1 done at: Fri Jul 31 20:17:53 2026
任务耗时: 4.0056 秒
finished

import threading
from time import ctime, sleep
import time


def job(n, sec):
    print(f"start job {n} at: {ctime()}")
    sleep(sec)
    print(f"job {n} done at: {ctime()}")


if __name__ == '__main__':
    threads = []
    for i, j in zip([1, 2, 3, 4], [4, 3, 2, 1]):
        t = threading.Thread(target=job, args=(i, j))
        threads.append(t)
    start_time = time.time()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    end_offset = time.time() - start_time
    print(f"任务耗时: {end_offset:.4f} 秒")
    print("finished")

start job 1 at: Fri Jul 31 20:21:55 2026
start job 2 at: Fri Jul 31 20:21:55 2026
start job 3 at: Fri Jul 31 20:21:55 2026
start job 4 at: Fri Jul 31 20:21:55 2026
job 4 done at: Fri Jul 31 20:21:56 2026
job 3 done at: Fri Jul 31 20:21:57 2026
job 2 done at: Fri Jul 31 20:21:58 2026
job 1 done at: Fri Jul 31 20:21:59 2026
任务耗时: 4.0054 秒
finished

import threading
from time import ctime, sleep
import time


class MyThread(threading.Thread):
    def __init__(self, func, args, name=""):
        threading.Thread.__init__(self)
        self.name = name
        self.func = func
        self.args = args

    def run(self):
        self.func(*self.args)


def job(n, sec):
    print(f"start job {n} at: {ctime()}")
    sleep(sec)
    print(f"job {n} done at: {ctime()}")


if __name__ == '__main__':
    threads = []
    for i, j in zip([1, 2, 3, 4], [4, 3, 2, 1]):
        t = MyThread(job, args=(i, j), name=job.__name__)
        threads.append(t)
    start_time = time.time()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    end_offset = time.time() - start_time
    print(f"任务耗时: {end_offset:.4f} 秒")
    print("finished")

from multiprocessing.dummy import Pool as ThreadPool
from time import ctime, sleep
import time

def job(n, sec):
    print(f"start job {n} at: {ctime()}")
    sleep(sec)
    print(f"job {n} done at: {ctime()}")


if __name__ == "__main__":
    pool = ThreadPool(processes=3)
    START_TIME = time.time()
    for i, j in zip([1, 2, 3, 4], [4, 3, 2, 1]):
        pool.apply_async(job, args=(i, j))
    pool.close()
    pool.join()
    end_offset = time.time() - START_TIME
    print(f"任务耗时: {end_offset:.4f} 秒")
    print("finished")

from concurrent.futures import ThreadPoolExecutor
from time import ctime, sleep
import time


def job(n, sec):
    print(f"start job {n} at: {ctime()}")
    sleep(sec)
    print(f"job {n} done at: {ctime()}")


if __name__ == '__main__':
    START_TIME = time.time()
    with ThreadPoolExecutor(max_workers=3) as executor:
        all_jobs = [executor.submit(job, i, j) for i, j in zip([1, 2, 3, 4], [4, 3, 2, 1])]
    end_offset = time.time() - START_TIME
    print(f"任务耗时: {end_offset:.4f} 秒")
    print("finished")

# -*- coding:utf-8 -*-
def consumer():
    r = ''
    while True:
        print("start...")
        n = yield r
        print("Consuming  running ....")
        if not n:
            return
        print('[CONSUMER] Consuming %s...' % n)
        r = '200 OK'


def produce(c):
    r = c.send(None)  # 等价于next(c)
    print('[PRODUCER] Consumer return: %s' % r)
    print("start consumer...")
    n = 0
    while n < 5:
        n = n + 1
        print('[PRODUCER] Producing %s...' % n)
        r = c.send(n)
        print('[PRODUCER] Consumer return: %s' % r)
    c.close()


c = consumer()
produce(c)

import asyncio

# 警告：以下代码在 Python 3.11+ 环境下运行会报错
@asyncio.coroutine
def compute_square(x):
    print(f"开始计算 {x} 的平方...")
    # yield from 相当于现在的 await，用于挂起当前协程并等待耗时 I/O 
    yield from asyncio.sleep(1)
    result = x * x
    print(f"计算完成: {result}")
    return result

@asyncio.coroutine
def main():
    print("主程序启动")
    # 等待 compute_square 执行完毕
    res = yield from compute_square(5)
    print(f"最终结果: {res}")

# 获取事件循环并运行
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

import asyncio

async def hello():
    print("Hello world!")
    # 异步调用asyncio.sleep(1):
    await asyncio.sleep(1)
    print("Hello again!")

asyncio.run(hello())

import asyncio
import threading

# 传入name参数:
async def hello(name):
    # 打印name和当前线程:
    print("Hello %s! (%s)" % (name, threading.current_thread))
    # 异步调用asyncio.sleep(1):
    await asyncio.sleep(1)
    print("Hello %s again! (%s)" % (name, threading.current_thread))
    return name

#用asyncio.gather()同时调度多个async函数：
async def main():
    L = await asyncio.gather(hello("Bob"), hello("Alice"))
    print(L)

asyncio.run(main())

import asyncio

async def wget(host):
    print(f"wget {host}...")
    # 连接80端口:
    reader, writer = await asyncio.open_connection(host, 80)
    # 发送HTTP请求:
    header = f"GET / HTTP/1.0\r\nHost: {host}\r\n\r\n"
    writer.write(header.encode("utf-8"))
    await writer.drain()

    # 读取HTTP响应:
    while True:
        line = await reader.readline()
        if line == b"\r\n":
            break
        print("%s header > %s" % (host, line.decode("utf-8").rstrip()))
    # Ignore the body, close the socket
    writer.close()
    await writer.wait_closed()
    print(f"Done {host}.")

async def main():
    await asyncio.gather(wget("www.sina.com.cn"), wget("www.sohu.com"), wget("www.163.com"))

asyncio.run(main())

import asyncio
import warnings

# 激活废弃警告，让 Python 3.12+ 显式打印出 get_event_loop() 的警告信息
warnings.simplefilter('default', DeprecationWarning)

async def background_job(name, duration):
    """一个简单的异步任务"""
    print(f"[{name}] 任务启动...")
    await asyncio.sleep(duration)
    print(f"[{name}] 任务在 {duration} 秒后成功完成！")
    return f"{name} 的数据"


# =====================================================================
# 方案 1：现代标准高层级模式（asyncio.run + get_running_loop）
# =====================================================================
async def modern_main():
    print("\n--- 开始方案 1：现代标准模式 ---")
    
    # 1. 在异步上下文内，安全地获取当前正在运行的 loop
    loop = asyncio.get_running_loop()
    print(f"成功获取运行中的 loop: {loop}")
    
    # 2. 使用 loop 形式创建并启动任务
    task1 = loop.create_task(background_job("Modern-Task-1", 1))
    
    # 3. 使用通常形式创建任务（内部也是自动找 running_loop）
    task2 = asyncio.create_task(background_job("Modern-Task-2", 1.5))
    
    # 并发等待它们完成
    results = await asyncio.gather(task1, task2)
    print(f"方案 1 完成，收到结果: {results}")


# =====================================================================
# 方案 2：底层手动管理模式（new_event_loop + set_event_loop）
# =====================================================================
def manual_loop_main():
    print("\n--- 开始方案 2：底层手动管理模式 ---")
    
    # 1. 显式创建一个全新的事件循环（替代旧的 get_event_loop 自动创建行为）
    custom_loop = asyncio.new_event_loop()
    
    # 2. 将该循环显式绑定到当前线程上下文中
    asyncio.set_event_loop(custom_loop)
    
    try:
        # 3. 使用这个 loop 形式创建并调度任务
        # 注意：这里尚未开始运行 loop，任务只是被放进了队列
        task = custom_loop.create_task(background_job("Manual-Task", 1))
        
        # 4. 显式运行该循环，直到指定的任务（或主逻辑）完成
        result = custom_loop.run_until_complete(task)
        print(f"方案 2 完成，收到结果: {result}")
        
    finally:
        # 5. 必须手动关闭循环以释放底层资源
        custom_loop.close()
        print("手动创建的 loop 已安全关闭。")


# =====================================================================
# 方案 3：为什么不推荐旧的 get_event_loop()？（演示警告）
# =====================================================================
def legacy_loop_demo():
    print("\n--- 演示方案 3：旧的 get_event_loop() 行为 ---")
    print("注意观察下方是否会触发 DeprecationWarning 警告：")
    
    try:
        # 在没有运行中的事件循环时调用此方法：
        # Python 3.10+ 会警告，Python 3.12+ 明确标记为废弃，未来版本将直接报错
        legacy_loop = asyncio.get_event_loop() 
        print(f"虽然拿到了隐式创建的 loop: {legacy_loop}")
        
        # 运行它以防止资源泄漏
        legacy_loop.run_until_complete(background_job("Legacy-Task", 0.5))
        legacy_loop.close()
    except Exception as e:
        print(f"运行失败（如果在某些严格的异步环境中会直接报错）: {e}")


# =====================================================================
# 主程序入口
# =====================================================================
if __name__ == "__main__":
    # 执行方案 1：这是 95% 的 Python 异步应用应该使用的标准入口
    asyncio.run(modern_main())
    
    # 执行方案 2：这是编写底层网络框架或多线程服务端时使用的控制入口
    manual_loop_main()
    
    # 执行方案 3：演示旧方法的弊端
    legacy_loop_demo()

import asyncio
import time

# 1. 声明一个协程函数
async def say_hello():
    print("Hello")
    # 注意：在异步代码中，绝对不能使用标准的 time.sleep()！
    # 因为 time.sleep 是同步阻塞的，会卡死整个单线程事件循环。
    await asyncio.sleep(1)
    print("World")

if __name__ == "__main__":
    # 2. asyncio.run 是高层入口，负责自动创建事件循环并运行协程
    asyncio.run(say_hello())

import asyncio
import time


async def fetch_data(id, delay):
    print(f"【📥 开始】下载任务 {id}...")
    await asyncio.sleep(delay)  # 模拟网络请求耗时
    print(f"【✅ 完成】任务 {id} 结束")
    return f"数据 {id}"


async def main():
    start = time.time()

    # 方式 A：使用 asyncio.gather 并发运行多个协程
    # gather 会把这些协程自动打包成 Task 并发执行，并等待所有任务完成
    results = await asyncio.gather(
        fetch_data(1, 3),
        fetch_data(2, 2),
        fetch_data(3, 1)
    )

    print("-" * 30)
    print(f"收到所有数据: {results}")
    print(f"总耗时: {time.time() - start:.2f} 秒")


if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import time


async def fetch_data(id, delay):
    print(f"【📥 开始】下载任务 {id}...")
    await asyncio.sleep(delay)  # 模拟网络请求耗时
    print(f"【✅ 完成】任务 {id} 结束")
    return f"数据 {id}"

async def main():
    # TaskGroup 会在 with 块结束时，自动确保里面的所有任务都执行完毕
    async with asyncio.TaskGroup() as tg:
        # tg.create_task 会让任务立刻在后台异步跑起来
        task1 = tg.create_task(fetch_data(1, 2))
        task2 = tg.create_task(fetch_data(2, 1))

    # 当走出 with 块后，可以通过 .result() 直接拿到结果
    print(task1.result(), task2.result())

if __name__ == "__main__":
    asyncio.run(main())

import time
from greenlet import greenlet


def a():  # 任务A
    for i in range(5):
        print('A' + str(i))
        gb.switch()
        time.sleep(0.1)


def b():  # 任务B
    for i in range(5):
        print('B' + str(i))
        gc.switch()
        time.sleep(0.1)


def c():  # 任务C
    for i in range(5):
        print('C' + str(i))
        ga.switch()
        time.sleep(0.1)


if __name__ == '__main__':
    ga = greenlet(a)
    gb = greenlet(b)
    gc = greenlet(c)

    ga.switch()

import time
import gevent
from gevent import monkey
 
# 有耗时操作时需要
monkey.patch_all()  # 打补丁，将程序中用到的耗时操作的代码，换为gevent中自己实现的模块
 
 
def a():  # 任务A
    for i in range(5):
        print('A' + str(i))
        time.sleep(0.1)
 
 
def b():  # 任务B
    for i in range(5):
        print('B' + str(i))
        time.sleep(0.1)
 
 
def c():  # 任务C
    for i in range(5):
        print('C' + str(i))
        time.sleep(0.1)
 
 
if __name__ == '__main__':
    g1 = gevent.spawn(a)
    g2 = gevent.spawn(b)
    g3 = gevent.spawn(c)
 
    g1.join()
    g2.join()
    g3.join()

import requests
import gevent
from gevent import monkey

monkey.patch_all()


def download(url):
    response = requests.get(url)  # 耗时操作
    content = response.text
    print('下载了{}的数据，长度：{}'.format(url, len(content)))


if __name__ == '__main__':
    urls = ['http://www.163.com', 'http://www.baidu.com']
    g1 = gevent.spawn(download, urls[0])
    g2 = gevent.spawn(download, urls[1])

    g1.join()
    g2.join()

import gevent
from gevent import monkey
monkey.patch_all()

import requests


def download(url):
    response = requests.get(url)  # 耗时操作
    content = response.text
    print('下载了{}的数据，长度：{}'.format(url, len(content)))


if __name__ == '__main__':
    urls = ['http://www.163.com', 'http://www.baidu.com']
    g1 = gevent.spawn(download, urls[0])
    g2 = gevent.spawn(download, urls[1])

    g1.join()
    g2.join()

    # gevent.joinall(g1, g2)

import threading
import time

lock1 = threading.Lock()
lock2 = threading.Lock()

def thread_A():
    with lock1:
        print("线程 A: 已获取锁 1，正在等待锁 2...")
        time.sleep(0.1)  # 迫使线程 B 有时间获取锁 2
        with lock2:
            print("线程 A: 成功获取锁 2！")

def thread_B():
    with lock2:
        print("线程 B: 已获取锁 2，正在等待锁 1...")
        time.sleep(0.1)
        with lock1:
            print("线程 B: 成功获取锁 1！")

# 启动后程序会永久卡死
threading.Thread(target=thread_A).start()
threading.Thread(target=thread_B).start()

import threading
import time

lock1 = threading.Lock()
lock2 = threading.Lock()

def thread_A():
    with lock1:
        print("线程 A: 已获取锁 1，正在等待锁 2...")
        time.sleep(0.1)  # 迫使线程 B 有时间获取锁 2
        with lock2:
            print("线程 A: 成功获取锁 2！")

# 修复线程 B 的顺序，使其与 A 一致
def thread_B_fixed():
    with lock1:  # 先拿锁 1
        print("线程 B: 已获取锁 1，正在等待锁 2...")
        with lock2:
            print("线程 B: 成功获取锁 2！")

# 启动后程序会永久卡死
threading.Thread(target=thread_A).start()
threading.Thread(target=thread_B_fixed).start()

import threading

lock = threading.Lock()

def recursive_function(count):
    with lock:  # 第一层获取成功
        print(f"进入层级: {count}")
        if count > 0:
            # 递归再次尝试获取同一个 lock，但它已经被自己锁住了！
            recursive_function(count - 1) 

recursive_function(2)

import threading

# 修复：只需将 Lock 换成 RLock
lock = threading.RLock()

def recursive_function(count):
    with lock:  # 第一层获取成功
        print(f"进入层级: {count}")
        if count > 0:
            # 递归再次尝试获取同一个 lock，但它已经被自己锁住了！
            recursive_function(count - 1)

recursive_function(2)

import asyncio

async_lock1 = asyncio.Lock()
async_lock2 = asyncio.Lock()

async def task_A():
    async with async_lock1:
        print("任务 A 拿到了锁 1")
        await asyncio.sleep(0.1)
        async with async_lock2:  # 等待任务 B 释放锁 2
            print("任务 A 拿到了锁 2")

async def task_B():
    async with async_lock2:
        print("任务 B 拿到了锁 2")
        await asyncio.sleep(0.1)
        async with async_lock1:  # 等待任务 A 释放锁 1
            print("任务 B 拿到了锁 1")

async def main():
    await asyncio.gather(task_A(), task_B())

asyncio.run(main())

try:
    # 如果 2 秒内拿不到锁，抛出 TimeoutError，打破死锁
    await asyncio.wait_for(async_lock2.acquire(), timeout=2.0)
    # 执行业务逻辑...
finally:
    if async_lock2.locked():
        async_lock2.release()

from multiprocessing import Process, Queue

def worker(q):
    # 写入非常大的数据，塞满队列缓冲区
    q.put("X" * 1000000) 
    print("子进程写入完毕")

if __name__ == "__main__":
    q = Queue()
    p = Process(target=worker, args=(q,))
    p.start()
    
    # ❌ 错误：在读取数据前先 join()。
    # 此时子进程卡在 put() 等待缓冲区释放，而主进程卡在 join() 等待子进程结束。
    p.join() 
    
    data = q.get()
    print("主进程获取数据")

from multiprocessing import Process, Queue


def worker(q):
    # 写入非常大的数据，塞满队列缓冲区
    q.put("X" * 1000000)
    print("子进程写入完毕")


if __name__ == "__main__":
    q = Queue()
    p = Process(target=worker, args=(q,))
    p.start()
    data = q.get()
    p.join()

    print("主进程获取数据")

from multiprocessing import Process, Queue

def worker(q):
    # 子进程放入数据
    q.put({"result": "success", "data": [1, 2, 3]})

if __name__ == "__main__":
    q = Queue()
    p = Process(target=worker, args=(q,))
    p.start()
    
    # 主进程获取数据
    data = q.get()
    print(f"主进程收到: {data}")
    p.join()

from multiprocessing import Process, Pipe

def worker(conn):
    conn.send("来自子进程的问候")
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = Pipe()
    p = Process(target=worker, args=(child_conn,))
    p.start()
    
    print(parent_conn.recv()) # 主进程接收
    p.join()

import threading
import queue
import time

q = queue.Queue()


def producer():
    for i in range(3):
        print(f"生产了:产品 {i}")
        q.put(f"产品 {i}")
        time.sleep(0.5)


def consumer():
    while True:
        item = q.get()  # 如果队列为空，会自动阻塞等待
        if item is None:
            break
        print(f"消费了: {item}")
        q.task_done()


# 启动线程
t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t1.start()
t2.start()

t1.join()
q.put(None)  # 发送结束信号通知消费者退出
t2.join()

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock: # 确保同一时间只有一个线程修改 counter
            counter += 1

import asyncio

async def producer(q):
    for i in range(3):
        await q.put(f"消息 {i}")
        await asyncio.sleep(0.1)
    
    # 🌟 生产结束，放入一个 None 告知消费者可以下班了
    await q.put(None)

async def consumer(q):
    while True:
        item = await q.get()
        
        # 🌟 如果收到 None，说明后面没消息了，直接退出
        if item is None:
            q.task_done()
            break
            
        print(f"处理协程消息: {item}")
        q.task_done()

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())

import asyncio
import threading
import time

async def async_worker(q):
    """在异步事件循环中运行的消费者"""
    while True:
        item = await q.get()
        print(f"异步循环收到来自普通线程的数据: {item}")
        if item == "STOP":
            break

def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

if __name__ == "__main__":
    # 1. 在单独的子线程中启动 asyncio 事件循环
    new_loop = asyncio.new_event_loop()
    t = threading.Thread(target=start_loop, args=(new_loop,), daemon=True)
    t.start()

    # 2. 在该循环中初始化一个异步队列
    async_queue = asyncio.run_coroutine_threadsafe(asyncio.Queue(), new_loop).result()
    
    # 3. 在异步循环中启动消费者任务
    asyncio.run_coroutine_threadsafe(async_worker(async_queue), new_loop)

    # 4. 主线程（普通同步线程）向异步队列推送数据
    for i in range(3):
        time.sleep(0.5)
        # 核心：跨线程向异步队列中 put 数据
        asyncio.run_coroutine_threadsafe(async_queue.put(f"数据-{i}"), new_loop)

    # 发送停止信号
    asyncio.run_coroutine_threadsafe(async_queue.put("STOP"), new_loop)
    time.sleep(1)