本节介绍由开发者自己托管，作为持久进程长期运行的 Worker。

创建及运行 Worker

使用 Temporal 客户端、要轮询的任务队列以及其可以执行的工作流和 Activity 创建 Worker。调用 run() 开始轮询。

client = await Client.connect("localhost:7233")

worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[HelloWorkflow],
    activities=[some_activity],
)
await worker.run()

run() 不会自行返回。它一直轮询，直到调用 shutdown()，然后在关闭完成后返回。Worker 也是异步上下文管理器：async with worker 在进入时启动 Worker，在退出时将其关闭。

注册工作流和 Activity

所有轮询同一任务队列的 Worker 必须注册相同的工作流类型和 Activity 类型。任务队列不会按类型进行路由，因此轮询该队列的任何 Worker 都可能收到该队列上的任何任务。如果 Worker 收到它未注册的类型的任务，该任务将失败。

在 workflows 参数中传入工作流列表，在 activities 参数中传入 Activity 列表，也可以两者都传。

使用 async def 定义的 Activity 将在 Worker 的事件循环上运行。使用普通 def 定义的 Activity 是同步的，需要执行器，因此需要通过 activity_executor 参数传入执行器：

worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[MyWorkflow],
    activities=[my_sync_activity],
    activity_executor=ThreadPoolExecutor(5),
)

同一个执行器可以在多个 Worker 之间共享。

配置 Worker 选项

Worker 构造函数接收关键字参数，用于控制并发限制、轮询器、超时时间和缓存等，其中包括 max_concurrent_activities、max_concurrent_workflow_tasks 和 max_cached_workflows。默认设置在大多数情况下都能正常工作。

运行版本化 Worker

在 deployment_config 中设置 Worker 的部署版本，启用版本管理，然后为 Worker 上的工作流设置默认版本行为。

worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[HelloWorkflow],
    activities=[some_activity],
    deployment_config=WorkerDeploymentConfig(
        version=WorkerDeploymentVersion(
            deployment_name="my-app",
            build_id="1.0",
        ),
        use_worker_versioning=True,
        default_versioning_behavior=VersioningBehavior.PINNED,
    ),
)

若要在每个工作流上设置行为，而不是在 Worker 上设置，请将 versioning_behavior 参数传递给 @workflow.defn。

关闭 Worker

通过退出 async with 代码块，关闭 Worker，该代码块将调用 shutdown()。如果要使 Worker 持续运行直到进程被中断，那么创建 asyncio.Event，并且在代码块内等待该事件。

在捕获 KeyboardInterrupt 的入口点处设置事件：

interrupt_event = asyncio.Event()

if __name__ == "__main__":
    loop = asyncio.new_event_loop()
    try:
        loop.run_until_complete(main())
    except KeyboardInterrupt:
        interrupt_event.set()
        loop.run_until_complete(loop.shutdown_asyncgens())

在代码块内等待 interrupt_event 将使 Worker 保持运行状态。一旦事件被设置，代码块退出，Worker 将停止轮询新任务，等待正在执行中的任务完成，最长等待时间为 graceful_shutdown_timeout。

worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[HelloWorkflow],
    activities=[some_activity],
    graceful_shutdown_timeout=timedelta(seconds=30),
)
async with worker:
    await interrupt_event.wait()