无法从 Spring 中的异步方法捕获抛出的异常

Question

我无法从 Spring 中的异步方法中捕获抛出的异常。我编写了一个未捕获的异常处理程序来捕获但没有成功。 该应用程序将能够启动任意数量的永久运行的异步作业。 我认为我的异步方法需要返回 Future 以便我可以将其存储在 hashmap 中并检查其状态或停止工作。我也可以通过存储它来获取所有正在运行的作业。 我认为我不能使用 get 方法，因为如果输入正确，它会阻塞，我的工作将永远运行。如果输入正常，我需要发送已启动状态。每当 Async 方法中发生异常时，它就会被抛出，但我无法捕捉到它。我怎样才能做到这一点？ 这是我的完整代码。

Application.java

@SpringBootApplication
@EnableAsync
public class Application  {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

AsyncConfig.java

@EnableAsync
@Configuration
public class AsyncConfig implements AsyncConfigurer {

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(5);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("MyExecutor-");
        executor.initialize();
        return executor;
    }


    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new AsyncExceptionHandler();
    }
}

AsyncExceptionHandler.java

public class AsyncExceptionHandler implements AsyncUncaughtExceptionHandler {
    @Override
    public void handleUncaughtException(Throwable throwable, Method method, Object... obj) {

        System.out.println("Exception Cause - " + throwable.getMessage());
        System.out.println("Method name - " + method.getName());
        for (Object param : obj) {
            System.out.println("Parameter value - " + param);
        }
    }
}

createBucket.java

@Service
public class createBucket {
    @Async
    public Future<String> start(String config){
        try {
            JSONObject map = new JSONObject(config);
            Jedis jedis = new Jedis(map.getString("jedisip"));
            jedis.auth(map.getString("password"));
            // code to make a kafka consumer subscribe to a topic given in config input
            while(true) {
                //forever running code which polls using a kafka consumer
            }
        }
        catch(JedisException j) {
            throw new JedisException("Some msg");
        }
    }
}

端点.java

@Controller
public class Endpoint {
    @Autowired
    private createBucket service;

    private Future<String> out;
    private HashMap<String, Future<String>> maps = new HashMap<>();

    @PostMapping(value = "/start", consumes = "application/json", produces = "application/json")
    public ResponseEntity<String> starttask(@RequestBody String conf) {
        try {
        out = service.start(conf);
        maps.put(conf, out);
    }
    catch (Exception e) {
        return new ResponseEntity<>("exception", HttpStatus.BAD_REQUEST);
    }
        return new ResponseEntity<>("{\"started\":\"true\"}", HttpStatus.CREATED);
    }
}

Answer 1

如官方文档中所述，AsyncUncaughtExceptionHandler 用于 void 返回值。 https://docs.spring.io/spring/docs/5.1.10.RELEASE/spring-framework-reference/integration.html#spring-integration

在您的场景中，我建议使用 CompletableFuture 和 DeferredResult：

    @Async
    public CompletableFuture<String> start(String config) {
        CompletableFuture completableFuture = new CompletableFuture();
        try {
            JSONObject map = new JSONObject(config);
            Jedis jedis = new Jedis(map.getString("jedisip"));
            jedis.auth(map.getString("password"));
            completableFuture.complete("started!");
        }
        catch(JedisException j) {
            completableFuture.completeExceptionally(j);
        }

        return completableFuture;
    }

    @PostMapping(value = "/start", consumes = "application/json", produces = "application/json")
    public DeferredResult<ResponseEntity> starttask(@RequestBody String conf) {

        CompletableFuture<String> start = service.start(conf);

        DeferredResult<ResponseEntity> deferredResult = new DeferredResult<>();

        start.whenComplete((res, ex) -> {
            if (ex == null) {
                ResponseEntity<String> successEntity = new ResponseEntity<>("{\"started\":\"true\"}", HttpStatus.CREATED);\
                deferredResult.setResult(successEntity);
            } else {
                // handle ex here!
                ResponseEntity<String> exEntity = new ResponseEntity<>("exception", HttpStatus.BAD_REQUEST);
                deferredResult.setResult(exEntity);
            }
        });

        return deferredResult;

    }

还有一个严重的问题。以下代码不是线程安全的。

private Future<String> out;
private HashMap<String, Future<String>> maps = new HashMap<>();