springboot + vue+ redis 多线程查询几个外部系统，数据量大崩溃问题完美解决。

springboot + vue+ redis 多线程查询几个外部系统，数据量大崩溃问题完美解决。

springboot + vue+ redis 多线程调用

• 需求描述
• vue 前端
• 后端开发
• • 1. 异步设置
  • 2.controller
  • 3. service
  • 4.serviceImpl

需求描述

自己的系统要接入几个外部系统的数据，即同一个查询条件需要同时从不同的局域网系统中获取数据并汇总到当前系统中展示。数据量大的时候，查询经常会崩掉。

解决办法是前端调用后端服务时，立刻返回一个任务id（因为前端响应时间大概是5s,超过5s就报错，立刻返回不会让用户有错觉），同时设置一个定时器setInterval()，根据返回的任务id按时查询redis缓存的进度状态，如果缓存状态为ongoing ，就继续定时任务，如果为end,则查询redis中返回的结果，并停止定时任务clearInterval()。如果为error,则停止定时任务clearInterval()，并通知用户任务出错。

后端通过多线程异步调用各个外部系统数据，进行中时设置redis的key为任务id，并设置进度状态为ongoing，出错时设置redis的key为任务id，进度状态设置为error，查询完成后设置redis的key为任务id，进度状态为end。 同时将结果缓存到redis中，这个redis的key可以设置为任务id + “:result” 同状态区分一下。

vue 前端

		watch: {
            'listLoading': function(newVal, oldVal) {
                if (newVal && !oldVal) {
                    let _this = this;
                    this.getProgressTimer = setInterval(async ()=>{
                        try {
                            let resp = await  getZbProgress(this.zbcTaskId);//查询进度
                            let progress = resp.bodyText;
                            if (progress == 'ongoing'){ //redis返回进度为ongoing 
                                this.$message.error('数据正在获取中...');
                            }else if (progress == 'end'){ //redis返回进度为end
                                if (_this.getProgressTimer) {
                                    clearInterval(_this.getProgressTimer); //停止定时任务
                                    _this.getProgressTimer = null;
                                }
                                let result = await getZbccrkResult(this.zbcTaskId); //查询结果
                                _this.formateTableData(result);//返回的结果渲染 并将动画设置为false
                            }else if (progress == 'error'){//redis返回进度为error
                                if (_this.getProgressTimer) {
                                    clearInterval(_this.getProgressTimer);//停止定时任务
                                    _this.getProgressTimer = null;
                                }
                                this.zbcTaskId = '';
                                this.$message.error('数据获取出错！');
                            }
                        } catch (e) {
                            if (_this.getProgressTimer) {
                                clearInterval(_this.getProgressTimer);//停止定时任务
                                _this.getProgressTimer = null;
                            }
                            this.zbcTaskId = '';
                            this.$message.error('数据获取出错！');
                        }
                    }, 1000);
                } else if (!newVal && oldVal) {
                    if (this.getProgressTimer) {
                        clearInterval(this.getProgressTimer);//停止定时任务
                        this.getProgressTimer = null;
                    }
                    this.zbcTaskId = '';
                }
            }
        },
		mounted() {
          this.queryZbInfo();
        },
        methods: {
            formateTableData(result){
                this.tableData = result.body;
                this.listLoading = false;
            },
     
            // 查询信息
            queryZbInfo: function (flg) {
                var data = {
                    zbc: this.zbcName ? this.zbcName : '',
                    cx : this.cx,
                    ch : this.ch,
                    kssj: this.kssj,
                    jssj: this.jssj,
                    jcnr: this.jcnr
                };
                zbToInOut(data).then(
                    (response) => {
                        this.zbcTaskId = response.bodyText;//获取的任务id
                        // 开启动画并监听
                        this.listLoading = true; 
                    },
                    (response) => {
                        this.$message.error('获取数据失败!');
                        this.listLoading = false;
                    }
                )
            },
        }

