无法捕获Spring中异步方法引发的异常

Question

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

一个pplication.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);
        }
    }
}

创建Bucket。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");
        }
    }
}

endpoint。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-集成

在您的场景中，我建议使用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<>();