1. 为什么需要关注连接对象的线程安全在Python多线程环境中处理连接对象时最容易被忽视却最致命的问题就是线程安全。我曾在实际项目中遇到过这样的场景一个看似运行良好的多线程数据库应用在线上环境运行几天后突然开始出现数据错乱和连接泄漏。经过长达72小时的排查最终发现问题出在多个线程共享同一个未加保护的数据库连接上。连接对象通常指那些与外部资源建立通信通道的对象比如数据库连接MySQL、PostgreSQL等网络套接字连接HTTP长连接文件系统句柄这些对象的特点是创建成本高TCP三次握手、认证流程等状态保持可能有事务状态、序列号等非原子操作查询-响应模式在多线程环境下如果多个线程同时操作同一个连接对象可能会引发数据交叉污染线程A的查询结果被线程B的响应覆盖协议状态混乱如HTTP的pipelining乱序资源泄漏连接未被正确关闭死锁线程间互相等待连接释放关键认知Python的GIL只保证字节码执行的原子性不保证你的连接对象操作是原子的。即使是在CPython中一个简单的conn.execute()也可能被GIL切换打断。2. 连接对象的线程安全等级划分不是所有连接对象都同样危险。根据我的经验可以将其分为三类2.1 完全非线程安全型典型代表SQLite连接特别是启用了WAL模式时某些NoSQL驱动的基础连接低级别的socket连接特征内部没有任何锁机制并发操作直接导致段错误或数据损坏必须由调用方完全控制访问2.2 条件线程安全型典型代表MySQL Connector/Pythonpsycopg2PostgreSQLRedis-py的基础连接特征单个方法调用是安全的但跨方法操作需要外部同步例如开始事务-执行查询-提交 这个序列需要加锁2.3 自维护线程安全型典型代表SQLAlchemy的连接池HTTPX的异步连接某些ORM的高级封装特征内部实现了连接复用策略对外暴露线程安全接口可能带来性能损耗判断方法实操技巧import inspect from threading import Lock def check_thread_safety(conn): 检查连接对象的线程安全特征 has_lock any( isinstance(getattr(conn, attr, None), Lock) for attr in dir(conn) ) methods inspect.getmembers(conn, inspect.ismethod) has_sync_decorators any( hasattr(m[1], __sync_decorator__) for m in methods ) return has_lock or has_sync_decorators3. 实战中的五种防护策略3.1 连接独占模式适合场景短生命周期线程连接使用时间极短实现方案class DedicatedConnection: def __init__(self, conn_factory): self._factory conn_factory self._local threading.local() def __enter__(self): if not hasattr(self._local, conn): self._local.conn self._factory() return self._local.conn def __exit__(self, *args): if hasattr(self._local, conn): self._local.conn.close() del self._local.conn # 使用示例 with DedicatedConnection(lambda: create_db_conn()) as conn: conn.execute(SELECT ...)优点每个线程获得独立连接自动清理资源 缺点连接数线程数可能耗尽资源3.2 带锁的共享连接适合场景长连接且连接创建成本极高实现方案class SharedConnectionWithLock: def __init__(self, conn): self._conn conn self._lock threading.RLock() def execute(self, query): with self._lock: cursor self._conn.cursor() try: cursor.execute(query) return cursor.fetchall() finally: cursor.close() # 使用示例 shared_conn SharedConnectionWithLock(create_expensive_conn()) result shared_conn.execute(SELECT ...)关键细节使用RLock允许同一线程重入确保cursor被正确关闭锁粒度控制在整个操作序列3.3 连接池模式适合场景大多数数据库应用最佳实践from queue import Queue class ConnectionPool: def __init__(self, size, factory): self._pool Queue(maxsizesize) for _ in range(size): self._pool.put(factory()) def get_conn(self): return self._pool.get() def release_conn(self, conn): self._pool.put(conn) # 使用示例 pool ConnectionPool(5, create_db_conn) conn pool.get_conn() try: conn.execute(...) finally: pool.release_conn(conn)性能调优点池大小应略大于平均并发线程数添加健康检查机制考虑引入超时回收3.4 代理模式引用计数适合场景需要精细控制连接生命周期高级实现from weakref import WeakKeyDictionary class ConnectionProxy: def __init__(self, factory): self._factory factory self._refcount WeakKeyDictionary() self._lock threading.Lock() self._real_conn None property def conn(self): thread threading.current_thread() with self._lock: if self._real_conn is None: self._real_conn self._factory() self._refcount[thread] self._refcount.get(thread, 0) 1 return self._real_conn def release(self): thread threading.current_thread() with self._lock: if thread in self._refcount: self._refcount[thread] - 1 if self._refcount[thread] 0: del self._refcount[thread] if not self._refcount: self._real_conn.close() self._real_conn None3.5 协程适配器适合场景混合使用线程和协程创新方案import asyncio from functools import partial class CoroutineAdapter: def __init__(self, sync_conn): self._sync_conn sync_conn self._loop asyncio.get_event_loop() self._lock asyncio.Lock() async def execute(self, query): async with self._lock: return await self._loop.run_in_executor( None, partial(self._sync_execute, query) ) def _sync_execute(self, query): # 在同步上下文中执行 cursor self._sync_conn.cursor() try: cursor.execute(query) return cursor.fetchall() finally: cursor.close()4. 常见陷阱与诊断方法4.1 幽灵连接问题症状连接数缓慢增长直至耗尽无明确的内存泄漏诊断步骤使用lsof -p pid查看实际连接状态在连接对象上添加finalizer日志import weakref def log_cleanup(conn): print(fConnection {id(conn)} finalized) conn create_conn() weakref.finalize(conn, log_cleanup, conn)4.2 交叉响应问题症状查询结果与请求不匹配随机出现数据错乱复现方法def race_condition_test(): conn create_shared_conn() results [] def worker(query): results.append((query, conn.execute(query))) threads [ threading.Thread(targetworker, args(fSELECT {i},)) for i in range(10) ] for t in threads: t.start() for t in threads: t.join() for query, result in results: print(f{query} {result}) # 观察不匹配情况4.3 死锁场景典型死锁链线程A持有连接锁等待数据锁线程B持有数据锁等待连接锁调试技巧使用faulthandler.dump_traceback_later(5)定期输出堆栈在锁获取时添加日志import time class DebugLock: def __init__(self): self._lock threading.Lock() self._holder None def acquire(self): print(fThread {threading.get_ident()} waiting at {time.time()}) self._lock.acquire() self._holder threading.get_ident() print(fThread {self._holder} acquired at {time.time()}) def release(self): print(fThread {self._holder} releasing at {time.time()}) self._lock.release()5. 性能优化进阶技巧5.1 锁粒度优化错误示范# 粗粒度锁性能差 class BadExample: def __init__(self, conn): self._conn conn self._lock threading.Lock() def operation(self): with self._lock: # 整个方法加锁 self._do_setup() result self._do_query() self._do_cleanup() return result优化方案# 细粒度锁 class OptimizedExample: def __init__(self, conn): self._conn conn self._query_lock threading.Lock() self._setup_lock threading.Lock() def operation(self): # 非关键路径不加锁 self._do_setup() # 仅保护核心操作 with self._query_lock: result self._do_query() self._do_cleanup() return result5.2 无锁设计模式适用场景读多写少实现方案import copy class LockFreeConnection: def __init__(self, conn): self._conn conn self._snapshot None self._version 0 self._write_lock threading.Lock() def read(self): if self._snapshot is None: with self._write_lock: self._snapshot copy.deepcopy(self._conn) self._version 1 return self._snapshot def write(self, operation): with self._write_lock: operation(self._conn) self._snapshot None # 使读取端重新快照5.3 基于CPU缓存的优化现代CPU特性利用from ctypes import c_long, Structure from threading import Thread class PaddedCounter(Structure): _fields_ [ (value, c_long), (_pad1, c_long * 7), # 缓存行填充(通常64字节/行) (lock, c_long), (_pad2, c_long * 7) ] def __init__(self): self.value 0 self.lock 0 def increment(self): while True: # 使用CAS原子操作 current self.value if self._compare_and_swap(0, 1): try: self.value 1 finally: self.lock 0 break def _compare_and_swap(self, expected, new): # 模拟原子CAS操作 if self.lock expected: self.lock new return True return False6. 行业标准方案对比6.1 数据库连接池方案方案线程安全机制适用场景性能影响SQLAlchemy连接池线程本地存储ORM应用中等Django DB Pool每个请求独立连接Web应用较低HikariCP无锁队列CAS操作高性能Java/Python应用极小psycopg2.pool简单的锁保护小型PostgreSQL应用中等6.2 网络连接管理库/框架线程模型连接复用策略特殊考虑requests连接池锁基于主机/端口需要手动session管理aiohttp异步单线程连接器抽象协程安全但非线程安全urllib3连接池细粒度锁支持keep-alive需要正确释放连接grpc多路复用流控基于HTTP/2复杂的状态管理6.3 文件系统处理方法线程安全级别推荐用法风险点直接open()不安全线程独占文件可能损坏数据fcntl锁进程级安全Unix系统跨进程不解决线程安全问题代理模式完全控制关键文件操作实现复杂度高队列串行化安全但性能低低频写入场景可能成为瓶颈7. 测试策略与验证方法7.1 确定性重现测试构建确定性竞争条件import random def deterministic_test(): conn create_conn() results [] barrier threading.Barrier(2) def worker1(): barrier.wait() conn.execute(BEGIN) conn.execute(INSERT ...) results.append(conn.execute(SELECT ...)) def worker2(): barrier.wait() conn.execute(BEGIN) conn.execute(INSERT ...) results.append(conn.execute(SELECT ...)) threads [ threading.Thread(targetworker1), threading.Thread(targetworker2) ] for t in threads: t.start() for t in threads: t.join() assert len(set(results)) 2 # 应该有两个不同结果7.2 模糊测试随机压力测试def fuzz_test(): from concurrent.futures import ThreadPoolExecutor conn_pool ConnectionPool(10, create_conn) def random_operation(_): with conn_pool.get_conn() as conn: time.sleep(random.random() * 0.1) conn.execute(random.choice(queries)) with ThreadPoolExecutor(max_workers50) as executor: for _ in range(1000): executor.submit(random_operation, None) assert conn_pool.active_count 0 # 检查连接泄漏7.3 静态分析使用mypy检测潜在问题# mypy: warn_unused_ignoresFalse from typing import ContextManager def check_connection_usage(conn: ContextManager) - None: with conn as c: # mypy会检查是否正确使用上下文 c.execute(...)8. 现代Python的最佳实践8.1 使用contextvarsPython 3.7新特性from contextvars import ContextVar conn_var ContextVar(database_connection) class ContextAwareConnection: def __init__(self, factory): self._factory factory def __enter__(self): try: return conn_var.get() except LookupError: conn self._factory() conn_var.set(conn) return conn def __exit__(self, *args): conn conn_var.get() conn.close() conn_var.reset()8.2 结构化并发使用Python 3.11的TaskGroupasync def structured_concurrency(): async with asyncio.TaskGroup() as tg: conn create_async_conn() for i in range(10): tg.create_task(worker(conn, i)) # 自动处理取消和异常8.3 类型注解增强使用PEP 484和PEP 589规范from typing import Protocol, runtime_checkable runtime_checkable class ThreadSafeConnection(Protocol): def execute(self, query: str) - list[tuple]: ... def close(self) - None: ... property def closed(self) - bool: ... def verify_connection(conn: ThreadSafeConnection): if not isinstance(conn, ThreadSafeConnection): raise TypeError(Connection does not meet thread safety protocol)9. 从设计模式角度思考9.1 资源租约模式from datetime import datetime, timedelta class ConnectionLease: def __init__(self, conn, ttl30): self._conn conn self._expire datetime.now() timedelta(secondsttl) self._lock threading.Lock() def is_valid(self): with self._lock: return datetime.now() self._expire def renew(self, ttl30): with self._lock: self._expire datetime.now() timedelta(secondsttl) property def conn(self): if not self.is_valid(): raise ValueError(Lease expired) return self._conn9.2 代理装饰器组合def synchronized_connection(cls): original_execute cls.execute def wrapped_execute(self, query): with self._lock: return original_execute(self, query) cls.execute wrapped_execute return cls synchronized_connection class SafeDBConnection(DBConnection): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._lock threading.RLock()9.3 反应式扩展from rx import operators as ops from rx.scheduler import ThreadPoolScheduler thread_scheduler ThreadPoolScheduler(10) def reactive_operations(): rx.from_iterable(queries).pipe( ops.flat_map(lambda q: rx.from_callable( lambda: execute_query(q), schedulerthread_scheduler )), ops.buffer_with_count(10), ops.subscribe_on(thread_scheduler) ).subscribe(handle_results)
1. 为什么需要关注连接对象的线程安全在Python多线程环境中处理连接对象时最容易被忽视却最致命的问题就是线程安全。我曾在实际项目中遇到过这样的场景一个看似运行良好的多线程数据库应用在线上环境运行几天后突然开始出现数据错乱和连接泄漏。经过长达72小时的排查最终发现问题出在多个线程共享同一个未加保护的数据库连接上。连接对象通常指那些与外部资源建立通信通道的对象比如数据库连接MySQL、PostgreSQL等网络套接字连接HTTP长连接文件系统句柄这些对象的特点是创建成本高TCP三次握手、认证流程等状态保持可能有事务状态、序列号等非原子操作查询-响应模式在多线程环境下如果多个线程同时操作同一个连接对象可能会引发数据交叉污染线程A的查询结果被线程B的响应覆盖协议状态混乱如HTTP的pipelining乱序资源泄漏连接未被正确关闭死锁线程间互相等待连接释放关键认知Python的GIL只保证字节码执行的原子性不保证你的连接对象操作是原子的。即使是在CPython中一个简单的conn.execute()也可能被GIL切换打断。2. 连接对象的线程安全等级划分不是所有连接对象都同样危险。根据我的经验可以将其分为三类2.1 完全非线程安全型典型代表SQLite连接特别是启用了WAL模式时某些NoSQL驱动的基础连接低级别的socket连接特征内部没有任何锁机制并发操作直接导致段错误或数据损坏必须由调用方完全控制访问2.2 条件线程安全型典型代表MySQL Connector/Pythonpsycopg2PostgreSQLRedis-py的基础连接特征单个方法调用是安全的但跨方法操作需要外部同步例如开始事务-执行查询-提交 这个序列需要加锁2.3 自维护线程安全型典型代表SQLAlchemy的连接池HTTPX的异步连接某些ORM的高级封装特征内部实现了连接复用策略对外暴露线程安全接口可能带来性能损耗判断方法实操技巧import inspect from threading import Lock def check_thread_safety(conn): 检查连接对象的线程安全特征 has_lock any( isinstance(getattr(conn, attr, None), Lock) for attr in dir(conn) ) methods inspect.getmembers(conn, inspect.ismethod) has_sync_decorators any( hasattr(m[1], __sync_decorator__) for m in methods ) return has_lock or has_sync_decorators3. 实战中的五种防护策略3.1 连接独占模式适合场景短生命周期线程连接使用时间极短实现方案class DedicatedConnection: def __init__(self, conn_factory): self._factory conn_factory self._local threading.local() def __enter__(self): if not hasattr(self._local, conn): self._local.conn self._factory() return self._local.conn def __exit__(self, *args): if hasattr(self._local, conn): self._local.conn.close() del self._local.conn # 使用示例 with DedicatedConnection(lambda: create_db_conn()) as conn: conn.execute(SELECT ...)优点每个线程获得独立连接自动清理资源 缺点连接数线程数可能耗尽资源3.2 带锁的共享连接适合场景长连接且连接创建成本极高实现方案class SharedConnectionWithLock: def __init__(self, conn): self._conn conn self._lock threading.RLock() def execute(self, query): with self._lock: cursor self._conn.cursor() try: cursor.execute(query) return cursor.fetchall() finally: cursor.close() # 使用示例 shared_conn SharedConnectionWithLock(create_expensive_conn()) result shared_conn.execute(SELECT ...)关键细节使用RLock允许同一线程重入确保cursor被正确关闭锁粒度控制在整个操作序列3.3 连接池模式适合场景大多数数据库应用最佳实践from queue import Queue class ConnectionPool: def __init__(self, size, factory): self._pool Queue(maxsizesize) for _ in range(size): self._pool.put(factory()) def get_conn(self): return self._pool.get() def release_conn(self, conn): self._pool.put(conn) # 使用示例 pool ConnectionPool(5, create_db_conn) conn pool.get_conn() try: conn.execute(...) finally: pool.release_conn(conn)性能调优点池大小应略大于平均并发线程数添加健康检查机制考虑引入超时回收3.4 代理模式引用计数适合场景需要精细控制连接生命周期高级实现from weakref import WeakKeyDictionary class ConnectionProxy: def __init__(self, factory): self._factory factory self._refcount WeakKeyDictionary() self._lock threading.Lock() self._real_conn None property def conn(self): thread threading.current_thread() with self._lock: if self._real_conn is None: self._real_conn self._factory() self._refcount[thread] self._refcount.get(thread, 0) 1 return self._real_conn def release(self): thread threading.current_thread() with self._lock: if thread in self._refcount: self._refcount[thread] - 1 if self._refcount[thread] 0: del self._refcount[thread] if not self._refcount: self._real_conn.close() self._real_conn None3.5 协程适配器适合场景混合使用线程和协程创新方案import asyncio from functools import partial class CoroutineAdapter: def __init__(self, sync_conn): self._sync_conn sync_conn self._loop asyncio.get_event_loop() self._lock asyncio.Lock() async def execute(self, query): async with self._lock: return await self._loop.run_in_executor( None, partial(self._sync_execute, query) ) def _sync_execute(self, query): # 在同步上下文中执行 cursor self._sync_conn.cursor() try: cursor.execute(query) return cursor.fetchall() finally: cursor.close()4. 常见陷阱与诊断方法4.1 幽灵连接问题症状连接数缓慢增长直至耗尽无明确的内存泄漏诊断步骤使用lsof -p pid查看实际连接状态在连接对象上添加finalizer日志import weakref def log_cleanup(conn): print(fConnection {id(conn)} finalized) conn create_conn() weakref.finalize(conn, log_cleanup, conn)4.2 交叉响应问题症状查询结果与请求不匹配随机出现数据错乱复现方法def race_condition_test(): conn create_shared_conn() results [] def worker(query): results.append((query, conn.execute(query))) threads [ threading.Thread(targetworker, args(fSELECT {i},)) for i in range(10) ] for t in threads: t.start() for t in threads: t.join() for query, result in results: print(f{query} {result}) # 观察不匹配情况4.3 死锁场景典型死锁链线程A持有连接锁等待数据锁线程B持有数据锁等待连接锁调试技巧使用faulthandler.dump_traceback_later(5)定期输出堆栈在锁获取时添加日志import time class DebugLock: def __init__(self): self._lock threading.Lock() self._holder None def acquire(self): print(fThread {threading.get_ident()} waiting at {time.time()}) self._lock.acquire() self._holder threading.get_ident() print(fThread {self._holder} acquired at {time.time()}) def release(self): print(fThread {self._holder} releasing at {time.time()}) self._lock.release()5. 性能优化进阶技巧5.1 锁粒度优化错误示范# 粗粒度锁性能差 class BadExample: def __init__(self, conn): self._conn conn self._lock threading.Lock() def operation(self): with self._lock: # 整个方法加锁 self._do_setup() result self._do_query() self._do_cleanup() return result优化方案# 细粒度锁 class OptimizedExample: def __init__(self, conn): self._conn conn self._query_lock threading.Lock() self._setup_lock threading.Lock() def operation(self): # 非关键路径不加锁 self._do_setup() # 仅保护核心操作 with self._query_lock: result self._do_query() self._do_cleanup() return result5.2 无锁设计模式适用场景读多写少实现方案import copy class LockFreeConnection: def __init__(self, conn): self._conn conn self._snapshot None self._version 0 self._write_lock threading.Lock() def read(self): if self._snapshot is None: with self._write_lock: self._snapshot copy.deepcopy(self._conn) self._version 1 return self._snapshot def write(self, operation): with self._write_lock: operation(self._conn) self._snapshot None # 使读取端重新快照5.3 基于CPU缓存的优化现代CPU特性利用from ctypes import c_long, Structure from threading import Thread class PaddedCounter(Structure): _fields_ [ (value, c_long), (_pad1, c_long * 7), # 缓存行填充(通常64字节/行) (lock, c_long), (_pad2, c_long * 7) ] def __init__(self): self.value 0 self.lock 0 def increment(self): while True: # 使用CAS原子操作 current self.value if self._compare_and_swap(0, 1): try: self.value 1 finally: self.lock 0 break def _compare_and_swap(self, expected, new): # 模拟原子CAS操作 if self.lock expected: self.lock new return True return False6. 行业标准方案对比6.1 数据库连接池方案方案线程安全机制适用场景性能影响SQLAlchemy连接池线程本地存储ORM应用中等Django DB Pool每个请求独立连接Web应用较低HikariCP无锁队列CAS操作高性能Java/Python应用极小psycopg2.pool简单的锁保护小型PostgreSQL应用中等6.2 网络连接管理库/框架线程模型连接复用策略特殊考虑requests连接池锁基于主机/端口需要手动session管理aiohttp异步单线程连接器抽象协程安全但非线程安全urllib3连接池细粒度锁支持keep-alive需要正确释放连接grpc多路复用流控基于HTTP/2复杂的状态管理6.3 文件系统处理方法线程安全级别推荐用法风险点直接open()不安全线程独占文件可能损坏数据fcntl锁进程级安全Unix系统跨进程不解决线程安全问题代理模式完全控制关键文件操作实现复杂度高队列串行化安全但性能低低频写入场景可能成为瓶颈7. 测试策略与验证方法7.1 确定性重现测试构建确定性竞争条件import random def deterministic_test(): conn create_conn() results [] barrier threading.Barrier(2) def worker1(): barrier.wait() conn.execute(BEGIN) conn.execute(INSERT ...) results.append(conn.execute(SELECT ...)) def worker2(): barrier.wait() conn.execute(BEGIN) conn.execute(INSERT ...) results.append(conn.execute(SELECT ...)) threads [ threading.Thread(targetworker1), threading.Thread(targetworker2) ] for t in threads: t.start() for t in threads: t.join() assert len(set(results)) 2 # 应该有两个不同结果7.2 模糊测试随机压力测试def fuzz_test(): from concurrent.futures import ThreadPoolExecutor conn_pool ConnectionPool(10, create_conn) def random_operation(_): with conn_pool.get_conn() as conn: time.sleep(random.random() * 0.1) conn.execute(random.choice(queries)) with ThreadPoolExecutor(max_workers50) as executor: for _ in range(1000): executor.submit(random_operation, None) assert conn_pool.active_count 0 # 检查连接泄漏7.3 静态分析使用mypy检测潜在问题# mypy: warn_unused_ignoresFalse from typing import ContextManager def check_connection_usage(conn: ContextManager) - None: with conn as c: # mypy会检查是否正确使用上下文 c.execute(...)8. 现代Python的最佳实践8.1 使用contextvarsPython 3.7新特性from contextvars import ContextVar conn_var ContextVar(database_connection) class ContextAwareConnection: def __init__(self, factory): self._factory factory def __enter__(self): try: return conn_var.get() except LookupError: conn self._factory() conn_var.set(conn) return conn def __exit__(self, *args): conn conn_var.get() conn.close() conn_var.reset()8.2 结构化并发使用Python 3.11的TaskGroupasync def structured_concurrency(): async with asyncio.TaskGroup() as tg: conn create_async_conn() for i in range(10): tg.create_task(worker(conn, i)) # 自动处理取消和异常8.3 类型注解增强使用PEP 484和PEP 589规范from typing import Protocol, runtime_checkable runtime_checkable class ThreadSafeConnection(Protocol): def execute(self, query: str) - list[tuple]: ... def close(self) - None: ... property def closed(self) - bool: ... def verify_connection(conn: ThreadSafeConnection): if not isinstance(conn, ThreadSafeConnection): raise TypeError(Connection does not meet thread safety protocol)9. 从设计模式角度思考9.1 资源租约模式from datetime import datetime, timedelta class ConnectionLease: def __init__(self, conn, ttl30): self._conn conn self._expire datetime.now() timedelta(secondsttl) self._lock threading.Lock() def is_valid(self): with self._lock: return datetime.now() self._expire def renew(self, ttl30): with self._lock: self._expire datetime.now() timedelta(secondsttl) property def conn(self): if not self.is_valid(): raise ValueError(Lease expired) return self._conn9.2 代理装饰器组合def synchronized_connection(cls): original_execute cls.execute def wrapped_execute(self, query): with self._lock: return original_execute(self, query) cls.execute wrapped_execute return cls synchronized_connection class SafeDBConnection(DBConnection): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._lock threading.RLock()9.3 反应式扩展from rx import operators as ops from rx.scheduler import ThreadPoolScheduler thread_scheduler ThreadPoolScheduler(10) def reactive_operations(): rx.from_iterable(queries).pipe( ops.flat_map(lambda q: rx.from_callable( lambda: execute_query(q), schedulerthread_scheduler )), ops.buffer_with_count(10), ops.subscribe_on(thread_scheduler) ).subscribe(handle_results)