后端开发

参考的博文https://zhuanlan.zhihu.com/p/134636915

1. 异步设置

工程的application中添加@EnableAsync；
添加async的配置文件AsyncConfiguration 。

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

@Configuration
@EnableAsync
public class AsyncConfiguration {

    @Bean("zbcrkExecutor")
    public Executor doSomethingExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 核心线程数：线程池创建时候初始化的线程数
        executor.setCorePoolSize(10);
        // 最大线程数：线程池最大的线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程
        executor.setMaxPoolSize(20);
        // 缓冲队列：用来缓冲执行任务的队列
        executor.setQueueCapacity(500);
        // 允许线程的空闲时间60秒：当超过了核心线程之外的线程在空闲时间到达之后会被销毁
        executor.setKeepAliveSeconds(60);
        // 线程池名的前缀：设置好了之后可以方便我们定位处理任务所在的线程池
        executor.setThreadNamePrefix("zbc-do-something-");
        // 缓冲队列满了之后的拒绝策略：由调用线程处理（一般是主线程）
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
        executor.initialize();
        return executor;
    }

}

2.controller

因为我的多线程调用是通过for循环进行，所以调用放在contrller中。

   @PostMapping("/zbToJx/returnJccrk")
    public String selectZbToJxJccrk(@RequestBody final CheckInOut checkinout) throws Exception{
        String taskId = "zbccrk_Task_" + new Long((new Date()).getTime()).toString();//任务id
        try{
            List zbcList = repStaFeign.getAllZbc();
            if (StringUtils.isEmpty(checkinout.getZbc())) {              
                List> comList = Lists.newArrayList();
                this.redisUtil.set(taskId,"ongoing"); //设置进度状态为进行中
                for (int i = 0; i < zbcList.size(); i++) {
                    CompletableFuture createOrder = reService.selectZbToJxJccrk(checkinout ,zbcList.get(i));//多线程查询数据
                    comList.add(createOrder);//保存每个系统返回的json数据
                }
                List zbccrks = Lists.newArrayList();//保存所有系统的查询结果
                for (CompletableFuture com : comList){
                    CompletableFuture.allOf(com).join();                 
                    JSONArray jsonArray = new JSONArray(com.get()); //获取的数据为json
                    List zbccrkList = jsonArray.toList(Zbccrk.class);//转成需要的list对象
                    zbccrks.addAll(zbccrkList);
                }
                this.redisUtil.set(taskId,"end"); //设置进度状态为结束           
                this.redisUtil.set(taskId+":zbcResult",zbccrks);//缓存查询结果             
            } 
        }catch(Exception e){
            e.printStackTrace();
            this.redisUtil.set(taskId,"error");//设置进度状态为出错
        }
        return taskId;
    }

3. service

CompletableFuture selectZbToJxJccrk(CheckInOut checkinout, Zbc zbc) throws Exception;

4.serviceImpl

多线程调用

	private ToZbFeignDynamicUrl toZbFeignDynamicUrl;

    @Override
    @Async("zbcrkExecutor")
    public CompletableFuture selectZbToJxJccrk(CheckInOut checkinout, Zbc zbc) throws Exception {
        // 配置feign
        toZbFeignDynamicUrl = Feign.builder()
                .encoder(new FeignEncoder())
                .decoder(new FeignStringDecoder())
                .target(ToZbFeignDynamicUrl.class, zbc.getUrl());     
        String result = toZbFeignDynamicUrl.selectZb(checkinout).getBody();
        return CompletableFuture.completedFuture(result);
    }

ToZbFeignDynamicUrl .java

@FeignClient(name = "zb-service-dynamic-url")
public interface ToZbFeignDynamicUrl {
    @RequestLine("POST /reportwork/zbToJx/returnJccrk")
    @Headers({"Content-Type: application/json"})
    ResponseEntity selectZb(@RequestBody CheckInOut checkinout) throws Exception;
}