DeferredResult实现异步处理+长轮询

桃扇骨 2021-06-22 15:38 614阅读 0赞

DeferredResult与Callable实现的功能类似，都是异步返回，只不过Callable不能直接设置超时时间，需要与FutureTask配合才行；DeferredResult可直接设置超时时间。

核心流程：

1、定义一个DeferredResult：DeferredResult<ResponseMsg<String>> deferredResult = new DeferredResult<>(OUT\_OF\_TIME, OUT\_OF\_TIME\_RESULT);

2、然后在主线程中直接返回deferredResult结果；此时servlet容器线程被释放，继续服务其他请求，以此提高吞吐量，后台任务线程执行耗时长的任务；

3、将任务放入队列中，后台定义一个专门执行任务的线程，循环执行队列中的任务；

4、执行完的任务，直接调用deferredResult.setResult()方法，即可将结果返回给客户端，和Callable、Future性质一样。

具体案例：携程Apollo系统，一个配置中心系统，使用AP模型的eureka作为配置的服务发现系统，客户端和服务端通过springmvc的DeferredResult加超时时间的长轮询机制实现服务实时通知。

1、客户端发起一个请求

2、服务端通过DeferredResult异步方式，hold住这个请求的同时释放容器线程

3、如果有变更，直接返回DeferredResult结果

4、如果没有变更，DeferredResult通过超时时间，等待60s，通过onTimeout记得把任务移除，给客户端返回超时，客户端重新发起请求。

5、如果开始无变更，hold住请求后发生了变更，DeferredResult有个map集合，变更的同时看看这个集合是不是有请求变更对应appid系统配置的，有的话，从map中取出DeferredResult，并通过setResult返回结果给客户端。

@RestController
    public class TaskController {
        private static final Logger log = LoggerFactory.getLogger(TaskController.class);
        //超时结果
        private static final ResponseMsg<String> OUT_OF_TIME_RESULT = new ResponseMsg<>(-1,"超时","out of time");
     
        //超时时间
        private static final long OUT_OF_TIME = 3000L;
        @Autowired
        private TaskQueue taskQueue;
     
        @RequestMapping(value = "/get",method = RequestMethod.GET)
        public DeferredResult<ResponseMsg<String>> getResult() {
            //建立DeferredResult对象，设置超时时间，以及超时返回超时结果
            DeferredResult<ResponseMsg<String>> result = new DeferredResult<>(OUT_OF_TIME, OUT_OF_TIME_RESULT);
            result.onTimeout(() -> {
                log.info("调用超时");
            });
            result.onCompletion(() -> {
                log.info("调用完成");
            });
            //并发，加锁
            synchronized (taskQueue) {
                taskQueue.put(result);
            }
            return result;
        }
    }

@Component
    public class TaskExecute {
     
        private static final Logger log = LoggerFactory.getLogger(TaskExecute.class);
     
        private static final Random random = new Random();
     
        //默认随机结果的长度
        private static final int DEFAULT_STR_LEN = 10;
     
        //用于生成随机结果
        private static final String str = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
     
        @Autowired
        private TaskQueue taskQueue;
     
     
        /**
         * 初始化启动
         */
        @PostConstruct
        public void init() {
     
            log.info("开始持续处理任务");
     
            new Thread(this::execute).start();
     
     
        }
     
     
        /**
         * 持续处理
         * 返回执行结果
         */
        private void execute() {
     
            while (true) {
     
                try {
     
                    //取出任务
                    Task task;
     
                    synchronized (taskQueue) {
     
                        task = taskQueue.take();
     
                    }
     
                    if (task != null) {
     
                        //设置返回结果
                        String randomStr = getRandomStr(DEFAULT_STR_LEN);
     
                        ResponseMsg<String> responseMsg = new ResponseMsg<String>(0, "success", randomStr);
     
                        log.info("返回结果:{}", responseMsg);
     
                        task.getTaskResult().setResult(responseMsg);
                    }
     
                    int time = random.nextInt(10);
     
                    log.info("处理间隔：{}秒", time);
     
                    Thread.sleep(time * 1000L);
     
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
     
     
            }
     
        }