DBOS Java：基于 Postgres 的持久化工作流框架

DBOS 提供构建于 Postgres 之上的轻量级持久化工作流。简单来说，它能帮助你编写长期运行、可靠的代码，使其在崩溃、重启和故障后仍能继续执行，同时不会丢失状态或重复工作。

工作流运行时，DBOS 会将其执行的每个步骤作为检查点保存到 Postgres 数据库中。当进程停止时——无论是执行失败、主动暂停，还是机器宕机——程序都可以从这些检查点恢复，准确还原此前的状态，并从中断的位置继续执行，就像什么都没发生过一样。

在实践中，这让构建可靠系统变得更加容易，适用于 AI Agent、数据同步、支付处理，以及任何需要几分钟、几天甚至几周才能完成的任务。你不必再临时拼凑重试逻辑和数据库检查点；DBOS 工作流提供了一套统一模型，确保程序能够从任意故障中恢复，并准确地从上次中断的位置继续执行。

这个库包含为程序添加持久化工作流所需的一切：不需要单独的服务或 orchestrator，除 Postgres 外也没有任何外部依赖。由于它只是一个库，你可以逐步将其引入现有项目，而且它能够直接配合 Spring 等框架使用。又因为它构建于 Postgres 之上，所以原生支持你熟悉的所有工具，包括备份工具、GUI 和 CLI 工具，并且兼容任意 Postgres 服务提供商。

工作流通过将自身状态作为检查点保存到 Postgres，让程序具备持久性。如果程序发生故障，那么在程序重启后，所有工作流都会自动从最后一个已完成的步骤恢复执行。

只需几行代码，将普通函数注册为工作流和步骤，就能为现有 Java 程序添加持久化工作流：

interface Example {
    public void workflow();
}

class ExampleImpl implements Example {

    private void stepOne() {
        System.out.println("Step one completed!");
    }

    private void stepTwo() {
        System.out.println("Step two completed!");
    }

    @Workflow
    public void workflow() {
        DBOS.runStep(() -> stepOne(), "stepOne");
        DBOS.runStep(() -> stepTwo(), "stepTwo");
    }
}

工作流尤其适合：

• 编排业务流程，使其能够从任意故障中无缝恢复。
• 构建可观测且容错的数据管道。
• 运行 AI Agent，或任何依赖不可靠、非确定性 API 的应用程序。

你可以异步运行工作流，而无须修改其接口或实现。

这非常适合长时间运行的后台工作流：你的代码可以稍后返回，检查任务是否完成，并获取执行结果。

var handle = DBOS.startWorkflow(()->example.exampleWorkflow("HelloDBOS"));
result = handle.getResult();

DBOS 队列可以帮助你持久、可靠地运行分布式任务。你可以从持久化工作流中将任务加入队列，随后由某个进程取出并执行。DBOS 会管理任务的执行：即使应用程序被中断，它也能保证任务最终完成，并确保调用方无须重新提交任务即可获得结果。

队列还提供流量控制能力，因此你可以按队列或按进程限制任务的并发数。你还可以为任务设置超时、限制队列任务的执行频率、对任务进行去重，或者设置任务优先级。

只需几行代码，就能为工作流添加队列。它们不需要单独的队列服务或消息代理，只需要 Postgres。

 public void queuedTasks() {
    for (int i = 0; i < 3; i++) {
        String workflowId = "child" + i;
        var options = new StartWorkflowOptions(workflowId).withQueue(q);
        List<WorkflowHandle<String>> handles = new ArrayList<>();
        handles.add(DBOS.startWorkflow(()->simpleService.childWorkflow(workflowId), options));
    }

    for (int i = 0 ; i < 3 ; i++) {
        String workflowId = "child"+i;
        var h = DBOS.retrieveWorkflow(workflowId);
        System.out.println(h.getResult());
    }
}

// In your main
var queue = new Queue("exampleQueue");
DBOS.registerQueue(queue);

你可以使用 cron 语法调度工作流，也可以通过持久化 sleep 将工作流暂停任意时长——哪怕是几天或几周——然后再继续执行。

只需一个 annotation，即可调度工作流：

public class SchedulerImpl implements Scheduler {
    
    @Workflow(name = "every5Second")
    @Scheduled(cron = "0/5 * * * * ?")
    public void every5Second(Instant schedule , Instant actual) {
        log.info("Executed workflow  "+  schedule.toString() + "   " + actual.toString()) ;
    }
}

// In your main
DBOS.registerWorkflows(Scheduler.class, new SchedulerImpl());

你可以暂停工作流，直到收到通知；也可以从工作流中发出事件，将进度更新发送给外部客户端。所有通知都会存储在 Postgres 中，因此能够以 exactly-once 语义发送和接收。在等待事件时，你还可以设置持久化超时。这样，即使期间发生中断或重启，也可以等待任意时长——哪怕是几天或几周——并在通知到达或超时时间结束后恢复执行。

例如，你可以构建一个可靠的计费工作流，持久地等待支付服务发来的通知，并以 exactly-once 语义处理该通知：

@Workflow(name = "billing")
public void billingWorkflow() {
    // Calculate the charge, then submit the bill to a payments service
    String paymentStatus = (String) DBOS.recv(PAYMENT_STATUS, paymentServiceTimeout);
    if (paymentStatus.equals("paid")) {
        // handle paid
    } else {
        // handle not paid
    }
}

@Workflow(name = "payment") 
public void payment(String targetWorkflowId) {
    DBOS.send(targetWorkflowId, PAYMENT_STATUS, "paid") ;
}

要开始使用，请按照快速入门指南安装这个开源库，并将其连接到 Postgres 数据库。随后，可以查阅编程指南，了解如何使用持久化工作流和队列进行开发。

https://docs.dbos.dev

https://docs.dbos.dev/examples

DBOS 与其他系统的对比

DBOS 和 Temporal 都提供持久化执行能力，但 DBOS 以构建于 Postgres 之上的轻量级库实现，而 Temporal 则通过外部编排服务器实现。

要将 DBOS 添加到程序中，只需安装这个开源库、将其连接到 Postgres，并使用 annotation 标注工作流和步骤。相比之下，要将 Temporal 添加到程序中，你必须重新设计程序架构：将工作流和步骤（activities）迁移到 Temporal worker，配置 Temporal server 来编排这些工作流，并且只能通过 Temporal client 访问工作流。相关页面对二者进行了更详细的比较。

如果你想提问或与社区成员交流，欢迎加入我们的 Discord！如果发现 bug 或希望提出功能请求，请随时在 GitHub 上创建 issue。如果你有兴趣参与贡献，请查阅我们的贡献指南。