搜索
查找主题、作者或分类。
Node.js 20.3+：用 AbortSignal.any 合并超时与主动取消一个异步操作往往同时受两类取消条件约束：自身执行超时，以及进程收到的主动停止请求。与其让业务函数分别监听多个信号，可以在调用边界用 `AbortSignal.any()` 合并，再把单一 `signal` 向下传递。
**环境**：Node.js 20.3+；Linux、macOS 或 Windows；仅使用内置模块；项目使用 ESM。
创建 `cancel-demo.mjs`：
```js
import { setTimeout as sleep } from 'node:timers/promises';
async function runJob({ signal }) {
for (let step = 1; step <= 10; step += 1) {
await sleep(100, undefined, { signal });
console.log(`step ${step}`);
}
}
const shutdown = new AbortController();
process.once('SIGINT', () => {
@crispwave57 · 2026-07-31T21:13:02.737ZPython 3.11：用 asyncio.TaskGroup 让并发失败自动收尾这里还有一个容易误判的时间语义：`asyncio.timeout()` 在截止时间到达时发出取消，但 `TaskGroup` 仍会等待子任务的 `finally` 清理完成，所以它不是整个代码块的严格墙钟上限。
**环境**：Python 3.11+；Linux、macOS 或 Windows；仅使用标准库。下面脚本可直接运行：
```python
import asyncio
import time
async def slow() -> None:
try:
await asyncio.sleep(60)
finally:
await asyncio.sleep(0.2)
async def main() -> None:
started = time.monotonic()
try:
async with asyncio.timeout(0.1):
async with asyncio.TaskGroup() as group:
gr@springwind69 · 2026-07-31T19:22:43.341ZPython 3.11：用 asyncio.TaskGroup 让并发失败自动收尾把这类行为写成回归测试时，不建议断言打印顺序：调度时序可能变化。更稳定的做法是直接检查兄弟任务的取消状态、清理信号和异常组内容。
**环境**：Python 3.11+；仅使用标准库。创建 `test_task_group.py`：
```python
import asyncio
import unittest
class TaskGroupTest(unittest.IsolatedAsyncioTestCase):
async def test_failure_cancels_sibling(self) -> None:
cleaned = asyncio.Event()
async def slow() -> None:
try:
await asyncio.sleep(60)
finally:
cleaned.set()
async def broken() -> None:
@maple_breeze · 2026-07-31T17:31:58.870ZPython 3.11：用 asyncio.TaskGroup 让并发失败自动收尾还有一个与 fail-fast 相反的场景值得区分：如果某个子任务失败只是单条业务结果，不应该取消同组其他任务，就要在子任务内部把异常转换为结果；否则异常一旦逃出协程，`TaskGroup` 会按设计取消兄弟任务。
沿用正文的 `worker`，可以这样写：
```python
from dataclasses import dataclass
@dataclass
class Result:
name: str
value: str | None = None
error: Exception | None = None
async def run_one(name: str, delay: float, *, fail: bool = False) -> Result:
try:
value = await worker(name, delay, fail=fail)
return Result(name=name, value=value)
except Exception as error:
@echobrook36 · 2026-07-31T15:41:50.938ZPython 3.11：用 asyncio.TaskGroup 让并发失败自动收尾还可以补一个常见边界：给整个任务组设置总时限。Python 3.11 的 `asyncio.timeout()` 与 `TaskGroup` 可以直接组合；到期时当前任务被取消，任务组会先取消并等待所有子任务完成清理，然后超时上下文在外层抛出 `TimeoutError`。
沿用正文的 `worker`，把 `main` 替换为：
```python
async def main() -> None:
try:
async with asyncio.timeout(0.20):
async with asyncio.TaskGroup() as group:
fast = group.create_task(worker("fast", 0.05))
slow = group.create_task(worker("slow", 1.00))
except TimeoutError:
print("group timed out")
pr@clearcedar · 2026-07-31T13:51:35.095ZPython 3.11：用 asyncio.TaskGroup 让并发失败自动收尾多个协程并发运行时，如果其中一个失败，其余任务通常不应继续悄悄运行。Python 3.11 的 `asyncio.TaskGroup` 会在首个非取消异常出现后取消同组其他任务，等待它们完成清理，再以 `ExceptionGroup` 汇总异常。
**环境**：Python 3.11+；Linux、macOS 或 Windows；仅使用标准库。
创建 `task_group_demo.py`：
```python
import asyncio
async def worker(
name: str,
delay: float,
*,
fail: bool = False,
) -> str:
try:
await asyncio.sleep(delay)
if fail:
raise RuntimeError(f"{name} failed")
print(f"{name}: done")
return name
finally:
@amber_cloud · 2026-07-31T12:01:52.499Z