将对上级的汇总用队列机制进行集中批量处理

xm-core/src/main/java/com/xm/core/ctrl/XmMenuController.java

@@ -15,6 +15,7 @@ import com.xm.core.PubTool;
 import com.xm.core.entity.XmMenu;
 import com.xm.core.entity.XmQuestion;
 import com.xm.core.entity.XmTask;
+import com.xm.core.queue.XmMenuSumParentsPushService;
 import com.xm.core.service.XmMenuService;
 import com.xm.core.service.XmGroupService;
 import com.xm.core.service.XmRecordService;
@@ -68,6 +69,10 @@ public class XmMenuController {
 	private XmGroupService groupService;
 
 
+	@Autowired
+	XmMenuSumParentsPushService pushService;
+
+
 	Map<String,Object> fieldsMap = BaseUtils.toMap(new XmMenu());
 	
 	@ApiOperation( value = "查询项目菜单表信息列表",notes="listXmMenu,条件之间是 and关系,模糊查询写法如 {studentName:'%才哥%'}")

xm-core/src/main/java/com/xm/core/ctrl/XmTaskController.java

@@ -16,6 +16,7 @@ import com.mdp.safe.client.entity.User;
 import com.mdp.safe.client.utils.LoginUtils;
 import com.xm.core.PubTool;
 import com.xm.core.entity.*;
+import com.xm.core.queue.XmTaskSumParentsPushService;
 import com.xm.core.service.*;
 import com.xm.core.service.cache.XmTaskCacheService;
 import com.xm.core.service.push.XmPushMsgService;
@@ -76,6 +77,9 @@ public class XmTaskController {
 	@Autowired
 	XmProductService xmProductService;
 
+	@Autowired
+	XmTaskSumParentsPushService pushService;
+
 	Map<String,Object> fieldsMap = BaseUtils.toMap(new XmTask());
 
 	@ApiOperation( value = "查询xm_task信息列表",notes="listXmTask,条件之间是 and关系,模糊查询写法如 {studentName:'%才哥%'}")
@@ -321,7 +325,7 @@ public class XmTaskController {
 					if(fieldKey.contains("budgetWorkload")){//如果调整了预估工时，需要重新计算进度数据
 						if(xmTasksDb.size()>0){
 							this.xmTaskService.batchUpdateBudgetWorkloadAndRate(xmTasksDb.stream().map(i->i.getId()).collect(Collectors.toSet()).stream().collect(Collectors.toList()),NumberUtil.getBigDecimal(xmTaskMap.get("budgetWorkload")));
-							this.xmTaskService.batchSumParents(xmTasksDb);
+							pushService.pushXmTasks(xmTasksDb);
 						}
 					}else{
 						xmTaskService.editSomeFields(xmTaskMap);

xm-core/src/main/java/com/xm/core/listener/XmMenuSumParentsListener.java

@@ -0,0 +1,96 @@
+package com.xm.core.listener;
+
+import com.mdp.mq.queue.MessageListener;
+import com.xm.core.entity.XmMenu;
+import com.xm.core.entity.XmMenu;
+import com.xm.core.queue.XmMenuSumParentsPushService;
+import com.xm.core.queue.XmMenuSumParentsPushService;
+import com.xm.core.service.XmMenuStateService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+@Service
+public class XmMenuSumParentsListener extends MessageListener<XmMenu> {
+
+        Map<String,Map<String,XmMenu>> menusAllMap =new HashMap<>();
+
+
+    @Autowired
+    XmMenuStateService xmMenuStateService;
+
+
+    @Override
+    public Object getQueueKey() {
+        return XmMenuSumParentsPushService.queueName;
+    }
+
+    @Override
+    public void handleMessage(XmMenu message) {
+        synchronized (menusAllMap){
+            Map<String,XmMenu> my=menusAllMap.get(message.getProductId());
+            if(my==null){
+                my=new HashMap<>();
+                menusAllMap.put(message.getProductId(),my);
+            }
+            my.put(message.getMenuId(),message);
+        }
+    }
+
+    /**
+     * 每隔一段时间更新一次数据库
+     */
+    @Scheduled(cron = "* 0/30 * * * ?")
+    public void autoUpdateToDb(){
+        Map<String,Map<String, XmMenu>> myMenusAllMap=new HashMap<>();
+        synchronized (this.menusAllMap){
+            myMenusAllMap.putAll(this.menusAllMap);
+        }
+        if(myMenusAllMap.size()>0){
+
+            List<XmMenu> menus=new ArrayList<>();
+            List<XmMenu> errors=new ArrayList<>();
+            myMenusAllMap.forEach((projectId,menusMap)->{
+                menus.addAll(menusMap.values());
+                if(menus.size()>100){
+                    try {
+                        xmMenuStateService.batchSumParents(menus);
+                        menus.clear();
+                    }catch (Exception e){
+                        errors.addAll(menus);
+                        menus.clear();
+                    }
+                }
+            });
+            if(menus.size()>0){
+                try {
+                    xmMenuStateService.batchSumParents(menus);
+                    menus.clear();
+                }catch (Exception e){
+
+                }
+            }
+            if(errors.size()>0){
+                try {
+                    xmMenuStateService.batchSumParents(errors);
+                    errors.clear();
+                }catch (Exception e){
+                    errors.clear();
+                }
+            }
+        }
+
+    }
+
+    @PostConstruct
+    public void init(){
+        super.popMessage();
+    }
+}

xm-core/src/main/java/com/xm/core/listener/XmTaskSumParentsListener.java

@@ -0,0 +1,95 @@
+package com.xm.core.listener;
+
+import com.mdp.core.service.BaseService;
+import com.mdp.mq.queue.MessageListener;
+import com.xm.core.entity.XmTask;
+import com.xm.core.queue.XmTaskSumParentsPushService;
+import com.xm.core.service.XmTaskService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import java.util.*;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+@Service
+public class XmTaskSumParentsListener extends MessageListener<XmTask> {
+
+        Map<String,Map<String,XmTask>> tasksAllMap=new HashMap<>();
+
+
+    @Autowired
+    XmTaskService xmTaskService;
+
+
+    @Override
+    public Object getQueueKey() {
+        return XmTaskSumParentsPushService.queueName;
+    }
+
+    @Override
+    public void handleMessage(XmTask message) {
+         synchronized (tasksAllMap){
+             Map<String,XmTask> my=tasksAllMap.get(message.getProjectId());
+             if(my==null){
+                 my=new HashMap<>();
+                 tasksAllMap.put(message.getProjectId(),my);
+             }
+             my.put(message.getId(),message);
+         }
+    }
+
+    /**
+     * 每隔一段时间更新一次数据库
+     */
+    @Scheduled(cron = "* 0/30 * * * ?")
+    public void autoUpdateToDb(){
+       Map<String,Map<String,XmTask>> myTasksAllMap=new HashMap<>();
+        synchronized (this.tasksAllMap){
+            myTasksAllMap.putAll(this.tasksAllMap);
+        }
+        if(myTasksAllMap.size()>0){
+
+            List<XmTask> tasks=new ArrayList<>();
+            List<XmTask> errors=new ArrayList<>();
+             myTasksAllMap.forEach((projectId,tasksMap)->{
+                 tasks.addAll(tasksMap.values());
+                 if(tasks.size()>100){
+                     try {
+                         xmTaskService.batchSumParents(tasks);
+                         tasks.clear();
+                     }catch (Exception e){
+                         errors.addAll(tasks);
+                         tasks.clear();
+                     }
+                 }
+             });
+             if(tasks.size()>0){
+                 try {
+                     xmTaskService.batchSumParents(tasks);
+                     tasks.clear();
+                 }catch (Exception e){
+
+                 }
+             }
+            if(errors.size()>0){
+                try {
+                    xmTaskService.batchSumParents(errors);
+                    errors.clear();
+                }catch (Exception e){
+                    errors.clear();
+                }
+            }
+        }
+
+    }
+
+    @PostConstruct
+    public void init(){
+        super.popMessage();
+    }
+}

xm-core/src/main/java/com/xm/core/queue/XmMenuSumParentsPushService.java

@@ -0,0 +1,35 @@
+package com.xm.core.queue;
+
+import com.mdp.mq.queue.Push;
+import com.xm.core.entity.XmMenu;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+
+import java.util.List;
+
+@Service
+public class XmMenuSumParentsPushService {
+
+    @Autowired
+    Push push;
+    
+    public static String queueName="xm-menu-sum-parents";
+    
+    public void pushXmMenu(XmMenu xmMenu){
+        if(xmMenu==null || StringUtils.hasText(xmMenu.getMenuId())){
+            return;
+        }
+        push.leftPush(queueName,xmMenu);
+    }
+
+    public void pushXmMenus(List<XmMenu> xmMenus){
+        if(xmMenus==null ||xmMenus.size()==0){
+            return;
+        }
+        for (XmMenu xmMenu : xmMenus) {
+            push.leftPush(queueName,xmMenu);
+        }
+        
+    }
+}

xm-core/src/main/java/com/xm/core/queue/XmTaskSumParentsPushService.java

@@ -0,0 +1,42 @@
+package com.xm.core.queue;
+
+import com.mdp.mq.queue.MessageListener;
+import com.mdp.mq.queue.Push;
+import com.xm.core.entity.XmTask;
+import com.xm.core.service.XmTaskService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+
+import javax.annotation.PostConstruct;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+@Service
+public class XmTaskSumParentsPushService {
+
+    @Autowired
+    Push push;
+
+    public static String queueName="xm-task-sum-parents";
+
+    public void pushXmTask(XmTask xmTask){
+        if(xmTask==null || StringUtils.hasText(xmTask.getId())){
+            return;
+        }
+        push.leftPush(queueName,xmTask);
+    }
+
+    public void pushXmTasks(List<XmTask> xmTasks){
+        if(xmTasks==null ||xmTasks.size()==0){
+            return;
+        }
+        for (XmTask xmTask : xmTasks) {
+            push.leftPush(queueName,xmTask);
+        }
+
+    }
+}

xm-core/src/main/java/com/xm/core/service/XmMenuService.java

@@ -4,6 +4,7 @@ import com.mdp.core.entity.Tips;
 import com.mdp.core.service.BaseService;
 import com.xm.core.entity.XmMenu;
 import com.xm.core.entity.XmTask;
+import com.xm.core.queue.XmMenuSumParentsPushService;
 import com.xm.core.vo.XmIterationMenuVo;
 import com.xm.core.vo.XmMenuVo;
 import com.xm.core.vo.XmPhaseMenusVo;
@@ -26,6 +27,9 @@ public class XmMenuService extends BaseService {
 	@Autowired
 	XmMenuStateService xmMenuStateService;
 
+	@Autowired
+	XmMenuSumParentsPushService pushService;
+
 	/**
 	 * 连同功能关联的项目需求计划数据一起带出
 	 *
@@ -86,7 +90,7 @@ public class XmMenuService extends BaseService {
 			this.batchUpdate(editList);
 		}
 		if (xmMenuList.size() > 0) {
-			this.xmMenuStateService.batchSumParents(xmMenuList);
+			pushService.pushXmMenus(xmMenuList);
 		}
 	}
 
@@ -215,7 +219,7 @@ public class XmMenuService extends BaseService {
 	public int insert(XmMenu xmMenu) {
 		int i = super.insert(xmMenu);
 		xmMenuStateService.batchLoadXmMenuToState(xmMenu.getProductId());
-		xmMenuStateService.sumParents(xmMenu);
+		pushService.pushXmMenu(xmMenu);
 		return i;
 	}
 
@@ -229,14 +233,14 @@ public class XmMenuService extends BaseService {
 	public void doBatchInsert(List<XmMenu> xmMenus) {
 		super.batchInsert(xmMenus);
 		this.xmMenuStateService.batchLoadXmMenuToState(xmMenus.get(0).getProductId());
-		this.xmMenuStateService.batchSumParents(xmMenus);
+		pushService.pushXmMenus(xmMenus);
 
 	}
 
 	@Transactional
 	public void doBatchDelete(List<XmMenu> canDelList) {
 		super.batchDelete(canDelList);
-		this.xmMenuStateService.batchSumParents(canDelList);
+		pushService.pushXmMenus(canDelList);
 	}
 
 	@Transactional
@@ -299,7 +303,7 @@ public class XmMenuService extends BaseService {
 	@Transactional
 	public void batchChangeParent(List<XmMenu> xmMenus,XmMenu parentMenu) {
 		super.update("batchChangeParent",map("menuIds",xmMenus.stream().map(i->i.getMenuId()).collect(Collectors.toList()),"pmenuId",parentMenu.getMenuId(),"parentPidPaths",parentMenu.getPidPaths()));
-		xmMenuStateService.sumParents(parentMenu);
+ 		pushService.pushXmMenu(parentMenu);
 	}
 
 

xm-core/src/main/java/com/xm/core/service/XmTaskService.java

@@ -13,6 +13,7 @@ import com.mdp.safe.client.utils.LoginUtils;
 import com.xm.core.entity.XmMenu;
 import com.xm.core.entity.XmTask;
 import com.xm.core.entity.XmTaskSkill;
+import com.xm.core.queue.XmTaskSumParentsPushService;
 import com.xm.core.vo.BatchRelTasksWithMenu;
 import com.xm.core.vo.BatchRelTasksWithPhase;
 import com.xm.core.vo.XmGroupVo;
@@ -53,6 +54,9 @@ public class XmTaskService extends BaseService {
 	@Autowired
     XmTaskSkillService xmTaskSkillService;
 
+	@Autowired
+	XmTaskSumParentsPushService pushService;
+
 
 	@Autowired
 	XmGroupService groupService;
@@ -60,7 +64,7 @@ public class XmTaskService extends BaseService {
 	@Transactional
 	public   int[] doBatchDelete(List<XmTask> batchValues) {
 		int[] i2= super.batchDelete(batchValues);
-		this.batchSumParents(batchValues);
+		pushService.pushXmTasks(batchValues);
 		return i2;
 	}
 
@@ -198,7 +202,7 @@ public class XmTaskService extends BaseService {
 		BeanUtils.copyProperties(xmTaskVo,xmTask);
 		this.insert(xmTask);
 		if(StringUtils.hasText(xmTask.getParentTaskid())){
-			this.sumParents(xmTask);
+			pushService.pushXmTask(xmTask);
 		}
 		//新增/更新附件
 		//xmAttachmentService.insertOrUpdate(xmTaskVo.getId(),TYPE,xmTaskVo.getAttachment());
@@ -222,7 +226,7 @@ public class XmTaskService extends BaseService {
 	public void deleteTask(XmTask xmTask) {
 		this.deleteByPk(xmTask);
 		if(StringUtils.hasText(xmTask.getParentTaskid())){
-			this.sumParents(xmTask);
+			pushService.pushXmTask(xmTask);
 		}
 	}
 	@Transactional
@@ -235,7 +239,7 @@ public class XmTaskService extends BaseService {
 		}
 		this.updateSomeFieldByPk(xmTask);
 		if(StringUtils.hasText(xmTaskDb.getParentTaskid())){
-			this.sumParents(xmTaskDb);
+			pushService.pushXmTask(xmTaskDb);
 		}
 		xmRecordService.addXmTaskRecord(xmTask.getProjectId(), xmTask.getId(), "项目-任务-更新任务基础信息", "更新任务"+xmTask.getName(),JSONObject.toJSONString(xmTask),null);  
 	}
@@ -251,7 +255,7 @@ public class XmTaskService extends BaseService {
 		xmTask2.setActEndTime(xmTask.getActEndTime());
 		this.updateSomeFieldByPk(xmTask);
 		if(StringUtils.hasText(xmTaskDb.getParentTaskid())){
-			this.sumParents(xmTaskDb);
+			pushService.pushXmTask(xmTaskDb);
 		}
 		//更新父任务的进度
 		//updateParentProgress(xmTask.getParentTaskid());
@@ -272,7 +276,7 @@ public class XmTaskService extends BaseService {
 		this.updateSomeFieldByPk(xmTask);
 
 		if(StringUtils.hasText(xmTaskDb.getParentTaskid())){
-			this.sumParents(xmTaskDb);
+			pushService.pushXmTask(xmTaskDb);
 		}
 		//更新父任务的进度
 		//updateParentProgress(xmTask.getParentTaskid());
@@ -384,7 +388,7 @@ public class XmTaskService extends BaseService {
 	@Transactional
 	public void batchImportFromTemplate(List<XmTask> xmTasks) {
 		this.batchInsert(xmTasks);
-		this.batchSumParents(xmTasks);
+		this.pushService.pushXmTasks(xmTasks);
 		
 	}
 	
@@ -410,7 +414,7 @@ public class XmTaskService extends BaseService {
 			all.addAll(editXmTasks);
 		}
 		if(all.size()>0){
-			this.batchSumParents(all);
+			this.pushService.pushXmTasks(all);
 		}
 	}
 
@@ -706,7 +710,7 @@ public class XmTaskService extends BaseService {
 	@Transactional
 	public void batchChangeParent(List<XmTask> xmTasks,XmTask parentTask) {
 		super.update("batchChangeParent",map("taskIds",xmTasks.stream().map(i->i.getId()).collect(Collectors.toList()),"parentTaskid",parentTask.getId(),"parentPidPaths",parentTask.getPidPaths()));
-		this.sumParents(parentTask);
+		pushService.pushXmTask(parentTask);
 	}
 
 	/**