人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习【免费下载链接】agent-core-javaopenJiuwen agent core Java是 openJiuwen Core的 Java版本提供AI Agent开发、运行、调优与演进相关的全套SDK能力。项目地址https://gitcode.com/openJiuwen/agent-core-java点击查看免费下载导读TaskScheduler是 openJiuwen agent-core-javaopenJiuwen Core 的 Java 实现中 Controller 体系的核心调度组件负责周期扫描处于SUBMITTED状态的任务为每个任务创建TaskExecutor、在虚拟线程中驱动执行输出、维护任务状态机并处理暂停、取消与完成信号。阅读本文后你将掌握 TaskScheduler 的任务调度循环、状态流转、并发模型、超时看门狗、事件发布与完成信号机制的完整实现原理并能在 ControllerAgent 场景下正确配置调度参数、注册自定义任务执行器。TaskScheduler 在 Controller 体系中的定位在 openJiuwen 的 Controller 架构中TaskScheduler与TaskManager、EventQueue、EventHandler一起构成 ControllerAgent 的核心运行链路。从 Controller 文档 可以看到Controller负责连接这些子模块并在会话维度上驱动任务创建、调度、流式输出和状态持久化而TaskScheduler在其中承担周期性拉取可执行任务并分派执行器的职责。从源码看TaskScheduler类定义于 TaskScheduler.java其 Javadoc 明确描述了四条核心工作流周期扫描待执行任务状态为SUBMITTED使用虚拟线程并发执行多个任务流式输出任务执行过程中产生的输出根据输出类型更新任务状态。该实现镜像自 Python 版openjiuwen/core/controller/modules/task_scheduler.py与 Python 版保持行为对齐。核心状态与依赖TaskScheduler构造时注入六个依赖并通过taskManager.setOnTaskSubmitted(this::notifyTaskSubmitted)挂接任务提交唤醒回调。其核心成员如下表成员类型说明configControllerConfig调度和超时配置调度间隔、并发上限、任务超时等。taskManagerTaskManager任务存储与状态更新入口同时负责优先级索引和父子任务关系维护。contextEngineContextEngine为执行器提供上下文能力。abilityManagerObject为执行器提供业务能力实际使用时会 instanceof 判断后转为AbilityManager。eventQueueEventQueue发布任务完成、交互和失败事件。taskExecutorRegistryTaskExecutorRegistrytaskType到执行器构造器的注册表。sessionsMapString, AgentSessionApi当前活跃会话。runningTasksMapString, RunningTaskEntry当前运行中的任务线程和执行器实例。cardBaseCard当前 Agent 卡片信息用于生成agentId等运行时标识。schedulerFuture/schedulerScheduledFuture/ScheduledExecutorService定时调度任务句柄与调度线程池。runningboolean调度器运行标记。wakeupPendingAtomicBoolean唤醒去重标记合并并发提交触发的即时扫描。lockReentrantLock保护runningTasks等共享结构的可重入锁。其中TaskExecutorRegistry是执行器分发的关键见 TaskExecutorRegistry.java它内部维护MapString, FunctionTaskExecutorDependencies, TaskExecutor taskExecutorBuilders支持动态注册addTaskExecutor、移除removeTaskExecutor与按类型获取getTaskExecutor若未找到对应类型的构造器会抛出AGENT_CONTROLLER_TASK_EXECUTION_ERROR错误。主要方法一览方法返回说明getTaskExecutorRegistry()TaskExecutorRegistry获取执行器注册表。pauseTask(String taskId)boolean在会话和执行器都存在时先调用canPause()/pause()再中断虚拟线程并把状态改为PAUSED。cancelTask(String taskId)boolean在会话和执行器都存在时先调用canCancel()/cancel()再中断虚拟线程并把状态改为CANCELED。start()void启动定时调度循环按scheduleInterval扫描SUBMITTED任务。stop()void停止调度线程、打断全部运行中任务并等待线程收尾。notifyTaskSubmitted()void任务进入SUBMITTED时唤醒调度循环带去重合并。ensureSessionCompletionSignal(String sessionId)void会话内无活跃任务时向会话写入all_tasks_processed结束块。调度循环与并发模型start()定时扫描的启动start()的逻辑TaskScheduler.java非常直观running true; scheduler OpenJiuwenExecutors.newScheduledThreadPool(task-scheduler, 1, true); long intervalMs (long) (config.getScheduleInterval() * 1000); schedulerFuture scheduler.scheduleWithFixedDelay( this::scheduleLoop, 0, intervalMs, TimeUnit.MILLISECONDS);它使用单线程的定时线程池以scheduleInterval默认 1.0 秒为固定延迟周期执行scheduleLoop()首次执行无延迟。scheduleLoop()任务扫描与并发控制每次扫描scheduleLoop()通过TaskManager.getTask(TaskFilter.builder().status(TaskStatus.SUBMITTED).build())拉取所有SUBMITTED任务逐个执行以下判定会话校验sessions.get(task.getSessionId())找不到会话则跳过该任务并发上限当config.getMaxConcurrentTasks() 0且当前运行时不支持虚拟线程时若runningTasks.size() config.getMaxConcurrentTasks()则停止本轮扫描break。从 OpenJiuwenExecutors.java 的注释可以看出设计意图JDK 21 虚拟线程下线程创建开销可忽略应由上层准入控制统一管控并发只有 JDK 17 平台线程模式下maxConcurrentTasks才作为并发闸生效默认值为max(40, CPU 核数 × 8)I/O 阻塞型任务并发槽位估算去重runningTasks.containsKey(taskId)已存在则跳过启动任务通过OpenJiuwenExecutors.newThread(() - executeTaskWrapper(taskId, session), task- taskId, true)创建线程并start()同时以new RunningTaskEntry(null, taskThread)登记到runningTasks。newThread的底层逻辑OpenJiuwenExecutors.java是JDK 21 及以上返回虚拟线程始终为守护线程JDK 17 返回平台线程此时isDaemon参数生效并为两种路径统一设置未捕获异常处理器。notifyTaskSubmitted()提交即唤醒为了降低调度延迟TaskManager.updateTaskStatus(..., SUBMITTED, ...)与addTask都会触发onTaskSubmitted回调。notifyTaskSubmitted()利用wakeupPendingAtomicBoolean做 CAS 去重——并发的一批任务提交最多触发一次即时扫描周期性调度仍然保留作为兜底防止任务堆积TaskScheduler.java。任务执行流程详解executeTaskWrapper()超时看门狗与异常包装executeTaskWrapper()是每个任务线程的入口。当config.getTaskTimeout() ! null时它为当前任务启动一个独立的 watchdog 线程池按taskTimeout * 1000毫秒调度一个中断任务Thread currentThread Thread.currentThread(); ScheduledExecutorService watchdog OpenJiuwenExecutors.newScheduledThreadPool(task-timeout- taskId, 1, true); long timeoutMs (long) (config.getTaskTimeout() * 1000); ScheduledFuture? timeout watchdog.schedule(() - { Loggers.CONTROLLER.error(Task {} timed out after {} seconds, taskId, config.getTaskTimeout()); currentThread.interrupt(); }, timeoutMs, TimeUnit.MILLISECONDS); try { executeTask(taskId, session); } finally { timeout.cancel(false); watchdog.shutdown(); }若执行期间收到InterruptedException调度器会记录 Task timeout after X seconds调用handleTaskExecutionFailure()把任务标记为FAILED并发布失败事件其他异常则记录执行失败信息后同样走失败处理。无论成功或失败finally块都会从runningTasks移除该任务条目并调用ensureSessionCompletionSignal()检查会话完成信号。executeTask()执行器实例化与流式消费executeTask()的具体步骤TaskScheduler.java通过taskManager.getTask(TaskFilter.byTaskId(taskId))取出Task找不到则抛AGENT_CONTROLLER_TASK_EXECUTION_ERROR构造TaskExecutorDependencies(config, abilityManager, contextEngine, taskManager, eventQueue)——该依赖封装类见 TaskExecutorDependencies.java用于减少构造参数传递复杂度经taskExecutorRegistry.getTaskExecutor(task.getTaskType(), dependencies)实例化执行器并把执行器写入RunningTaskEntry将任务状态更新为WORKING调用executor.executeAbility(taskId, session)获得IteratorControllerOutputChunk逐个 chunk先session.writeStream(chunk)写入会话流再依据 payload 类型分支处理。TaskExecutor是抽象基类TaskExecutor.java定义了executeAbility、canPause、pause、canCancel、cancel五个抽象方法以及PauseCheckResult、CancelCheckResult两个 record 类型。不同业务任务类型需要实现自己的TaskExecutor子类并通过TaskExecutorRegistry注册。ControllerOutputChunk 与 payload 类型执行器产出的是 ControllerOutputChunk.java它扩展会话流体系的OutputSchema包含index、payload、lastChunk三个核心字段。payload 类型由 ControllerOutputPayload.java 限定为五类payload 类型常量值含义EventType.TASK_COMPLETIONtask_completion任务完成EventType.TASK_INTERACTIONtask_interaction需要交互等待输入EventType.TASK_FAILEDtask_failed任务失败TASK_PROCESSINGprocessing处理中继续消费ALL_TASKS_PROCESSEDall_tasks_processed全部任务已处理会话完成块任务状态机任务状态由 TaskStatus.java 定义共 9 个取值SUBMITTED、WORKING、PAUSED、INPUT_REQUIRED、COMPLETED、CANCELED、FAILED、WAITING、UNKNOWN。其注释给出的状态流转为submitted - working - (completed | failed | paused | canceled) working - input-required - (继续执行或取消)TaskScheduler在executeTask()中按 payload 类型驱动状态迁移task_completion→COMPLETED、task_interaction→INPUT_REQUIRED、task_failed→FAILEDprocessing则继续消费下一个 chunk。暂停与取消机制pauseTask()与cancelTask()采用前置检查 执行器协作 线程中断 状态更新四步模式暂停pauseTaskTaskScheduler.java校验任务存在且会话存在从runningTasks取RunningTaskEntry任务未运行则返回false若执行器非空先executor.canPause(taskId, session)检查是否可暂停不可暂停时返回false并记录原因再executor.pause(...)让执行器协作暂停中断执行线程taskManager.updateTaskStatus(taskId, TaskStatus.PAUSED)。取消cancelTaskTaskScheduler.java若任务仍为SUBMITTED尚未开始直接置为CANCELED并返回true若任务已处于CANCELED/COMPLETED/FAILED终态返回false运行中的任务走与暂停相同的检查/协作/中断流程最终置为CANCELED。事件发布与完成信号publishTaskEvent()输出回送 EventQueuepublishTaskEvent()会把输出 chunk 的 payload 转换成对应事件再回送到EventQueueTaskScheduler.javatask_completion→TaskCompletionEvent(payloadData, task)task_interaction→TaskInteractionEvent(payloadData, task)task_failed→ 提取TextDataFrame中的错误文本写入task.setErrorMessage(...)再构造TaskFailedEvent(errorMsg, task)事件发布前还会执行mergeEventMetadata()将 payload 元数据与任务元数据合并payload 优先使事件携带完整的上下文信息。EventQueue.publishEvent(card.getId(), session, event)按{agentId}_{sessionId}_{eventType}主题格式见 EventQueue.java同步分发给已订阅的EventHandler。会话完成信号areAllTasksCompleted()ensureSessionCompletionSignal()在任务结束后检查会话级完成状态若config.isSuppressCompletionSignal()为true直接返回不发送完成块否则调用areAllTasksCompleted(sessionId)通过TaskFilter.bySessionId(sessionId)拉取会话内全部任务仅把SUBMITTED和WORKING视为活跃状态——因此暂停、取消、失败和输入等待INPUT_REQUIRED等状态都允许会话收到完成信号全部非活跃时构造ControllerOutputPayload.allTasksProcessed(All tasks have been successfully processed)对应的 chunklastChunktrue写入会话流。这一设计保证了会话流能够及时收尾避免客户端长时间挂起等待。调度相关配置参数TaskScheduler 行为完全由 ControllerConfig.java 驱动与调度直接相关的参数如下配置项JSON 字段默认值说明maxConcurrentTasksmax_concurrent_tasksdefaultTaskConcurrency()max(40, CPU核数×8)JDK 17 平台线程模式下的并发上限JDK 21 虚拟线程模式下该闸自动放开。scheduleIntervalschedule_interval1.0调度扫描间隔秒setter 强制要求 0.1。taskTimeouttask_timeoutnull任务超时秒非空时启用 watchdog 中断setter 要求为正数或 null。suppressCompletionSignalsuppress_completion_signalfalse是否抑制会话完成信号。defaultTaskPrioritydefault_task_priority1新任务默认优先级。enableTaskPersistenceenable_task_persistencefalse是否启用任务持久化。eventQueueSizeevent_queue_size10000事件队列大小。eventTimeoutevent_timeout300.0事件超时秒要求 100。streamFirstFrameTimeoutstream_first_frame_timeout30.0流式首帧超时。stop()优雅停机stop()的处理顺序TaskScheduler.javarunning false阻止后续扫描中断runningTasks中所有存活线程schedulerFuture.cancel(true)取消定时任务通过OpenJiuwenExecutors.shutdown(scheduler)关闭调度线程池并同步注销托管线程池注册表中的记录对所有运行线程join(5000)等待收尾超时则继续清理不阻塞清空runningTasks与sessions。源码级验证TaskSchedulerTest仓库为 TaskScheduler 提供了专门的焦点测试 TaskSchedulerTest.java覆盖了四个关键行为执行器注册与依赖注入验证TaskExecutorDependencies五个依赖与注入实例同一性注册test_task类型后可获取TestTaskExecutor实例移除后再次获取会抛出RuntimeException提交唤醒 元数据合并设置 5 秒调度间隔远超测试时长通过taskManager.addTask提交任务验证notifyTaskSubmitted唤醒路径并断言事件元数据同时包含 payload 的payload_key与任务的task_key且shared键按payload 优先合并取消语义对SUBMITTED任务首次cancelTask返回true置为CANCELED再次取消返回false完成信号抑制suppressCompletionSignaltrue时不写完成块关闭抑制后同一会话立即收到all_tasks_processed块。测试中的TestTaskExecutor展示了自定义执行器的最小实现只需实现executeAbility返回 chunk 迭代器以及 pause/cancel 四元组即可接入调度器。小结TaskScheduler以定时扫描 提交唤醒双通道拉取SUBMITTED任务以虚拟线程承载任务执行以ControllerOutputChunk流式回传输出并驱动任务状态机同时通过 watchdog 实现超时兜底、通过TaskExecutorRegistry实现任务类型的可扩展注册、通过EventQueue与all_tasks_processed完成块衔接事件处理与会话收尾。理解它的调度循环、状态迁移与并发控制是开发基于 openJiuwen ControllerAgent 的多任务、可交互、可中断业务场景的基础。可进一步阅读Controller 文档TaskScheduler 在控制器整体链路中的调用关系ControllerConfig 源码全部调度与事件参数TaskManager 源码任务存储、优先级索引与父子关系TaskSchedulerTest调度器核心行为验证用例赞分享人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习【免费下载链接】agent-core-javaopenJiuwen agent core Java是 openJiuwen Core的 Java版本提供AI Agent开发、运行、调优与演进相关的全套SDK能力。项目地址https://gitcode.com/openJiuwen/agent-core-java点击查看免费下载相关推荐rg 使用指南从安装、搜索到排查的完整流程附速查表rg 使用指南从安装、搜索到排查的完整流程附速查表 ripgrep缩写 rg是基于正则表达式的文件搜索工具rg 搜索默认递归遍历目录跳过隐藏文件、人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习OpenSquilla 调度指南用 cron 命令组管理周期性 Agent 任务OpenSquilla 调度指南用 cron 命令组管理周期性 Agent 任务 OpenSquilla 的调度Scheduling能力让网关gatew人工智能大模型AI Agent交互助手工具调用MCP 服务Agent 记忆RAG本地部署NetBox 的 Job 模型深度解析后台任务的调度、执行与生命周期管理NetBox 的 Job 模型深度解析后台任务的调度、执行与生命周期管理 NetBox 通过 Job 模型统一管理所有后台任务的调度、执行记录与状态跟踪涵盖后端网络数据建模上一篇ClawPanel第三引擎DeepSeek Harness受管运行时安装、模型渠道同步与独立工作台实战下一篇Squad并行执行与Git Worktree实战多个AI Agent同时开工互不冲突的完整做法创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考