2019/1/7 backup 针对备用es调整读取策略

README.md

+# messageflow-监测系统消息流模块
+
+标签（空格分隔）：readme
+
+---
+
+##介绍项目
+监测系统消息流模块，主要功能：消息流、定向监测消息流、自动标注、事件采集
+
+---
+##项目文档地址 
+
+>参考监测系统文档
+
+---
+##项目运行的服务器及目录
+> 项目地址：/home/zhiwei/jar/messageflow
+ 服务器地址：
+  内网地址：192.168.0.2
+   外网地址：115.236.59.87
+
+
+
+

src/main/java/com/zhiwei/messageflow/ES4RedisRunner.java

@@ -61,7 +61,7 @@ public class ES4RedisRunner implements ApplicationRunner {
 								log.error("主定时器异常{}{}",e.getMessage(),e.getStackTrace());
 							}
 						}
-					}, 60*1000L, 50 * 1000L);
+					}, 100L, 50 * 1000L);
 
 			
 			

src/main/java/com/zhiwei/messageflow/ES4RedisStart.java

@@ -2,7 +2,9 @@ package com.zhiwei.messageflow;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.LogManager;
@@ -15,6 +17,7 @@ import com.zhiwei.messageflow.mongo.bean.PlatformNew;
 import com.zhiwei.messageflow.mongo.bean.Project;
 import com.zhiwei.messageflow.mongo.dao.PlatformDao;
 import com.zhiwei.messageflow.mongo.dao.ProjectDao;
+import com.zhiwei.messageflow.util.Tools;
 
 @Component
 public class ES4RedisStart {
@@ -29,44 +32,72 @@ public class ES4RedisStart {
 
 	@Autowired
 	private ES4RedisTask es4RedisTask;
-	
+
 	@Autowired
 	private ESGetCommonId esGetCommonId;
 
+	private Map<String, Boolean> PROJECTMAP = new HashMap<>();
+
+	private static final int PROJECTCOUNT = 9;
+
 	/**
 	 * 启动线程
 	 */
 	public void startThread() throws JsonParseException, JsonMappingException, IOException, InterruptedException {
-		//更新commonid
-		esGetCommonId.getCommonId();		
+		// 更新commonid
+		esGetCommonId.getCommonId();
 		// 项目列表
 		List<Project> projects = projectDao.getAllProjects();
 
+		// 本轮项目计数
+		int pcount = 0;
+		//new Map
+		if (Tools.isEmpty(PROJECTMAP.keySet())||isProjectMapFull()) {
+			projects.forEach(p -> {
+				PROJECTMAP.put(p.getProjectName(), false);
+			});
+			PROJECTMAP.keySet().forEach(p -> {
+				System.out.println(p+"\t"+PROJECTMAP.get(p));
+			});
+		}else {
+			PROJECTMAP.keySet().forEach(p -> {
+				System.out.println(p+"\t"+PROJECTMAP.get(p));
+			});
+		}
+
 		// 公共平台名列表
 //		List<String> platformNames = platformDao.getPublicPlatformName();
 
 		// 遍历项目
 		for (Project project : projects) {
-			
+
 //			if(!project.getProjectName().equals("测试")) {
 //				continue;
 //			}
+			if (pcount >= PROJECTCOUNT) {
+				continue;
+			}
+
+			if (!canProjectRead(project)) {
+				continue;
+			} else {
+				pcount++;
+			}
 
 			/**
 			 * 项目全部平台(公共+私有)
 			 */
 			List<PlatformNew> allplatformNames = platformDao.findPlatformByProject(project);
 //			System.out.println(allplatformNames);
-			
+
 //			List<String> allplatformNames = new ArrayList<>();
 //			allplatformNames.addAll(platformNames);
 //			allplatformNames.addAll(project.getDataPt());
-			
 
 			// 获取线程
 			ES4RedisThreadNew es4RedisThread = ES4RedisThreadNew.getThread(project.getProjectName(), project,
 					allplatformNames, es4RedisTask);
-			
+
 //			ES4RedisThread es4RedisThread = ES4RedisThread.getThread(project.getProjectName(), project,
 //					allplatformNames, es4RedisTask);
 
@@ -80,4 +111,33 @@ public class ES4RedisStart {
 			es4RedisThread.start();
 		}
 	}
+	/**
+	 * 判断项目map是否一轮结束
+	 * @param projects 
+	 * @Title: isProjectMapFull 
+	 * @Description: 判断项目map是否一轮结束
+	 * @param @return 设定文件 
+	 * @return boolean 返回类型
+	 */
+	private boolean isProjectMapFull() {
+		return PROJECTMAP.values().parallelStream().allMatch(v -> v);
+	}
+
+	/**
+	 * 判断项目本轮内是否可以获取消息流
+	 * 
+	 * @Title: canProjectRead
+	 * @Description: 判断项目本轮内是否可以获取消息流
+	 * @param @param project
+	 * @param @return 设定文件
+	 * @return boolean 返回类型
+	 */
+	private boolean canProjectRead(Project project) {
+		if(PROJECTMAP.get(project.getProjectName())) {
+			return false;
+		}else {
+			PROJECTMAP.put(project.getProjectName(), true);
+			return true;
+		}
+	}
 }