1.加入kafka传电子路单至数据中台

王通
1 parent fab5a228
Showing 1 changed file with 25 additions and 10 deletions
src/main/java/com/bsth/server_rs/schedule/real/ScheduleRealService.java
@@ -69,6 +69,8 @@ public class ScheduleRealService implements InitializingBean {
     private volatile long timestamp;
+    private final static int KAFKA_BATCH_SIZE = 100;
+
     static {
         secretKey = ConfigUtil.get("http.control.secret.key");
         url = ConfigUtil.get("http.control.service_data_url");// + "/execSchList?secretKey=" + secretKey;
@@ -302,22 +304,35 @@ public class ScheduleRealService implements InitializingBean {
                     List<ScheduleRealInfo> scheduleRealInfos = scheduleRealInfoRepository.findByDates(dates, new Date(timestamp));
                     timestamp = dateTime.getMillis();
-                    List<ScheduleRealInfoVo> scheduleRealInfoVos = new ArrayList<>();
+                    Queue<ScheduleRealInfoVo> queue = new ConcurrentLinkedDeque<>();
                     for (ScheduleRealInfo sch : scheduleRealInfos) {
                         ScheduleRealInfoVo scheduleRealInfoVo = new ScheduleRealInfoVo();
                         BeanUtils.copyProperties(sch, scheduleRealInfoVo);
                         Car car = CarBufferData.findOne(scheduleRealInfoVo.getClZbh());
                         scheduleRealInfoVo.setCarPlate(car == null ? "" : car.getCarPlate());
-                        scheduleRealInfoVos.add(scheduleRealInfoVo);
+                        queue.add(scheduleRealInfoVo);
+                    }
+
+                    int size = scheduleRealInfos.size();
+                    for (int i = 0, len = size % KAFKA_BATCH_SIZE == 0 ? size / KAFKA_BATCH_SIZE : size / KAFKA_BATCH_SIZE + 1;i < len;i++) {
+                        List<ScheduleRealInfoVo> scheduleRealInfoVos = new ArrayList<>();
+                        for (int j = 0; j < KAFKA_BATCH_SIZE;j++) {
+                            ScheduleRealInfoVo scheduleRealInfoVo = queue.poll();
+                            if (scheduleRealInfoVo == null) {
+                                break;
+                            }
+                            scheduleRealInfoVos.add(scheduleRealInfoVo);
+                        }
+
+                        Map<String, Object> data = new HashMap<>();
+                        data.put("datatype", "waybill");
+                        data.put("datas", scheduleRealInfoVos);
+
+                        ObjectMapper mapper = new ObjectMapper();
+                        String json = mapper.writeValueAsString(data);
+                        logger.info(json);
+                        kafkaTemplate.send("PDGJ_JQDD", json);
                     }
-                    Map<String, Object> data = new HashMap<>();
-                    data.put("datatype", "waybill");
-                    data.put("datas", scheduleRealInfoVos);
-
-                    ObjectMapper mapper = new ObjectMapper();
-                    String json = mapper.writeValueAsString(data);
-                    logger.info(json);
-                    kafkaTemplate.send("PDGJ_JQDD", json);
                 } catch (Exception e) {
                     logger.error("kafka发电子路单调度异常", e);
                 }