同步设备

meter-reading-job/src/main/java/com/huaxu/zoniot/service/impl/MeterFileSyncServiceImpl.java

@@ -25,6 +25,8 @@ import javax.annotation.Resource;
 import java.nio.charset.Charset;
 import java.util.*;
 
+import static com.google.common.collect.Lists.newArrayList;
+
 /**
  * <p></p>
  *
@@ -328,10 +330,43 @@ public class MeterFileSyncServiceImpl  implements MeterFileSyncService {
     }
 
     private void sendRabbitData(List<DeviceData> dataList) {
-        for (DeviceData data : dataList) {
-            String msg = JSON.toJSONString(data);
-            log.info("transfer mq msg:{}",msg);
-            rabbitTemplate.convertAndSend(exchange, queue,msg.getBytes(Charset.forName("UTF-8")));
+
+        if (dataList != null && dataList.size() > 0) {
+            int total = dataList.size();
+            int pageSize = 5000;
+            int totalPage = total / pageSize;
+            if (total % pageSize != 0) {
+                totalPage++;
+            }
+            for (int i = 1; i <= totalPage; i++)
+            {
+                log.info("current page = " + i);
+                List<DeviceData> records = newArrayList();
+                int start = (i - 1) * pageSize;
+                int j = 0;
+
+                for (int index = 0; (j < pageSize) && (index < total); j++)
+                {
+                    index = start + j;
+                    records.add(dataList.get(index));
+                    index++;
+                }
+                log.info("current page size = " + records.size());
+
+                for (DeviceData data : records) {
+                    String msg = JSON.toJSONString(data);
+                    log.info("transfer mq msg:{}",msg);
+                    rabbitTemplate.convertAndSend(exchange, queue,msg.getBytes(Charset.forName("UTF-8")));
+                }
+
+                try {
+                    Thread.sleep(30000);
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
+                }
+            }
         }
+
+
     }
 }