tessl/maven-io-quarkus--quarkus-smallrye-fault-tolerance
Build fault-tolerant network services
—
Pending
asynchronous-strategies.mddocs/
Asynchronous Strategies
Non-blocking execution patterns with both blocking and non-blocking asynchronous strategies for improved scalability and resource utilization.
Capabilities
Basic Asynchronous Execution
Standard asynchronous execution using CompletionStage for non-blocking operations.
@Asynchronous
public CompletionStage<ReturnType> asyncMethod();
@Asynchronous
public CompletableFuture<ReturnType> asyncMethodWithFuture();Usage Example
@ApplicationScoped
public class AsyncService {
// Async HTTP call
@Asynchronous
@Timeout(10000)
public CompletionStage<ApiResponse> callExternalApiAsync(String endpoint) {
return CompletableFuture.supplyAsync(() -> {
return httpClient.get(endpoint);
});
}
// Async database operation
@Asynchronous
@Retry(maxRetries = 3)
public CompletableFuture<List<User>> findUsersAsync(UserQuery query) {
return CompletableFuture.supplyAsync(() -> {
return userRepository.find(query);
});
}
// Async file processing
@Asynchronous
@Bulkhead(value = 4, waitingTaskQueue = 10)
public CompletionStage<ProcessingResult> processFileAsync(File file) {
return CompletableFuture.supplyAsync(() -> {
return fileProcessor.process(file);
});
}
}Non-blocking Asynchronous Execution
SmallRye-specific non-blocking asynchronous execution using Mutiny reactive types.
@AsynchronousNonBlocking
public Uni<ReturnType> nonBlockingAsyncMethod();
@AsynchronousNonBlocking
public Multi<ReturnType> nonBlockingStreamMethod();Usage Example
@ApplicationScoped
public class ReactiveService {
@Inject
MutinyHttpClient httpClient;
@Inject
ReactiveUserRepository userRepository;
// Non-blocking HTTP call
@AsynchronousNonBlocking
@Timeout(8000)
@Retry(maxRetries = 2)
public Uni<ApiResponse> callApiNonBlocking(String endpoint) {
return httpClient.get(endpoint)
.map(response -> ApiResponse.from(response));
}
// Non-blocking database query returning single result
@AsynchronousNonBlocking
@CircuitBreaker(requestVolumeThreshold = 10, failureRatio = 0.5)
public Uni<User> findUserNonBlocking(Long userId) {
return userRepository.findById(userId);
}
// Non-blocking stream processing
@AsynchronousNonBlocking
@Bulkhead(value = 6, waitingTaskQueue = 15)
public Multi<ProcessedItem> processItemsNonBlocking(List<Item> items) {
return Multi.createFrom().iterable(items)
.map(item -> processor.process(item));
}
}Async with Fault Tolerance Combinations
Combining asynchronous execution with comprehensive fault tolerance strategies.
@Asynchronous
@Retry(maxRetries = 3, delay = 1000)
@CircuitBreaker(requestVolumeThreshold = 15, failureRatio = 0.4)
@Timeout(15000)
@Fallback(fallbackMethod = "asyncFallback")
public CompletionStage<ReturnType> resilientAsyncMethod();Usage Example
@ApplicationScoped
public class ResilientAsyncService {
// Comprehensive async resilience pattern
@Asynchronous
@Retry(maxRetries = 3, delay = 2000)
@CircuitBreaker(requestVolumeThreshold = 10, failureRatio = 0.3)
@Timeout(20000)
@Fallback(fallbackMethod = "processPaymentFallback")
@Bulkhead(value = 5, waitingTaskQueue = 12)
public CompletionStage<PaymentResult> processPaymentAsync(PaymentRequest request) {
return CompletableFuture.supplyAsync(() -> {
return paymentGateway.process(request);
});
}
public CompletionStage<PaymentResult> processPaymentFallback(PaymentRequest request) {
return CompletableFuture.completedFuture(
PaymentResult.queued(request.getTransactionId())
);
}
// Non-blocking with comprehensive fault tolerance
@AsynchronousNonBlocking
@Retry(maxRetries = 5)
@ExponentialBackoff(factor = 2, maxDelay = 30000)
@CircuitBreaker(requestVolumeThreshold = 20, failureRatio = 0.5)
@RateLimit(value = 100, window = 1, windowUnit = ChronoUnit.MINUTES)
public Uni<DataProcessingResult> processDataAsync(DataProcessingRequest request) {
return dataProcessor.processAsync(request)
.onFailure().retry().withBackOff(Duration.ofSeconds(1))
.onFailure().recoverWithItem(DataProcessingResult.failed());
}
}Async Context Propagation
Proper context propagation for security, transaction, and request context in async operations.
Usage Example
@ApplicationScoped
public class ContextAwareAsyncService {
@Inject
SecurityContext securityContext;
@Inject
RequestContext requestContext;
// Async operation with security context propagation
@Asynchronous
@RolesAllowed("USER")
public CompletionStage<UserData> getUserDataAsync(Long userId) {
return CompletableFuture.supplyAsync(() -> {
// Security context is automatically propagated
return userService.getData(userId);
});
}
// Non-blocking with request context
@AsynchronousNonBlocking
@Transactional
public Uni<SaveResult> saveDataAsync(DataEntity data) {
// Transaction context is propagated to reactive chain
return Uni.createFrom().item(() -> {
return dataRepository.save(data);
});
}
// Complex async workflow with context
@Asynchronous
@Authenticated
public CompletionStage<WorkflowResult> executeWorkflowAsync(WorkflowRequest request) {
return CompletableFuture
.supplyAsync(() -> workflowEngine.validate(request))
.thenCompose(validationResult -> {
if (validationResult.isValid()) {
return workflowEngine.executeAsync(request);
} else {
return CompletableFuture.completedFuture(
WorkflowResult.invalid(validationResult.getErrors())
);
}
});
}
}Reactive Streams and Event Processing
Asynchronous event processing and reactive streams handling.
Usage Example
@ApplicationScoped
public class EventProcessingService {
@Inject
EventBus eventBus;
// Event stream processing
@AsynchronousNonBlocking
@RateLimit(value = 1000, window = 1, windowUnit = ChronoUnit.SECONDS)
public Multi<ProcessedEvent> processEventStream(Multi<Event> eventStream) {
return eventStream
.map(event -> eventProcessor.process(event))
.filter(result -> result.isValid())
.onFailure().retry().withBackOff(Duration.ofMillis(100));
}
// Async event publishing
@Asynchronous
@Bulkhead(value = 8, waitingTaskQueue = 20)
public CompletionStage<PublishResult> publishEventAsync(DomainEvent event) {
return CompletableFuture.supplyAsync(() -> {
return eventBus.publish(event);
});
}
// Message processing with backpressure
@AsynchronousNonBlocking
@Retry(maxRetries = 3)
public Uni<MessageProcessingResult> processMessageAsync(Message message) {
return messageProcessor.processAsync(message)
.onFailure(MessageFormatException.class).recoverWithNull()
.onFailure().retry().withBackOff(Duration.ofSeconds(1));
}
}Types
Asynchronous Core Types
// Java standard async types
interface CompletionStage<T> {
CompletionStage<T> thenApply(Function<T, U> fn);
CompletionStage<T> thenCompose(Function<T, CompletionStage<U>> fn);
CompletionStage<T> thenCombine(CompletionStage<U> other, BiFunction<T, U, V> fn);
CompletionStage<T> handle(BiFunction<T, Throwable, U> fn);
CompletionStage<T> exceptionally(Function<Throwable, T> fn);
}
interface CompletableFuture<T> extends CompletionStage<T> {
static <U> CompletableFuture<U> completedFuture(U value);
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier);
static CompletableFuture<Void> runAsync(Runnable runnable);
}Mutiny Reactive Types
// Mutiny Uni for single async values
interface Uni<T> {
Uni<T> map(Function<T, U> mapper);
Uni<T> flatMap(Function<T, Uni<U>> mapper);
Uni<T> onFailure();
Uni<T> onItem();
CompletionStage<T> subscribeAsCompletionStage();
}
// Mutiny Multi for streams
interface Multi<T> {
Multi<T> map(Function<T, U> mapper);
Multi<T> filter(Predicate<T> predicate);
Multi<T> onFailure();
Multi<T> onItem();
Uni<List<T>> collect().asList();
}Context Types
// Security context for async operations
interface SecurityContext {
Principal getUserPrincipal();
boolean isUserInRole(String role);
}
// Request context propagation
interface RequestContext {
String getRequestId();
Map<String, Object> getAttributes();
}Install with Tessl CLI
npx tessl i tessl/maven-io-quarkus--quarkus-smallrye-fault-tolerance