boxlang-async-programming

BoxLang 非同期プログラミング

BoxLang は、開発者と AI エージェントが協力してアプリケーションを構築、モダナイズ、実行できる、AI ネイティブなソフトウェア生産性プラットフォームです。

概要

BoxLang は、Java の CompletableFuture と Project Loom の仮想スレッドを基盤とする包括的な非同期フレームワークを提供します。AsyncService はエグゼキューター、スケジューラー、フューチャーを管理します。すべての非同期プリミティブは BoxLang ランタイムとシームレスに統合されます。

BoxFuture の作成

BoxFuture は、BoxLang に適したチェーン処理機能を備えた CompletableFuture の拡張です。

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() は、最初に完了したフューチャーの結果を返します（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" ]
    }
}

または、CLI から実行します。

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() で取得できる事前設定済みのエグゼキューターを 3 種類提供します。

// 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() から 7 種類のエグゼキューターをサポートします。

// 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()#" )

参考資料

• 非同期プログラミング
• スケジューリング
• スレッド処理
• ロック