boxlang-async-programming

BoxLang 异步编程

BoxLang 是一个 AI 原生软件生产力平台，用于构建、现代化改造和运行应用程序，让开发者与 AI Agent 协同工作。

概述

BoxLang 提供了全面的异步框架，基于 Java 的 CompletableFuture 和 Project Loom 虚拟线程构建。AsyncService 管理执行器、调度器 和 future。所有异步原语都与 BoxLang 运行时无缝集成。

创建 BoxFuture

BoxFuture 扩展了 CompletableFuture，支持适配 BoxLang 的链式调用。

futureNew() — 主要 BIF（v1.4.0+）

// Create a future from a function (runs asynchronously)
var future = futureNew( () => fetchDataFromAPI() )

// Create a completed future with a value
var future = futureNew( "Hello World" )

// Create an empty future (complete later)
var future = futureNew()

// Create with a specific executor
var future = futureNew( () => heavyCalculation(), "cpu-tasks" )

asyncRun() — 简化的异步执行

// Run a function asynchronously on the default io-tasks executor
var future = asyncRun( () => fetchDataFromAPI() )

// Run with specific executor
var future = asyncRun( () => processCPUWork(), "cpu-tasks" )

// Get the result (blocks until complete)
var result = future.get()

// With timeout
var result = future.get( 5, "seconds" )

使用 then、thenAsync 和 onError 链式调用

• then() — 在同一线程上运行转换（适用于快速、轻量的转换）
• thenAsync() — 在执行器线程上运行转换（适用于繁重的工作或 I/O 操作）

asyncRun( () => fetchUser( userId ) )
    .then( (user) => enrichWithProfile( user ) )          // same thread
    .thenAsync( (user) => sendWelcomeEmail( user ) )      // executor thread
    .then( (user) => user )
    .onError( (error) => {
        logError( error.message )
        return getDefaultUser()
    })
    .get()

核心结果方法

// Blocking retrieval
var result = future.get()
var result = future.get( 5000 )              // timeout in ms
var result = future.get( 5, "seconds" )      // timeout with unit

// Safe retrieval with defaults
var result = future.getOrDefault( "fallback" )
var result = future.joinOrDefault( 0 )       // join() variant with default

// Get as Attempt object (functional error handling)
var attempt = future.getAsAttempt()
if ( attempt.isPresent() ) {
    doSomething( attempt.get() )
}

使用 asyncAll 并行执行

asyncAll() 并发运行多个操作并返回一个 BoxFuture<Array> — 调用一次 .get() 即可按顺序获取所有结果：

// Returns BoxFuture<Array> — .get() resolves to [result1, result2, result3]
var results = asyncAll([
    () => fetchOrders( userId ),
    () => fetchProfile( userId ),
    () => fetchPreferences( userId )
]).get()

var [orders, profile, prefs] = results

// Mix lambdas, closures, and pre-created futures
var results = asyncAll([
    () => fetchOrders( userId ),           // function
    futureNew( () => fetchProfile( id ) ), // pre-created future
    () => fetchPreferences( userId )       // function
]).get()

使用 asyncAny 竞速完成

asyncAny() 返回最先完成的 future 的结果（v1.4.0+）：

var fastestResult = asyncAny([
    () => fetchFromPrimaryDB(),
    () => fetchFromReplicaDB(),
    () => fetchFromCache()
]).get()

使用 asyncAllApply 并行处理集合

并行对数组或结构体中的每个元素应用函数（v1.4.0+）：

// Array processing in parallel
var userIds = [ 1, 2, 3, 4, 5 ]
var profiles = asyncAllApply(
    userIds,
    ( id ) => fetchUserProfile( id )  // each item processed in parallel
)
// profiles = [ profile1, profile2, profile3, profile4, profile5 ]

// Struct processing in parallel
var config = { db: "prod-db", cache: "redis", queue: "rabbit" }
var validated = asyncAllApply(
    config,
    ( item ) => validateConfig( item.key, item.value )  // item = { key, value }
)

BoxFuture::allApply() 还提供了一个无需错误处理器的重载（v1.18+）：

results = BoxFuture::allApply( myArray, ( item ) => processItem( item ) )

runAsync()、asyncAll() 和 asyncAllApply() 会将其 JDBC 连接和事务与调用方隔离（v1.18+），因此异步工作会获取新的连接，而不是与父级共享连接。

命名执行器

