Asynchronous Processing Configuration

1. Overview

The FT Device Network Service implements asynchronous processing to efficiently handle high volumes of device data without blocking the main application flow. This approach allows the service to maintain responsiveness even under heavy load conditions, such as when processing telemetry from thousands of devices simultaneously.

Asynchronous processing is primarily used for:

  1. Device Data Processing - Parsing and storing device telemetry data

  2. MQTT Message Handling - Processing incoming device messages from the MQTT broker

  3. Batch Database Operations - Performing bulk inserts into ClickHouse

  4. Diagnostic Test Processing - Handling time-consuming network diagnostic operations

Spring’s @Async annotation is used throughout the application to execute methods in separate threads, managed by dedicated thread pools configured for specific workloads.

2. Configuration

The asynchronous processing is configured in the AsyncConfig class, which defines multiple thread pools with appropriate sizing for different types of tasks:

Unresolved include directive in modules/ROOT/pages/async.adoc - include::../src/main/java/com/friendly/network/config/AsyncConfig.java[]

The configuration uses the following properties from application.yml which can be overridden with environment variables:

app:
  task:
    executor:
      core-pool-size: ${TASK_CORE_POOL_SIZE:10}
      max-pool-size: ${TASK_MAX_POOL_SIZE:20}
      queue-capacity: ${TASK_QUEUE_CAPACITY:500}

  • TASK_CORE_POOL_SIZE - The number of threads to keep in the pool, even if they are idle (default: 10)

  • TASK_MAX_POOL_SIZE - The maximum number of threads to allow in the pool (default: 20)

  • TASK_QUEUE_CAPACITY - The capacity of the queue used for holding tasks before they are executed (default: 500)

3. Thread Pool Executors

The application defines two distinct thread pool executors:

3.1. taskExecutor

The taskExecutor is a general-purpose thread pool for most asynchronous operations in the application. It is configured with the following parameters:

  • Core pool size: ${app.task.executor.core-pool-size}

  • Max pool size: ${app.task.executor.max-pool-size}

  • Queue capacity: ${app.task.executor.queue-capacity}

  • Thread name prefix: app-task-

This executor is suitable for most standard operations that don’t require specialized thread allocation.

3.2. dataProcessingExecutor

The dataProcessingExecutor is designed specifically for data processing tasks that may be more resource-intensive. It is configured with double the capacity of the general-purpose executor:

  • Core pool size: ${app.task.executor.core-pool-size} * 2

  • Max pool size: ${app.task.executor.max-pool-size} * 2

  • Queue capacity: ${app.task.executor.queue-capacity} * 2

  • Thread name prefix: data-proc-

This executor is used for processing device data, performing batch database operations, and handling other data-intensive tasks.

4. Usage in the Application

Asynchronous processing is used throughout the application, primarily in the following areas:

4.1. Device Data Processing

The DeviceDataProcessingService uses asynchronous methods to process incoming device data:

@Async
public CompletableFuture<Void> processDeviceData(String serial, String rawData) {
    // Implementation details
}

@Async
public CompletableFuture<Void> processDeviceData(String serial, List<ParameterDto> parameters) {
    // Implementation details
}

These methods return CompletableFuture objects, allowing the caller to be notified when processing completes or handle any errors that occur during processing.

4.2. MQTT Message Handling

MQTT messages are processed asynchronously to prevent blocking the MQTT client thread:

@Override
public void handleMessage(Message<?> message) {
    executorService.submit(() -> {
        // Process MQTT message
    });
}

4.3. Database Operations

Batch database operations are performed asynchronously to optimize performance:

executorService.submit(() -> {
    try {
        // Perform batch database operation
    } catch (Exception e) {
        // Error handling
    }
});

5. Monitoring and Error Handling

The application includes comprehensive monitoring and error handling for asynchronous operations:

  • Metrics - Thread pool utilization, queue size, and task completion times are tracked using Micrometer metrics

  • Logging - Asynchronous operations log their status, including successful completion and any errors

  • Exception Handling - Uncaught exceptions in asynchronous tasks are captured and logged

  • CompletableFuture - Error handling is managed through CompletableFuture’s exception handling mechanisms

← Back | Main Page