作者 王勇

添加多线程,获取出港数据统计

@@ -5,11 +5,15 @@ import com.sunyo.wlpt.cgonms.provide.feign.GetCgoAsmFeign; @@ -5,11 +5,15 @@ import com.sunyo.wlpt.cgonms.provide.feign.GetCgoAsmFeign;
5 import com.sunyo.wlpt.cgonms.provide.feign.GetDataWareHouseFeign; 5 import com.sunyo.wlpt.cgonms.provide.feign.GetDataWareHouseFeign;
6 import com.sunyo.wlpt.cgonms.provide.feign.GetTransportFeign; 6 import com.sunyo.wlpt.cgonms.provide.feign.GetTransportFeign;
7 import com.sunyo.wlpt.cgonms.provide.service.*; 7 import com.sunyo.wlpt.cgonms.provide.service.*;
  8 +import com.sunyo.wlpt.cgonms.provide.thread.ExitThreadPoolFactory;
  9 +import lombok.extern.slf4j.Slf4j;
8 import org.springframework.web.bind.annotation.*; 10 import org.springframework.web.bind.annotation.*;
9 11
10 import javax.annotation.Resource; 12 import javax.annotation.Resource;
  13 +import java.text.SimpleDateFormat;
11 import java.util.Date; 14 import java.util.Date;
12 import java.util.List; 15 import java.util.List;
  16 +import java.util.concurrent.*;