// Three pre-configured runtime executors:
// "io-tasks"        — virtual threads (default, best for I/O-bound work)
// "cpu-tasks"       — scheduled pool, 20 threads (best for CPU-bound work)
// "scheduled-tasks" — scheduled pool, 20 threads (for cron/periodic tasks)

// Pass executor name as second arg to asyncRun / futureNew
asyncRun( () => cpuIntensiveWork(), "cpu-tasks" )
futureNew( () => fetchData(), "io-tasks" )

// Access an executor by name
var executor = executorGet( "io-tasks" )

异步流水线

// Sequential pipeline
var result = futureNew( () => loadRawData() )
    .then( (data) => parseData( data ) )
    .then( (parsed) => validate( parsed ) )
    .then( (valid) => persist( valid ) )
    .onError( (e) => rollback() )
    .get()

// Mix sequential and parallel stages
var pipeline = asyncRun( () => fetchConfig() )
    .then( (config) => {
        // Fan out: run two tasks in parallel with the config
        var results = asyncAll([
            () => buildReport( config ),
            () => sendNotifications( config )
        ]).get()
        return results
    })
    .get()

定时任务

创建一个 BoxLang Scheduler 类（Scheduler.bx），并在 boxlang.json 中注册：

// schedulers/MyScheduler.bx
class {

    property name="scheduler"
    property name="logger"

    function configure() {
        scheduler.setSchedulerName( "MyApp-Scheduler" )
        scheduler.setTimezone( "UTC" )

        // Register tasks with fluent DSL
        scheduler.task( "cleanupExpiredSessions" )
            .call( () => sessionService.cleanup() )
            .every( 15, "minutes" )
            .onFailure( (task, error) => logError( error.message ) )

        scheduler.task( "dailyReport" )
            .call( () => reportService.generate() )
            .every( 1, "day" )
            .startOn( "00:00" )
    }

    void function onStartup() {
        logger.info( "Scheduler started: #scheduler.getSchedulerName()#" )
    }

    void function onShutdown() {
        logger.info( "Scheduler shutting down" )
    }

    void function onAnyTaskError( required task, required exception ) {
        logger.error( "Task '#task.getName()#' failed: #exception.message#" )
    }
}

在 boxlang.json 中注册：

{
    "scheduler": {
        "schedulers": [ "/path/to/schedulers/MyScheduler.bx" ]
    }
}

或从命令行运行：

boxlang schedule /path/to/schedulers/MyScheduler.bx

thread 组件

用于显式线程管理：

// Start a named thread
thread name="backgroundWorker" action="run" {
    // Code runs in a new thread
    processLargeDataset( datasetId )
}

// Start multiple threads
thread name="worker1" action="run" {
    processChunk( chunk1 )
}
thread name="worker2" action="run" {
    processChunk( chunk2 )
}

// Wait for threads to finish
thread action="join" name="worker1,worker2" timeout=30000

// Access thread results
var result1 = cfthread.worker1.result
var result2 = cfthread.worker2.result

使用 bx:lock 进行分布式锁定

bx:lock 可防止竞争条件，并支持分布式锁定：

// Exclusive lock (one thread at a time)
bx:lock name="userUpdate_#userId#" type="exclusive" timeout=10 {
    // Only one request can be here at a time for this userId
    user = userService.update( userId, data )
}

// Read lock (multiple readers, exclusive writer)
bx:lock name="configCache" type="readonly" timeout=5 {
    var config = cacheGet( "appConfig" )
}
bx:lock name="configCache" type="exclusive" timeout=5 {
    cachePut( "appConfig", freshConfig )
}

// Scoped locks (application, session, request)
bx:lock scope="application" type="exclusive" timeout=10 {
    if ( !application.initialized ) {
        initializeApp()
        application.initialized = true
    }
}

文件和目录监视器（v1.12+）

// Watch a directory for changes
var watcher = directoryWatcher(
    path     = expandPath( "./config" ),
    onChange = (event) -> {
        // event.type: "create", "modify", "delete"
        // event.path: full path to changed file
        reloadConfig( event.path )
    },
    filter   = "*.json",
    recurse  = false
)

watcher.start()

// Stop later
watcher.stop()

异步代码中的错误处理

// onError continues the chain with a recovery value
var result = asyncRun( () => flakyOperation() )
    .onError( (err) => {
        logError( err.message )
        return defaultValue   // chain continues with this
    })
    .then( (val) => process( val ) )
    .get()

