Commit 3e02e6978e7df0608d05fa2153cb7a3b3b74ed9c
1 parent
d559b01b
megr .....
Showing
1 changed file
with
17 additions
and
8 deletions
src/main/java/com/bsth/data/gpsdata/arrival/GpsRealAnalyse.java
| @@ -41,7 +41,8 @@ public class GpsRealAnalyse { | @@ -41,7 +41,8 @@ public class GpsRealAnalyse { | ||
| 41 | @Autowired | 41 | @Autowired |
| 42 | GpsRealData gpsRealData; | 42 | GpsRealData gpsRealData; |
| 43 | 43 | ||
| 44 | - static ExecutorService threadPool = Executors.newFixedThreadPool(25); | 44 | + final static int POOL_SIZE = 20; |
| 45 | + static ExecutorService threadPool = Executors.newFixedThreadPool(POOL_SIZE + 1); | ||
| 45 | 46 | ||
| 46 | public static long st; | 47 | public static long st; |
| 47 | public static CountDownLatch count; | 48 | public static CountDownLatch count; |
| @@ -60,19 +61,27 @@ public class GpsRealAnalyse { | @@ -60,19 +61,27 @@ public class GpsRealAnalyse { | ||
| 60 | return; | 61 | return; |
| 61 | st = System.currentTimeMillis(); | 62 | st = System.currentTimeMillis(); |
| 62 | 63 | ||
| 63 | - //按线路分组gps | 64 | + //按设备号分组数据（一个设备号的多条数据，必须在一个线程里跑） |
| 64 | ArrayListMultimap multimap = ArrayListMultimap.create(); | 65 | ArrayListMultimap multimap = ArrayListMultimap.create(); |
| 65 | for (GpsEntity gps : list) { | 66 | for (GpsEntity gps : list) { |
| 66 | - multimap.put(gps.getLineId(), gps); | 67 | + multimap.put(gps.getDeviceId(), gps); |
| 67 | } | 68 | } |
| 68 | - | ||
| 69 | - Set<String> ks = multimap.keySet(); | ||
| 70 | - | 69 | + List<String> deviceList = new ArrayList<>(multimap.keySet()); |
| 70 | + | ||
| 71 | + //数据均分给20个线程 | ||
| 72 | + ArrayListMultimap dataListMap = ArrayListMultimap.create(); | ||
| 73 | + int size = deviceList.size(), threadIndex=0, threadSize = size / POOL_SIZE; | ||
| 74 | + for(int i = 0; i < size; i++){ | ||
| 75 | + if(i % threadSize == 0) | ||
| 76 | + threadIndex ++; | ||
| 77 | + dataListMap.putAll(threadIndex, multimap.get(deviceList.get(i))); | ||
| 78 | + } | ||
| 79 | + Set<Integer> ks = dataListMap.keySet(); | ||
| 71 | logger.info("analyse gps size: " + list.size() + ", ks: " + ks.size()); | 80 | logger.info("analyse gps size: " + list.size() + ", ks: " + ks.size()); |
| 72 | count = new CountDownLatch(ks.size()); | 81 | count = new CountDownLatch(ks.size()); |
| 73 | 82 | ||
| 74 | - for (String lineCode : ks) { | ||
| 75 | - threadPool.execute(new SignalHandleThread(multimap.get(lineCode), count)); | 83 | + for (Integer index : ks) { |
| 84 | + threadPool.execute(new SignalHandleThread(dataListMap.get(index), count)); | ||
| 76 | } | 85 | } |
| 77 | 86 | ||
| 78 | //等待子线程结束 | 87 | //等待子线程结束 |