13 17
14 /** 18 /**
15 * @author 子诚 19 * @author 子诚
  1 +package com.sunyo.wlpt.cgonms.provide.controller;
  2 +
  3 +import com.google.common.collect.Lists;
  4 +
  5 +import com.sunyo.wlpt.cgonms.provide.domain.*;
  6 +import com.sunyo.wlpt.cgonms.provide.feign.GetCgoAsmFeign;
  7 +import com.sunyo.wlpt.cgonms.provide.feign.GetDataWareHouseFeign;
  8 +import com.sunyo.wlpt.cgonms.provide.feign.GetTransportFeign;
  9 +import com.sunyo.wlpt.cgonms.provide.service.*;
  10 +import com.sunyo.wlpt.cgonms.provide.thread.ExitThreadPoolFactory;
  11 +import lombok.extern.slf4j.Slf4j;
  12 +import org.springframework.web.bind.annotation.*;
  13 +
  14 +import javax.annotation.Resource;
  15 +import java.text.SimpleDateFormat;
  16 +import java.util.Date;
  17 +import java.util.List;
  18 +import java.util.concurrent.CountDownLatch;
  19 +import java.util.concurrent.ThreadPoolExecutor;
  20 +
  21 +/**
  22 + * @author 子诚
  23 + * Description:多线程获取出港数据统计
  24 + * 时间:2020/5/26 21:18
  25 + */
  26 +@CrossOrigin
  27 +@Slf4j
  28 +@RequestMapping("cgoNms")
  29 +@RestController
  30 +public class NmsController {
  31 + @Resource
  32 + private ResultExitDataService resultExitDataService;
  33 +
  34 + @Resource
  35 + private PrepareMasterService prepareMasterService;
  36 +
  37 + @Resource
  38 + private BasAirportService basAirportService;
  39 +
  40 + @Resource
  41 + private TbAirportCodeService tbAirportCodeService;
  42 +
  43 + @Resource
  44 + private BasAreaService basAreaService;
  45 +
  46 + @Resource
  47 + private GetCgoAsmFeign getCgoAsmFeign;
  48 +
  49 + @Resource
  50 + private GetTransportFeign getTransportFeign;
  51 +
  52 + @Resource
  53 + GetDataWareHouseFeign getDataWareHouseFeign;
  54 +
  55 + /**
  56 + * 线程数量
  57 + */
  58 + private final static int THREAD_ACCOUNT = 50;
  59 +
  60 + /**
  61 + * 开启多线程
  62 + */
  63 + ThreadPoolExecutor threadPool = ExitThreadPoolFactory.instance();
  64 +
  65 + /**
  66 + * 获取开始时间,结束时间
  67 + */
  68 + final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  69 +
  70 +
  71 + /**
  72 + * 根据航班日期、航班号,获取出出港业务统计数据
  73 + *
  74 + * @param flightDate 航班日期
  75 + * @param flightNo 航班号
  76 + * @return {@link ResultExitData}
  77 + */
  78 + @GetMapping("/getInfo")
  79 + public List<ResultExitData> getData(
  80 + @RequestParam(value = "flightDate", required = false) Date flightDate,
  81 + @RequestParam(value = "flightNo", required = false) String flightNo) {
  82 +
  83 + final String startTime = sdf.format(new Date());
  84 + System.out.println("开始时间:" + startTime);
  85 + System.out.println("出港数据,开始获取");
  86 + /**
  87 + * 查询出新舱单里面回执表里面的数据条件
  88 + */
  89 + ResultExitData resultExitData = new ResultExitData();
  90 + resultExitData.setFlightDate(flightDate);
  91 + resultExitData.setFlightNo(flightNo);
  92 + List<ResultExitData> resultList = resultExitDataService.getResultExitDataInfo(resultExitData);
  93 + System.out.println("长度为:" + resultList.size());
  94 + //控制index,查询数据的时候,保证索引的位置不出错
  95 + int index = 0;
  96 + try {
  97 + // 如果数据大于50个,则每次获取前50条。逐次获取
  98 + if (resultList != null && !resultList.isEmpty() && resultList.size() > THREAD_ACCOUNT) {
  99 + while (index < resultList.size()) {
  100 + //如果剩余未处理的数量 大于 等于线程数量,则遍历次数为线程数量
  101 + if (resultList.size() - index >= THREAD_ACCOUNT) {
  102 + CountDownLatch latch = new CountDownLatch(THREAD_ACCOUNT);
  103 + log.trace("获取开始{},剩余数量:{}", startTime, resultList.size());
  104 + int i;
  105 + for (i = 0; i < THREAD_ACCOUNT; i++) {
  106 + ResultExitData result = threadJob(resultList.get(i + index), latch, threadPool);
  107 + resultList.set(i, result);
  108 + }
  109 + index = index + i;
  110 + //完成一次,就等待。每次减1,为0的时候往下执行
  111 + latch.await();
  112 + }
  113 + //如果剩余未处理的数量小于线程数量,则遍历次数为
  114 + else if (resultList.size() - index < THREAD_ACCOUNT) {
  115 + CountDownLatch latch = new CountDownLatch(resultList.size() - index);
  116 + log.trace("获取开始{},剩余数量:{}", startTime, resultList.size());
  117 + int i;
  118 + for (i = 0; i < resultList.size() - index; i++) {
  119 + ResultExitData result = threadJob(resultList.get(i + index), latch, threadPool);
  120 + resultList.set(i, result);
  121 + }
  122 + index = index + i;
  123 + //完成一次,就等待。每次减1,为0的时候往下执行
  124 + latch.await();
  125 + }
  126 + }
  127 + }
  128 + // 如果数据小于等于50个,则每次获取前50条。逐次获取
  129 + else if (resultList != null && !resultList.isEmpty() && resultList.size() <= THREAD_ACCOUNT) {
  130 + CountDownLatch latch = new CountDownLatch(resultList.size());
  131 + log.trace("获取开始{},剩余数量:{}", startTime, resultList.size());
  132 + for (int i = 0; i < resultList.size(); i++) {
  133 + ResultExitData result = threadJob(resultList.get(i), latch, threadPool);
  134 + resultList.set(i, result);
  135 + }
  136 + //完成一次,就等待。等所有的全部完成,再一起返回
  137 + latch.await();
  138 + }
  139 + } catch (Exception e) {
  140 + e.printStackTrace();
  141 + log.error("获取目录文件出错", e);
  142 + }
  143 +
  144 + final String endTime = sdf.format(new Date());
  145 + System.out.println("结束时间:" + endTime);
  146 + System.out.println("出港信息,获取完毕");
  147 + System.out.println("index的值为:" + index);
  148 + return resultList;
  149 + }
  150 +
  151 + /**
  152 + * 使用线程获取数据.
  153 + *
  154 + * @param result 返回数据
  155 + * @param latch 计数器
  156 + * @param threadPool 线程池
  157 + * @return
  158 + */
  159 + private ResultExitData threadJob(ResultExitData result, CountDownLatch latch, ThreadPoolExecutor threadPool) {
  160 + Runnable run = new Runnable() {
  161 + @Override
  162 + public void run() {
  163 +
  164 + //设置航班
  165 + result.setSegment(result.getOriginatingStation() + " — " + result.getDestinationStation());
  166 +
  167 + //设置承运人二字码
  168 + result.setCarrier(result.getFlightNo().substring(0, 2));
  169 +
  170 + /**
  171 + * 3.根据 理货信息 中的目的站代码,获取 新舱单机场代码表 中的相应数据
  172 + *
  173 + * 机场所属城市代码
  174 + */
  175 + BasAirport ba = basAirportService.getBasAirportInfo(result.getDestinationStation());
  176 +
  177 + /**
  178 + * 4.根据 新舱单机场代码表 中的 机场所属城市代码 ,获取 天博——机场代码表 的相应数据
  179 + *
  180 + * 城市代码
  181 + * 区域id
  182 + * 国家代码
  183 + */
  184 + TbAirportCode ta = tbAirportCodeService.getTbAirportCodeInfo(ba.getCityId());
  185 + // 所属国家代码
  186 + result.setCountry(ta.getCountry());
  187 +
  188 + /**
  189 + * 5.根据 天博——机场代码表 中的 区域id,获取 新舱单——区域代码表 的相应数据
  190 + *
  191 + * 区域id
  192 + * 国际国内标识;1,国外;2,国内
  193 + * 区域描述,中文
  194 + */
  195 + BasArea basArea = basAreaService.getBasAreaInfo(ta.getArea());
  196 + //航班目的区域,中文描述,所属洲
  197 + result.setAreaDescChn(basArea.getAreaDescChn());
  198 +
  199 + /**
  200 + * 6.根据 理货信息 中的运单号,获取预配表中的相应数据
  201 + *
  202 + * 发货人名称
  203 + * 收货人名称
  204 + * 航班目的站
  205 + */
  206 + List<PrepareMaster> pm = prepareMasterService.getPrepareMasterInfo(result.getWaybillNoMaster());
  207 + if (pm != null && pm.size() > 0) {
  208 + //发货人名称
  209 + result.setShipperName(pm.get(0).getShipperName());
  210 + //收货人名称
  211 + result.setConsigneeName(pm.get(0).getConsigneeName());
  212 + //获取目的货物目的站
  213 + result.setAimStation(pm.get(0).getDestinationStation());
  214 + }
  215 + /**
  216 + * 获取数据仓库的对应数据
  217 +
  218 + // if (result.getWaybillNoMaster() != null && result.getWaybillNoMaster() != "") {}
  219 + ResultExitData dataWareHouseInfo=new ResultExitData();
  220 + dataWareHouseInfo = getDataWareHouseFeign.getInfo(result.getWaybillNoMaster());
  221 + //设置体积
  222 + result.setVolumeMeasure(dataWareHouseInfo.getVolumeMeasure());
  223 + //设置计费重量
  224 + result.setTotalGrossWeightMeasure(dataWareHouseInfo.getTotalGrossWeightMeasure());
  225 + */
  226 +
  227 + /**
  228 + * 获取运输工具的对应数据
  229 + */
  230 + ResultExitData transportInfo = new ResultExitData();
  231 + transportInfo = getTransportFeign.getInfo(result.getFlightNo(), result.getFlightDate());
  232 + //设置航班公司
  233 + result.setAirCompany(transportInfo.getAirCompany());
  234 + //设置机型
  235 + result.setCfTp(transportInfo.getCfTp());
  236 + //设置机号
  237 + result.setCfNo(transportInfo.getCfNo());
  238 + //设置航班计划日期
  239 + result.setFlightPlanDate(transportInfo.getFlightPlanDate());
  240 + //设置航班(起飞)时间
  241 + result.setFlightTime(transportInfo.getFlightTime());
  242 + //设置航班计划时间
  243 + result.setFlightPlanTime(transportInfo.getFlightPlanTime());
  244 +
  245 + /**
  246 + * 获取代理人的相关数据
  247 + */
  248 +// if (result.getWaybillNoMaster() != null && result.getWaybillNoMaster() != "") { }
  249 + ResultExitData asmInfo = new ResultExitData();
  250 + asmInfo = getCgoAsmFeign.getInfo(result.getWaybillNoMaster());
  251 + //设置代理人全称
  252 + result.setFullName(asmInfo.getFullName());
  253 + //设置货主类型
  254 + result.setTheShipperType(asmInfo.getTheShipperType());
  255 + //设置品名
  256 + result.setSdCargoName(asmInfo.getSdCargoName());
  257 + //设置二级类名称
  258 + result.setTwoTypeName(asmInfo.getTwoTypeName());
  259 + //设置一级类名称
  260 + result.setTypeName(asmInfo.getTypeName());
  261 + latch.countDown();
  262 + }
  263 + };
  264 + threadPool.execute(run);
  265 + return result;
  266 + }
  267 +
  268 +
  269 +}
  1 +package com.sunyo.wlpt.cgonms.provide.thread;
  2 +
  3 +import java.util.ArrayList;
  4 +import java.util.Date;
  5 +import java.util.Iterator;
  6 +import java.util.List;
  7 +import java.util.concurrent.ThreadFactory;
  8 +
  9 +/**
  10 + * @author 子诚
  11 + * Description:创建线程工厂
  12 + * 时间:2020/5/26 17:57
  13 + */
  14 +public class ExitThreadFactory implements ThreadFactory {
  15 +
  16 + private int counter;
  17 + private String name;
  18 + private List<String> stats;
  19 +
  20 + public ExitThreadFactory(String name)
  21 + {
  22 + counter = 1;
  23 + this.name = name;
  24 + stats = new ArrayList<String>();
  25 + }
  26 +
  27 + @Override
  28 + public Thread newThread(Runnable runnable)
  29 + {
  30 + Thread t = new Thread(runnable, name + "-Thread_" + counter);
  31 + counter++;
  32 + stats.add(String.format("Created thread %d with name %s on %s \n", t.getId(), t.getName(), new Date()));
  33 + return t;
  34 + }
  35 +
  36 + public String getStats()
  37 + {
  38 + StringBuffer buffer = new StringBuffer();
  39 + Iterator<String> it = stats.iterator();
  40 + while (it.hasNext())
  41 + {
  42 + buffer.append(it.next());
  43 + }
  44 + return buffer.toString();
  45 + }
  46 +
  47 +}
  48 +
  1 +package com.sunyo.wlpt.cgonms.provide.thread;
  2 +
  3 +import java.util.concurrent.LinkedBlockingQueue;
  4 +import java.util.concurrent.ThreadPoolExecutor;
  5 +import java.util.concurrent.TimeUnit;
  6 +
  7 +/**
  8 + * @author 子诚
  9 + * Description:创建线程池
  10 + * 时间:2020/5/26 17:51
  11 + */
  12 +public class ExitThreadPoolFactory {
  13 + private static ThreadPoolExecutor threadPool;
  14 +
  15 + public static ThreadPoolExecutor instance() {
  16 + if (threadPool == null) {
  17 + ExitThreadFactory exitThreadFactory = new ExitThreadFactory("exit");
  18 + threadPool = new ThreadPoolExecutor(10, 100,
  19 + 0L, TimeUnit.MILLISECONDS,
  20 + new LinkedBlockingQueue<Runnable>(1024),
  21 + exitThreadFactory,
  22 + new ThreadPoolExecutor.AbortPolicy());
  23 + }
  24 + return threadPool;
  25 + }
  26 +}