es.davy.ai

Preguntas y respuestas de programación confiables

¿Tienes una pregunta?

Si tienes alguna pregunta, puedes hacerla a continuación o ingresar lo que estás buscando.

Aspecto que no funciona con Completable Future

11 Jul, 2023 Programación

Tengo una lista de Consumidores que quiero ejecutar en múltiples hilos. Tengo un aspecto para registrar y manejar todas las excepciones. A continuación se muestra el fragmento de código.

El problema es que el aspecto funciona bien hasta que se crean varios hilos. Después de eso, en caso de excepción (@LogFailure), el código no entra en el consejo.

@Override
public void consumerExecutor(Map<><iprocessdto>, IProcessDTO> map, IContext context) {
    List<><void>> completableFutures = new ArrayList<>();

    Iterator consumerIterator = map.keySet().iterator();

    while (consumerIterator.hasNext()) {
        Consumer<iprocessdto> consumer = (Consumer<iprocessdto>) consumerIterator.next();

        CompletableFuture<void> future = CompletableFuture.runAsync(() -> {
            executeSingleConsumer(0, consumer, map.get(consumer));
        });

        completableFutures.add(future);
    }

    try {
        CompletableFuture<void> allFuturesResult = CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[completableFutures.size()]));

        allFuturesResult.thenApply(v ->
                completableFutures.stream()
                        .map(future -> future.join())
                        .collect(Collectors.toList())
        );
    } catch (CancellationException | CompletionException exception) {

    }
}

@LogFailure
@LogSuccess
private void executeSingleConsumer(int tries, Consumer<iprocessdto> consumer, IProcessDTO processDto) {
    try {
        consumer.accept(null);
    } catch (Exception ex) {
        if (tries < maxAttempts) {
            executeSingleConsumer(tries + 1, consumer, processDto);
        } else if (tries == maxAttempts) {
            throw new ProcessException();
        } else {
            throw ex;
        }
    }
}

@AfterThrowing(pointcut = "execution(@com.LogFailure * *.*(..))", argNames = "jp,ex", throwing = "ex")
public void handleLogFailure(JoinPoint jp, Exception ex) {
    // excepciones manejadas
}

¿Alguna idea de qué estoy haciendo mal?

Tags:  , , , ,

Answer

  1. Avatar for davy.ai
    davy.ai

    El problema aquí es que el método executeSingleConsumer se está llamando dentro del método CompletableFuture.runAsync, lo que significa que el aspecto @LogFailure no se activa en caso de una excepción. Esto ocurre porque la excepción se maneja dentro del método executeSingleConsumer y no se propaga al llamador del CompletableFuture. Por lo tanto, el aspecto no puede interceptar la excepción.

    Para solucionar este problema, puedes modificar el método executeSingleConsumer para que lance la excepción en lugar de manejarla. Luego, puedes usar el método exceptionally de la clase CompletableFuture para interceptar cualquier excepción y manejarla usando el aspecto @LogFailure. A continuación se muestra el código modificado:

    private void executeSingleConsumer(int tries, Consumer<iprocessdto> consumer, IProcessDTO processDto) throws Exception {
    try {
        consumer.accept(null);
    } catch (Exception ex) {
        if (tries < maxAttempts) {
            executeSingleConsumer(tries + 1, consumer, processDto);
        } else if (tries == maxAttempts) {
            throw new ProcessException();
        } else {
            throw ex;
        }
    }
}

@Override
public void consumerExecutor(Map<><iprocessdto>, IProcessDTO> map, IContext context) {
    List<><void>> completableFutures = new ArrayList<>();
    Iterator consumerIterator = map.keySet().iterator();

    while (consumerIterator.hasNext()) {
        Consumer<iprocessdto> consumer = (Consumer<iprocessdto>) consumerIterator.next();
        CompletableFuture<void> future = CompletableFuture.runAsync(() -> {
            try {
                executeSingleConsumer(0, consumer, map.get(consumer));
            } catch (Exception ex) {
                throw new RuntimeException(ex);
            }
        }).exceptionally(ex -> {
            handleLogFailure(null, (Exception) ex);
            return null;
        });
        completableFutures.add(future);
    }

    try {
        CompletableFuture<void> allFuturesResult = CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[completableFutures.size()]));
        allFuturesResult.thenApply(v ->
                completableFutures.stream().
                        map(future -> future.join()).
                        collect(Collectors.toList())
        );
    } catch (CancellationException | CompletionException exception) {

    }
}

    En este código modificado, el método executeSingleConsumer ahora lanza cualquier excepción que encuentre. El método CompletableFuture.runAsync ahora llama a este método dentro de un bloque try-catch y lanza una RuntimeException si se captura alguna excepción. Luego, el método exceptionally de la clase CompletableFuture intercepta esta RuntimeException y llama al método handleLogFailure con la excepción capturada.

    Con esta modificación, el aspecto @LogFailure debería poder interceptar cualquier excepción lanzada por el método executeSingleConsumer, incluso cuando se ejecuta en múltiples hilos.

Comments are closed.