// exceptionally (Java-style)
var future = asyncRun( () => riskyWork() )
future.exceptionally( (err) => fallback )

// Check completion state
if ( future.isDone() ) { ... }
if ( future.isCompletedExceptionally() ) { ... }
if ( future.isCancelled() ) { ... }

// Cancel a future
future.cancel( true )

带超时和回退的异步操作

// orTimeout — throws TimeoutException after delay
var result = asyncRun( () => slowExternalAPI() )
    .orTimeout( 3, "seconds" )
    .onError( (err) => getCachedResult() )
    .get()

// completeOnTimeout — resolves with default value instead of throwing
var result = asyncRun( () => slowExternalAPI() )
    .completeOnTimeout( getDefaultData(), 3, "seconds" )
    .get()

虚拟线程最佳实践

// Virtual threads (io-tasks, default) — ideal for:
// - HTTP calls, database queries, file I/O, any blocking I/O
asyncRun( () => httpClient.get( url ) )             // good — io-tasks
asyncRun( () => queryExecute( sql ) )               // good — io-tasks

// CPU pool — ideal for CPU-intensive work:
// - Image processing, encryption, data transformation
asyncRun( () => encryptLargeFile( path ), "cpu-tasks" )
futureNew( () => processLargeDataset( data ), "cpu-tasks" )

执行器类型和运行时配置

预配置的运行时执行器

BoxLang 提供三个预配置的执行器，可通过 executorGet() 访问：

// Access pre-configured runtime executors
var ioExecutor        = executorGet( "io-tasks" )        // Virtual threads for I/O
var cpuExecutor       = executorGet( "cpu-tasks" )       // Scheduled pool for CPU work
var scheduledExecutor = executorGet( "scheduled-tasks" ) // For scheduled tasks

// Use directly
var future = ioExecutor.submit( () => fetchFromAPI( url ) )
var result = future.get()

运行时执行器配置（boxlang.json）：

{
    "executors": {
        "io-tasks":        { "type": "virtual" },
        "cpu-tasks":       { "type": "scheduled", "threads": 10 },
        "scheduled-tasks": { "type": "scheduled", "threads": 10 }
    }
}

创建自定义执行器

BoxLang 通过 executorNew() 支持七种执行器类型：

// Virtual threads — best for I/O-bound tasks (default, JVM 21+)
var ioPool = executorNew( "virtual", "my-io-pool" )

// Fixed thread pool — CPU-bound tasks with predictable concurrency
var cpuPool = executorNew( type="fixed", name="cpu-pool", threads=8 )

// Cached pool — grows/shrinks dynamically for bursty workloads
var dynPool = executorNew( "cached", "burst-pool" )

// Single thread — sequential guaranteed-order execution
var seqPool = executorNew( "single", "sequential-processor" )

// Fork-join — recursive divide-and-conquer algorithms
var fjPool = executorNew( type="fork_join", name="fj-pool", parallelism=8 )

// Work-stealing — auto load-balancing across threads
var wsPool = executorNew( type="work_stealing", name="load-balanced", parallelism=8 )

调度执行 API

scheduled 执行器类型提供 scheduleOnce、scheduleAtFixedRate 和 scheduleWithFixedDelay：

var executor = executorNew( type="scheduled", name="scheduler", threads=5 )

// Run once after a delay
var future = executor.scheduleOnce(
    () => performMaintenance(),
    5,        // delay value
    "seconds" // time unit
)

// Run at a fixed rate (regardless of task duration)
var future = executor.scheduleAtFixedRate(
    () => healthCheck(),
    0,        // initial delay
    30,       // period
    "seconds"
)

// Run with fixed delay BETWEEN executions (waits for task to finish first)
var future = executor.scheduleWithFixedDelay(
    () => processQueue(),
    10,       // initial delay
    5,        // delay between completions
    "seconds"
)

// Submit a callable and get result
var resultFuture = executor.submit( () => complexCalculation() )
var result = resultFuture.get()

// Shutdown executor when done
executor.shutdown()

执行器统计信息

var stats = executor.getStatistics()
writeOutput( "Active threads:   #stats.activeCount#" )
writeOutput( "Completed tasks:  #stats.completedTaskCount#" )
writeOutput( "Pool size:        #stats.poolSize#" )
writeOutput( "Is shutdown:      #executor.isShutdown()#" )

参考资料

• 异步编程
• 调度
• 线程处理
• 锁定