dag-flow

A DAG-based parallel computation framework for Java

Simplify multi-threaded task orchestration — declare dependencies, and the framework maximizes parallelism automatically.

English | 中文

Features

DAG-based parallel execution — Automatically maximizes parallelism based on declared dependencies using CompletableFuture
Multiple command types — SyncCommand (caller thread), AsyncCommand (I/O pool), CalcCommand (CPU pool), BatchCommand (fan-out with ALL/ANY/AT_LEAST_N strategies)
Cycle detection — DFS-based cycle detection before execution with clear error reporting
Fluent builder API — Chain .node().depend() calls; builder is reusable across runs
Auto-naming — node(Class) auto-generates names (fetchOrder#0, fetchOrder#1, ...); node(Function) uses node#0, node#1, ...
Lambda support — node() accepts Function<C, R> or Consumer<C> for lightweight nodes
Extensible architecture — JobBuilder is designed for extension; create custom builders for third-party integrations
Resilience4j integration — dag-flow-resilience4j module provides CircuitBreaker, Retry, Bulkhead, RateLimiter, TimeLimiter support
Alibaba Sentinel integration — dag-flow-sentinel module provides flow control, degrade, and fallback support via Sentinel SphU.entry()
Spring Cloud Circuit Breaker integration — dag-flow-spring-cloud-circuitbreaker module wraps nodes with Spring Cloud CircuitBreaker abstraction
Spring Boot Starter — dag-flow-spring-boot-starter auto-configures dag-flow engine; dependSpringBean() works out of the box
Virtual threads — useVirtualThreads() enables Java 21 virtual threads for all non-sync nodes; traditional thread pools remain the default
Smart thread pools — I/O pool (2x–8x cores) and CPU pool (cores+1) with CallerRunsPolicy
OpenTelemetry tracing — Built-in distributed tracing via opentelemetry-api; creates root span per DAG run, child spans per node, and batch-item spans — no-op when SDK is absent
Replay profiling — enableReplay() records per-node start/end timing; print text-based Gantt charts and waterfall timelines; Spring Boot web UI at /dagflow/replay with configurable LRU cache
Node timeout — Per-node and DAG-level timeout support via CompletableFuture.orTimeout(); declare via builder .timeout() or command timeout() method
Lightweight retry — Built-in per-node retry with configurable max attempts and delay — no external dependencies needed
Conditional execution — dependIf() / dependIfNot() enable runtime branch selection; result-aware overloads receive the dependency's typed return value
Fallback / default values — Commands can define fallback(context, cause) to return degraded results on failure
DAG-level error handling — JobBuilder.onError() lets you decide whether to rethrow or ignore a failed run and inspect partial results
Serialization — DagFlowContext, DagFlowReplay, and NodeReplayRecord are Serializable
Thread-safe builder — JobBuilder.run() creates per-execution node snapshots; the same builder can be reused concurrently
Error propagation — Node exceptions propagate as ExecutionException; downstream nodes are cancelled

Module Structure

Module	Description
`dag-flow-core`	Core framework: DAG builder, runner, command API, thread pools
`dag-flow-resilience4j`	Resilience4j extension (CircuitBreaker, Retry, Bulkhead, RateLimiter, TimeLimiter)
`dag-flow-sentinel`	Alibaba Sentinel extension (flow control, degrade, fallback)
`dag-flow-spring-cloud-circuitbreaker`	Spring Cloud Circuit Breaker abstraction extension
`dag-flow-spring-boot-starter`	Spring Boot 4 auto-configuration starter

Quick Start

Installation

Add to your build.gradle:

dependencies {
    // Core (required)
    implementation 'com.lesofn:dag-flow-core:1.0-SNAPSHOT'

    // Spring Boot Starter (optional — auto-configures dag-flow in Spring Boot apps)
    implementation 'com.lesofn:dag-flow-spring-boot-starter:1.0-SNAPSHOT'

    // Resilience4j extension (optional)
    implementation 'com.lesofn:dag-flow-resilience4j:1.0-SNAPSHOT'

    // Alibaba Sentinel extension (optional)
    implementation 'com.lesofn:dag-flow-sentinel:1.0-SNAPSHOT'

    // Spring Cloud Circuit Breaker extension (optional)
    implementation 'com.lesofn:dag-flow-spring-cloud-circuitbreaker:1.0-SNAPSHOT'
}

1. Define a Context

The context carries request data and provides access to upstream results:

public class OrderContext extends DagFlowContext {
    private String orderId;
    // getters & setters
}

2. Define Command Nodes

// Async I/O node (runs on async thread pool)
public class FetchOrder implements AsyncCommand<OrderContext, Order> {
    @Override
    public Order run(OrderContext context) {
        return orderService.getById(context.getOrderId());
    }
}

// CPU-bound node (runs on calc thread pool)
public class CalcDiscount implements CalcCommand<OrderContext, BigDecimal> {
    @Override
    public BigDecimal run(OrderContext context) {
        Order order = context.getResult(FetchOrder.class);
        return discountEngine.calculate(order);
    }
}

3. Build and Run the DAG

OrderContext context = new OrderContext();
context.setOrderId("12345");

JobRunner<OrderContext> runner = new JobBuilder<OrderContext>()
        .node(FetchOrder.class)
        .node(FetchUser.class)
        .node(CalcDiscount.class).depend(FetchOrder.class, FetchUser.class)
        .node(BuildResult.class).depend(CalcDiscount.class)
        .run(context);

Result result = runner.getResult(BuildResult.class);

This builds and executes the following DAG:

FetchOrder   FetchUser      ← parallel (no mutual dependency)
        \     /
      CalcDiscount           ← waits for both
           |
       BuildResult

Examples

Runnable end-to-end examples are in the dag-flow-examples module. They cover e-commerce pricing, API orchestration with fallback/timeout, batch processing, Resilience4j/Sentinel/Spring Cloud Circuit Breaker resilience, virtual thread benchmarking, Spring bean integration, and Spring Boot replay.

# Run all examples
./gradlew :dag-flow-examples:run

# Run the Spring Boot replay test
./gradlew :dag-flow-examples:test

See dag-flow-examples/README.md for a complete list of example classes and individual run commands.

Architecture

Command Hierarchy

DagFlowCommand<C, R>                  // Base: R run(C context)
├── SyncCommand<C, R>                  // Runs on caller thread
├── AsyncCommand<C, R>                 // Runs on I/O thread pool
├── CalcCommand<C, R>                  // Runs on CPU thread pool
│   ├── FunctionCommand                // Lambda Function<C, R> wrapper
│   └── ConsumerCommand                // Lambda Consumer<C> wrapper
└── BatchCommand<C, P, R>              // Fan-out per param → Map<P, R> (ALL/ANY/atLeast)

Extensions (SyncCommand-based):
├── Resilience4jCommand                // dag-flow-resilience4j: Resilience4j decorator wrapper
├── SentinelCommand                    // dag-flow-sentinel: Sentinel resource wrapper
└── SpringCloudCircuitBreakerCommand     // dag-flow-spring-cloud-circuitbreaker: Spring Cloud Circuit Breaker wrapper

Core Components

Component	Description
`JobBuilder<C>`	Fluent API for DAG construction and node registration (extensible)
`JobRunner<C>`	`CompletableFuture`-based execution engine with result retrieval
`DagFlowContext`	Abstract context — subclass to carry request data and access upstream results by class or node name
`DagNode`	Runtime node wrapping a command with its future and dependencies
`DagNodeCheck`	DFS-based cycle detection, runs before execution
`DagFlowDefaultExecutor`	Default thread pool configuration for async and calc nodes
`DagFlowTracing`	OpenTelemetry tracing utility — root, node, and batch-item spans
`DagFlowReplay`	Immutable execution record with per-node timing data
`DagFlowReplayPrinter`	Text-based Gantt chart and waterfall timeline renderer

Default Thread Pools

Pool	Core	Max	Queue	Use Case
Async (I/O)	CPU × 2	CPU × 8	CPU × 16	Network calls, DB queries, file I/O
Calc (CPU)	CPU + 1	CPU + 1	CPU × 4	Computation, transformation, aggregation

Both pools use CallerRunsPolicy as the rejection handler.

Advanced Usage

Auto-Naming

When you use node(Class) without an explicit name, dag-flow auto-generates a name using the pattern className#0, className#1, etc.:

new JobBuilder<OrderContext>()
        .node(FetchOrder.class)       // auto-named "fetchOrder#0"
        .node(FetchOrder.class)       // auto-named "fetchOrder#1"
        .node(CalcDiscount.class)     // auto-named "calcDiscount#0"
        .run(context);

Lambda nodes use the prefix node: node#0, node#1, ...

You can also provide an explicit name:

builder.node("myCustomName", FetchOrder.class);

Retrieving Results by Name

Results can be fetched by node class or by node name. Auto-named nodes use the generated name (fetchOrder#0, node#1, etc.), and explicitly named nodes use the provided name.

JobRunner<OrderContext> runner = new JobBuilder<OrderContext>()
        .node("fetchOrder", FetchOrder.class)
        .node("calcDiscount", CalcDiscount.class).depend("fetchOrder")
        .run(context);

// By class
Order order = runner.getResult(FetchOrder.class);
Order orderFromContext = context.getResult(FetchOrder.class);

// By name
Order orderByName = runner.getResult("fetchOrder");
Order orderByNameFromContext = context.getResult("fetchOrder");

Both JobRunner and DagFlowContext provide getResult(Class) and getResult(String), so named or auto-named nodes can be retrieved without a class reference.

Lambda Nodes

For lightweight logic, skip creating a class:

new JobBuilder<OrderContext>()
        .node(FetchOrder.class)
        .node("format", (Function<OrderContext, String>) ctx -> {
            Order order = ctx.getResult(FetchOrder.class);
            return order.toString();
        }).depend(FetchOrder.class)
        .run(context);

Batch Command

Fan-out a set of parameters into parallel sub-tasks:

public class BatchFetch implements BatchCommand<MyContext, Long, String> {
    @Override
    public Set<Long> batchParam(MyContext context) {
        return Set.of(1L, 2L, 3L);
    }

    @Override
    public String run(MyContext context, Long param) {
        return fetchById(param);
    }
}

// Result is Map<Long, String>
Map<Long, String> results = runner.getResult(BatchFetch.class);

Batch Execution Strategies

By default, BatchCommand waits for all sub-tasks to complete. Override batchStrategy() to change this:

Strategy	Description
`BatchStrategy.ALL`	Wait for all sub-tasks (default)
`BatchStrategy.ANY`	Return as soon as 1 sub-task completes; cancel the rest
`BatchStrategy.atLeast(n)`	Return as soon as n sub-tasks complete; cancel the rest

public class FastBatchFetch implements BatchCommand<MyContext, Long, String> {
    @Override
    public Set<Long> batchParam(MyContext context) {
        return Set.of(1L, 2L, 3L, 4L, 5L);
    }

    @Override
    public String run(MyContext context, Long param) {
        return fetchFromReplica(param);
    }

    @Override
    public BatchStrategy batchStrategy() {
        return BatchStrategy.ANY;             // first result wins
        // return BatchStrategy.atLeast(3);   // wait for at least 3
    }
}

Batch Executor

BatchCommand can override executor() to run sub-tasks on a custom executor. Returning null (the default) uses the DAG's async executor.

public class BatchFetch implements BatchCommand<MyContext, Long, String> {
    @Override
    public Set<Long> batchParam(MyContext context) {
        return Set.of(1L, 2L, 3L);
    }

    @Override
    public String run(MyContext context, Long param) {
        return fetchById(param);
    }

    @Override
    public Executor executor() {
        return myCustomExecutor;
    }
}

In-Class Dependency Declaration

Commands can declare their own dependencies:

public class CalcDiscount implements CalcCommand<OrderContext, BigDecimal> {
    @Override
    public Class<? extends DagFlowCommand<OrderContext, ?>> dependNode() {
        return FetchOrder.class;
    }

    @Override
    public BigDecimal run(OrderContext context) { ... }
}

// No need to call .depend() in the builder
new JobBuilder<OrderContext>()
        .node(CalcDiscount.class)      // auto-resolves dependency on FetchOrder
        .run(context);

Resilience4j Integration

Use dag-flow-resilience4j module to add fault tolerance to DAG nodes:

// Add dependency: implementation 'com.lesofn:dag-flow-resilience4j:1.0-SNAPSHOT'

CircuitBreaker cb = CircuitBreaker.of("myService", CircuitBreakerConfig.ofDefaults());
Retry retry = Retry.of("myService", RetryConfig.custom().maxAttempts(3).build());

Resilience4jCommand<MyContext, String> command =
        new Resilience4jCommand<>(ctx -> callRemoteService(ctx))
                .withCircuitBreaker(cb)
                .withRetry(retry);

JobRunner<MyContext> runner = new Resilience4jJobBuilder<MyContext>()
        .addResilience4jNode("protectedCall", command)
        .node(DownstreamJob.class).depend("protectedCall")
        .run(context);

Supported decorators: CircuitBreaker, Retry, Bulkhead, RateLimiter, TimeLimiter — can be combined freely.

Alibaba Sentinel Integration

Use dag-flow-sentinel to protect DAG nodes with Sentinel rules (flow control, degrade, hot-param, etc.):

// Add dependency: implementation 'com.lesofn:dag-flow-sentinel:1.0-SNAPSHOT'

import com.alibaba.csp.sentinel.slots.block.RuleConstant;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;

FlowRule rule = new FlowRule();
rule.setResource("myResource");
rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
rule.setCount(100);   // 100 QPS limit
FlowRuleManager.loadRules(Collections.singletonList(rule));

SentinelCommand<MyContext, String> command =
        new SentinelCommand<>("myResource", ctx -> callRemoteService(ctx))
                .withFallback((ctx, cause) -> "fallback");

JobRunner<MyContext> runner = new SentinelJobBuilder<MyContext>()
        .addSentinelNode("protectedCall", command)
        .node(DownstreamJob.class).depend("protectedCall")
        .run(context);

The fallback is invoked when Sentinel blocks the call (BlockException) or the business function fails.

Spring Cloud Circuit Breaker Integration

Use dag-flow-spring-cloud-circuitbreaker to wrap nodes with any Spring Cloud Circuit Breaker implementation (e.g. Resilience4j):

// Add dependency: implementation 'com.lesofn:dag-flow-spring-cloud-circuitbreaker:1.0-SNAPSHOT'

CircuitBreakerFactory<?, ?> factory = ...; // injected or created manually
CircuitBreaker cb = factory.create("myService");

SpringCloudCircuitBreakerCommand<MyContext, String> command =
        new SpringCloudCircuitBreakerCommand<>(ctx -> callRemoteService(ctx))
                .withCircuitBreaker(cb)
                .withFallback(cause -> "fallback");

JobRunner<MyContext> runner = new SpringCloudCircuitBreakerJobBuilder<MyContext>()
        .addCircuitBreakerNode("protectedCall", command)
        .node(DownstreamJob.class).depend("protectedCall")
        .run(context);

Works with any org.springframework.cloud.client.circuitbreaker.CircuitBreaker implementation.

Spring Boot Starter

Add dag-flow-spring-boot-starter to auto-configure the dag-flow engine in Spring Boot applications:

// build.gradle
implementation 'com.lesofn:dag-flow-spring-boot-starter:1.0-SNAPSHOT'

Once installed, SpringContextHolder is auto-registered and dependSpringBean() works out of the box — no manual configuration needed:

// Spring beans that implement DagFlowCommand can be used as DAG nodes
@Component
public class OrderService implements AsyncCommand<OrderContext, Order> {
    @Autowired
    private OrderRepository orderRepository;

    @Override
    public Order run(OrderContext context) {
        return orderRepository.findById(context.getOrderId());
    }
}

// Reference Spring beans by name in DAG construction
new JobBuilder<OrderContext>()
        .node(CalcDiscount.class)
        .dependSpringBean("orderService")   // resolved from Spring ApplicationContext
        .run(context);

Configuration via application.properties / application.yml:

# Disable dag-flow auto-configuration (default: true)
dagflow.enabled=false

Virtual Threads (Java 21+)

Enable virtual threads for all non-sync nodes with a single call:

JobRunner<MyContext> runner = new JobBuilder<MyContext>()
        .useVirtualThreads()                   // enable virtual threads
        .node(FetchOrder.class)                // AsyncCommand → virtual thread
        .node(FetchUser.class)                 // AsyncCommand → virtual thread
        .node(CalcDiscount.class).depend(FetchOrder.class, FetchUser.class)
        .run(context);

SyncCommand — still runs on the caller thread (unchanged)
AsyncCommand / CalcCommand — runs on virtual threads instead of platform thread pools
Without useVirtualThreads(), the traditional I/O and CPU thread pools are used (default behavior)

OpenTelemetry Tracing

dag-flow automatically creates distributed tracing spans for every DAG execution. When opentelemetry-api is on the classpath:

A root span (dagflow.run) is created per DAG execution with a dagflow.node.count attribute
Each node gets a child span (dagflow.node.<name>) with dagflow.node.name and dagflow.node.type attributes
Each batch sub-task gets its own child span (dagflow.batch.<name>) with the dagflow.batch.param attribute

When no OpenTelemetry SDK is configured, all tracing operations are no-op with zero overhead.

// Option 1: Use GlobalOpenTelemetry (default — zero configuration needed)
// Just configure OpenTelemetry SDK globally as usual

// Option 2: Provide a custom instance
DagFlowTracing.setOpenTelemetry(myOpenTelemetrySdk);

// Then run the DAG as normal — spans are created automatically
new JobBuilder<MyContext>()
        .node(FetchOrder.class)
        .node(CalcDiscount.class).depend(FetchOrder.class)
        .run(context);

// Reset to GlobalOpenTelemetry
DagFlowTracing.setOpenTelemetry(null);

Spring Boot configuration:

# Disable tracing (default: true)
dagflow.tracing-enabled=false

Replay Profiling

Enable execution profiling to record per-node timing data. After execution, print text-based Gantt charts or waterfall timelines:

JobRunner<MyContext> runner = new JobBuilder<MyContext>()
        .enableReplay()                           // enable profiling
        .node(FetchOrder.class)
        .node(FetchUser.class)
        .node(CalcDiscount.class).depend(FetchOrder.class, FetchUser.class)
        .node(BuildResult.class).depend(CalcDiscount.class)
        .run(context);

// Get the replay record
DagFlowReplay replay = runner.getReplayRecord();

// Print text-based Gantt chart
System.out.println(replay.toGantt());

// Print Chrome DevTools-style waterfall timeline
System.out.println(replay.toTimeline());

Example Gantt chart output:

DAG Execution [4 nodes, 250ms, SUCCESS]  14:30:05.123
──────────────────────────────────────────────────────────────
FetchOrder   |████████████                                  |   0- 120ms [Async]
FetchUser    |██████                                        |   0-  60ms [Async]
CalcDiscount |            ██████████████                    | 120- 220ms [Calc]
BuildResult  |                          ████                | 220- 250ms [Sync]
──────────────────────────────────────────────────────────────

Example waterfall timeline output:

DAG Timeline [4 nodes, 250ms, SUCCESS]  14:30:05.123
──────────────────────────────────────────────────────────────
             0ms   50ms  100ms 150ms 200ms 250ms
             |     |     |     |     |     |
FetchOrder   ██████████████░░░░░░░░░░░░░░░░  120ms  Async  pool-1-thread-2
FetchUser    ████████░░░░░░░░░░░░░░░░░░░░░░   60ms  Async  pool-1-thread-3
CalcDiscount ░░░░░░░░░░░░██████████████░░░░  100ms  Calc   ForkJoin-1
BuildResult  ░░░░░░░░░░░░░░░░░░░░░░░░████░░   30ms  Sync   main
──────────────────────────────────────────────────────────────

Spring Boot web UI: When using dag-flow-spring-boot-starter, enable the replay web endpoint:

# application.properties
dagflow.replay-enabled=true
dagflow.replay-cache-size=100   # LRU cache: keep last N executions (default: 100)

Then visit http://localhost:8080/dagflow/replay for an interactive HTML waterfall visualization.

Endpoint	Description
`GET /dagflow/replay`	HTML list of cached execution records
`GET /dagflow/replay/{id}`	HTML detail with waterfall chart
`GET /dagflow/replay/{id}/text`	Plain text Gantt + timeline
`GET /dagflow/replay/api`	JSON list
`GET /dagflow/replay/api/{id}`	JSON detail

Custom Executor

Override the default thread pool for specific nodes:

// Per-node override in the command class
public class CustomJob implements AsyncCommand<MyContext, String> {
    @Override
    public Executor executor() {
        return myCustomExecutor;
    }
    // ...
}

// BatchCommand can override the executor for its sub-tasks
public class CustomBatch implements BatchCommand<MyContext, Long, String> {
    @Override
    public Executor executor() {
        return myCustomExecutor;
    }
    // ...
}

// Or pass executor to lambda nodes
builder.node("custom", myFunction, myExecutor);

Timeout Control

dag-flow supports both per-node and DAG-level timeouts to prevent blocking executions from hanging indefinitely.

Node-Level Timeout (Builder)

new JobBuilder<OrderContext>()
        .node(FetchOrder.class).timeout(Duration.ofSeconds(3))   // 3s timeout for this node
        .node(CalcDiscount.class).depend(FetchOrder.class)
        .run(context);

Node-Level Timeout (Command Declaration)

Commands can declare their own timeout by overriding the timeout() method:

public class FetchOrder implements AsyncCommand<OrderContext, Order> {
    @Override
    public Duration timeout() {
        return Duration.ofSeconds(3);   // declare timeout in the command itself
    }

    @Override
    public Order run(OrderContext context) {
        return orderService.getById(context.getOrderId());
    }
}

Builder-set timeout takes priority over command-declared timeout.

DAG-Level Timeout

new JobBuilder<OrderContext>()
        .node(FetchOrder.class)
        .node(CalcDiscount.class).depend(FetchOrder.class)
        .dagTimeout(Duration.ofSeconds(10))   // entire DAG must finish in 10s
        .run(context);

When a timeout is exceeded, the node's CompletableFuture completes exceptionally with TimeoutException, which propagates as ExecutionException.

Lightweight Retry

Built-in per-node retry in the core module — no external dependencies required:

new JobBuilder<OrderContext>()
        .node(FetchOrder.class).retry(3, Duration.ofMillis(100))   // 3 retries, 100ms delay
        .node(CalcDiscount.class).depend(FetchOrder.class)
        .run(context);

On failure, the node retries up to maxRetries times with the specified delay between attempts
After all retries are exhausted, the exception propagates (or fallback is invoked if configured)
Retry is per-node; each node's retry policy is independent
For batch commands, each batch item is retried independently

Conditional Execution

Use dependIf() and dependIfNot() to create dynamic DAGs where branches are selected at runtime based on upstream results:

new JobBuilder<OrderContext>()
        .node(CheckVip.class)
        .node(VipDiscount.class)
            .dependIf(CheckVip.class, ctx -> ctx.getResult(CheckVip.class))
        .node(NormalDiscount.class)
            .dependIfNot(CheckVip.class, ctx -> ctx.getResult(CheckVip.class))
        .node(BuildResult.class).depend(VipDiscount.class, NormalDiscount.class)
        .run(context);

                CheckVip
               /        \
   [if true]  /          \  [if false]
    VipDiscount    NormalDiscount
               \        /
              BuildResult       ← gets non-null result from the executed branch

dependIf(dep, predicate) — add dependency and execute only if predicate returns true
dependIfNot(dep, predicate) — add dependency and execute only if predicate returns false
Skipped nodes complete with null result; downstream nodes still trigger
The condition predicate is evaluated after all dependencies complete, so ctx.getResult() is safe to call

Fallback / Default Values

Commands can define a fallback() method to return a degraded result when execution fails:

public class FetchOrder implements AsyncCommand<OrderContext, Order> {
    @Override
    public Order run(OrderContext context) throws Exception {
        return orderService.getById(context.getOrderId());   // may throw
    }

    @Override
    public Order fallback(OrderContext context, Throwable cause) {
        return Order.empty();   // return a default/degraded result
    }

    @Override
    public boolean hasFallback() {
        return true;   // must return true to enable fallback
    }
}

When a node fails (after all retries if retry is configured), fallback(context, cause) is invoked
If fallback succeeds, the node completes normally with the fallback result
Downstream nodes see the fallback result as a normal result
If fallback itself throws, the exception propagates as usual
hasFallback() must return true to activate the fallback mechanism

Combining retry + fallback:

new JobBuilder<OrderContext>()
        .node(FetchOrder.class).retry(2, Duration.ofMillis(100))   // retry 2 times, then fallback
        .run(context);

Project Structure

dag-flow/
├── dag-flow-core/                           # Core module
│   └── src/main/java/com/lesofn/dagflow/
│       ├── JobBuilder.java                  # Fluent DAG construction (extensible)
│       ├── JobRunner.java                   # CompletableFuture execution engine
│       ├── api/
│       │   ├── DagFlowCommand.java          # Base command interface
│       │   ├── SyncCommand.java             # Synchronous command
│       │   ├── AsyncCommand.java            # Async (I/O) command
│       │   ├── CalcCommand.java             # CPU-bound command
│       │   ├── BatchCommand.java            # Batch fan-out command
│       │   ├── BatchStrategy.java           # Batch strategies: ALL, ANY, atLeast(N)
│       │   ├── context/                     # Context & injection interfaces
│       │   ├── depend/                      # Dependency declaration interface
│       │   └── function/                    # Lambda wrappers
│       ├── exception/                       # DagFlowBuildException, CycleException, etc.
│       ├── executor/                        # Default thread pool configuration
│       ├── model/                           # DagNode, DagNodeCheck, DagNodeFactory
│       ├── replay/                          # Replay profiling (DagFlowReplay, Printer)
│       ├── tracing/                         # OpenTelemetry tracing (DagFlowTracing)
│       └── spring/                          # Optional Spring integration
├── dag-flow-resilience4j/                   # Resilience4j extension module
│   └── src/main/java/com/lesofn/dagflow/resilience4j/
│       ├── Resilience4jCommand.java         # Resilience4j decorator wrapper
│       └── Resilience4jJobBuilder.java      # Builder with addResilience4jNode()
├── dag-flow-sentinel/                       # Alibaba Sentinel extension module
│   └── src/main/java/com/lesofn/dagflow/sentinel/
│       ├── SentinelCommand.java             # Sentinel resource wrapper
│       └── SentinelJobBuilder.java          # Builder with addSentinelNode()
├── dag-flow-spring-cloud-circuitbreaker/    # Spring Cloud Circuit Breaker extension module
│   └── src/main/java/com/lesofn/dagflow/springcloud/circuitbreaker/
│       ├── SpringCloudCircuitBreakerCommand.java   # Circuit breaker wrapper
│       └── SpringCloudCircuitBreakerJobBuilder.java # Builder with addCircuitBreakerNode()
└── dag-flow-spring-boot-starter/            # Spring Boot 4 auto-configuration starter
    └── src/main/java/com/lesofn/dagflow/spring/boot/
        ├── autoconfigure/
        │   ├── DagFlowAutoConfiguration.java    # Auto-registers SpringContextHolder + replay beans
        │   └── DagFlowProperties.java           # dagflow.enabled, replay-enabled, etc.
        └── replay/
            ├── DagFlowReplayStore.java          # LRU cache for execution records
            └── DagFlowReplayController.java     # Web endpoint (/dagflow/replay)

Performance Benchmark

Platform thread pool vs Virtual threads comparison (Java 21, 8 vCPU):

Scenario	Platform Threads	Virtual Threads	Speedup
10 Parallel I/O (100ms each)	200.97 ms	101.24 ms	1.99x
50 Parallel I/O (50ms each)	302.88 ms	58.43 ms	5.18x
100 Parallel I/O (20ms each)	222.91 ms	26.19 ms	8.51x
8 Parallel CPU Nodes	4.65 ms	4.47 ms	1.04x
Mixed DAG (5 I/O → CPU → I/O)	102.71 ms	102.53 ms	1.00x
Multi-Layer DAG (3×10, 30ms)	121.80 ms	62.03 ms	1.96x

Key takeaways:

Virtual threads excel at I/O-bound workloads — up to 8.5x faster with 100 concurrent blocking nodes
For CPU-bound tasks, performance is comparable (virtual threads add negligible overhead)
The more concurrent I/O nodes, the greater the advantage of virtual threads over fixed-size thread pools

Run benchmarks yourself:

./gradlew :dag-flow-core:benchmark    # Benchmarks only (excluded from default test)

Build & Test

./gradlew build                                 # Build all modules
./gradlew test                                  # Run all tests (Spock + JUnit Platform)
./gradlew :dag-flow-core:test                  # Run core tests only
./gradlew :dag-flow-resilience4j:test          # Run Resilience4j tests only
./gradlew :dag-flow-sentinel:test             # Run Sentinel tests only
./gradlew :dag-flow-spring-cloud-circuitbreaker:test  # Run Spring Cloud Circuit Breaker tests only
./gradlew :dag-flow-spring-boot-starter:test   # Run Spring Boot Starter tests only
./gradlew :dag-flow-core:benchmark             # Run performance benchmarks
./gradlew clean build                           # Clean build

Requirements

Java 21+
Gradle 8.10+ (wrapper included)

License

Apache License 2.0

Name		Name	Last commit message	Last commit date
dag-flow-core		dag-flow-core
dag-flow-examples		dag-flow-examples
dag-flow-resilience4j		dag-flow-resilience4j
dag-flow-sentinel		dag-flow-sentinel
dag-flow-spring-boot-starter		dag-flow-spring-boot-starter
dag-flow-spring-cloud-circuitbreaker		dag-flow-spring-cloud-circuitbreaker
gradle/wrapper		gradle/wrapper
.gitignore		.gitignore
AGENTS.md		AGENTS.md
LICENSE		LICENSE
README.md		README.md
README_zh.md		README_zh.md
build.gradle.kts		build.gradle.kts
gradlew		gradlew
gradlew.bat		gradlew.bat
settings.gradle.kts		settings.gradle.kts

Latest commit

History

Folders and files

Repository files navigation