工作流持久化执行，以 Java 库的形式

// Kafka 监听器：重复投递不得启动第二个作业——
// 事件 id 是幂等键
@KafkaListener(topics = "orders")
public void onOrderCreated(
    String eventId, String orderId, String userId) {
    try {
        jobSystem.startJob(
            JobOptions.jobWithExternalId(eventId),
            this::processOrder, orderId, userId);
    } catch (JobAlreadyExistsException e) {
        // 重复投递：作业已存在
        log.info("Event {} already has a job", eventId);
    }
}

public void processOrder(String orderId, String userId) {
    // 重试直至成功——事件不会丢失
}

// Kafka 监听器：重复投递不得启动第二个作业——
// 事件 id 是幂等键
@KafkaListener(topics = "orders")
fun onOrderCreated(
    eventId: String, orderId: String, userId: String
) {
    try {
        jobSystem.startJob(
            JobOptions.jobWithExternalId(eventId),
            this::processOrder, orderId, userId)
    } catch (e: JobAlreadyExistsException) {
        // 重复投递：作业已存在
        log.info("Event {} already has a job", eventId)
    }
}

fun processOrder(orderId: String, userId: String) {
    // 重试直至成功——事件不会丢失
}

// 作业运行期间接收指令
public void exportReport(ExportRequest request) {
    Signal<ExportCommand> command = Signal.of();
    for (String page : request.pages()) {
        ExportCommand cmd = command.waitAndPop();
        render(page, cmd);
    }
}

// 从任何位置——例如 REST 控制器：
jobSystem.signal(job, new ExportCommand("change-scope"));

// 作业运行期间接收指令
fun exportReport(request: ExportRequest) {
    val command: Signal<ExportCommand> = Signal.of()
    for (page in request.pages()) {
        val cmd = command.waitAndPop()
        render(page, cmd)
    }
}

// 从任何位置——例如 REST 控制器：
jobSystem.signal(job, ExportCommand("change-scope"))

@Service
public class ReportScheduler {
    private final RecurringJobScheduler recurring;

    public void register() {
        // 每晚 02:00 执行——本身就是一个持久化作业
        recurring.scheduleAtCron(
            "nightly-report", "0 0 2 * * *",
            this::generateReport);
    }

    public void generateReport() {
        // 崩溃后自动恢复
    }
}

@Service
class ReportScheduler(
    private val recurring: RecurringJobScheduler,
) {
    fun register() {
        // 每晚 02:00 执行——本身就是一个持久化作业
        recurring.scheduleAtCron(
            "nightly-report", "0 0 2 * * *",
            this::generateReport)
    }

    fun generateReport() {
        // 崩溃后自动恢复
    }
}

// 数据库中的分块：借助保存的索引，重启后
// 可以从停止处继续，无需重复处理已完成的分块
public void processBacklog() {
    // 返回已保存的索引，首次运行返回 0
    Integer index = Job.computeState("index",
        Integer.class, current ->
            current == null ? 0 : current);

    // 从保存的索引开始获取分块
    List<Chunk> chunks =
        chunkRepository.findFrom(index);

    for (int i = 0; i < chunks.size(); i++) {
        Job.throwIfInterruptedOrCancelled();
        chunkService.handle(chunks.get(i));
        Job.setState("index", index + i + 1);
    }
}

// 数据库中的分块：借助保存的索引，重启后
// 可以从停止处继续，无需重复处理已完成的分块
fun processBacklog() {
    // 返回已保存的索引，首次运行返回 0
    val index = Job.computeState("index",
        Integer::class.java) { it ?: 0 }

    // 从保存的索引开始获取分块
    val chunks =
        chunkRepository.findFrom(index)

    chunks.forEachIndexed { i, chunk ->
        Job.throwIfInterruptedOrCancelled()
        chunkService.handle(chunk)
        Job.setState("index", index + i + 1)
    }
}

// Saga：每个步骤注册自己的补偿。
// 失败时，已完成的步骤按逆序回滚。
public String placeOrder(String orderId) {
    Deque<Runnable> compensate = new ArrayDeque<>();
    try {
        String paymentId = jobSystem.execute(
            paymentService::charge, String.class, orderId);
        compensate.push(() -> jobSystem.execute(
            paymentService::refund, orderId, paymentId));

        String shipmentId = jobSystem.execute(
            deliveryService::ship, String.class, orderId);
        compensate.push(() -> jobSystem.execute(
            deliveryService::cancelShipment, shipmentId));

        jobSystem.execute(loyaltyService::addPoints, orderId);
        return paymentId;
    } catch (JobFailure e) {
        compensate.forEach(Runnable::run); // 逆序回滚
        throw e;
    }
}

// Saga：每个步骤注册自己的补偿。
// 失败时，已完成的步骤按逆序回滚。
fun placeOrder(orderId: String): String {
    val compensate = ArrayDeque<() -> Unit>()
    try {
        val paymentId = jobSystem.execute(
            paymentService::charge, String::class.java, orderId)
        compensate.addFirst {
            jobSystem.execute(
                paymentService::refund, orderId, paymentId)
        }

        val shipmentId = jobSystem.execute(
            deliveryService::ship, String::class.java, orderId)
        compensate.addFirst {
            jobSystem.execute(
                deliveryService::cancelShipment, shipmentId)
        }

        jobSystem.execute(loyaltyService::addPoints, orderId)
        return paymentId
    } catch (e: JobFailure) {
        compensate.forEach { it() } // 逆序回滚
        throw e
    }
}

// 作业在步骤之间检查是否被取消
public void exportReport(ExportRequest request) {
    for (String page : request.pages()) {
        Job.throwIfInterruptedOrCancelled();
        render(page);
    }
}

// 从任何位置：取消会级联到子作业
jobSystem.cancel(job);

// 作业在步骤之间检查是否被取消
fun exportReport(request: ExportRequest) {
    for (page in request.pages()) {
        Job.throwIfInterruptedOrCancelled()
        render(page)
    }
}

// 从任何位置：取消会级联到子作业
jobSystem.cancel(job)