Merge branch 'fix/内容管理消息发送优化' into 'wrdp'

[内容管理]消息发送优化解决内存溢出问题

See merge request o2oa/o2oa!3058

o2server/x_cms_assemble_control/src/main/java/com/x/cms/assemble/control/ThisApplication.java

@@ -42,7 +42,7 @@ public class ThisApplication {
 	// 执行数据库中的批处理操作
 	public static final QueueBatchOperation queueBatchOperation = new QueueBatchOperation();
 	// Document发布时，向所有阅读者推送通知
-	public static final QueueSendDocumentNotify queueSendDocumentNotify = new QueueSendDocumentNotify();
+	public static QueueSendDocumentNotify queueSendDocumentNotify = new QueueSendDocumentNotify();
 	private static final ConcurrentHashMap<String, DataImportStatus> importStatus = new ConcurrentHashMap<>();
 
 	public static Context context() {
@@ -58,6 +58,7 @@ public class ThisApplication {
 		context().startQueue(queueDataRowImport);
 		context().startQueue(queueDocumentUpdate);
 		context().startQueue(queueDocumentViewCountUpdate);
+		queueSendDocumentNotify.initFixedThreadPool(3);
 		context().startQueue(queueSendDocumentNotify);
 
 		// 每天凌晨2点执行一次

o2server/x_cms_assemble_control/src/main/java/com/x/cms/assemble/control/jaxrs/document/ActionPersistPublishAndNotify.java

@@ -69,20 +69,20 @@ public class ActionPersistPublishAndNotify extends BaseAction {
 				modifyDocStatus( id, "published", effectivePerson.getDistinguishedName() );
 				document.setDocStatus("published");
 				document.setPublishTime(new Date());
-				
+
 				document = documentPersistService.refreshDocInfoData( document );
-				
+
 				Wo wo = new Wo();
 				wo.setId( document.getId() );
 				result.setData( wo );
-				
+
 				//检查是否需要删除热点图片
 				try {
 					ThisApplication.queueDocumentUpdate.send( document );
 				} catch ( Exception e1 ) {
 					e1.printStackTrace();
 				}
-				
+
 			} catch (Exception e) {
 				Exception exception = new ExceptionDocumentInfoProcess(e, "系统将文档状态修改为发布状态时发生异常。Id:" + id);
 				result.error(exception);
@@ -186,7 +186,7 @@ public class ActionPersistPublishAndNotify extends BaseAction {
 				logger.error(e, effectivePerson, request, null);
 			}
 		}
-		
+
 		//判断是否需要发送通知消息
 		if (check) {
 			try {
@@ -204,7 +204,7 @@ public class ActionPersistPublishAndNotify extends BaseAction {
 //					}
 					if( categoryInfo.getSendNotify() ){
 						logger.info("try to add notify object to queue for document:" + document.getTitle() );
-						ThisApplication.queueSendDocumentNotify.send( document );
+						ThisApplication.queueSendDocumentNotify.send( document.getId() );
 					}
 				}
 			} catch (Exception e) {
@@ -221,21 +221,21 @@ public class ActionPersistPublishAndNotify extends BaseAction {
 	}
 
 	public static class Wi {
-		
+
 		public static WrapCopier<Wi, Document> copier = WrapCopierFactory.wi( Wi.class, Document.class, null, JpaObject.FieldsUnmodify );
-		
+
 		@FieldDescribe( "文档读者." )
 		private List<PermissionInfo> readerList = null;
-		
+
 		@FieldDescribe( "文档编辑者." )
 		private List<PermissionInfo> authorList = null;
-		
+
 		@FieldDescribe( "图片列表." )
 		private List<String> cloudPictures = null;
-		
+
 		@FieldDescribe( "不修改权限（跳过权限设置，保留原来的设置）." )
 		private Boolean skipPermission  = false;
-		
+
 		public Boolean getSkipPermission() {
 			return skipPermission;
 		}
@@ -269,8 +269,8 @@ public class ActionPersistPublishAndNotify extends BaseAction {
 		}
 
 	}
-	
+
 	public static class Wo extends WoId {
 
 	}
-}
+}

o2server/x_cms_assemble_control/src/main/java/com/x/cms/assemble/control/jaxrs/document/ActionPersistPublishByWorkFlow.java

@@ -211,12 +211,11 @@ public class ActionPersistPublishByWorkFlow extends BaseAction {
 
 		if (check) {
 			try {
-				JsonElement docData = XGsonBuilder.instance().toJsonTree(wi.getDocData());
 				wi.setDocStatus("published");
 				if(wi.getPublishTime()==null) {
 					wi.setPublishTime(new Date());
 				}
-				document = documentPersistService.save(wi, docData);
+				document = documentPersistService.save(wi, wi.getDocData());
 			} catch (Exception e) {
 				check = false;
 				Exception exception = new ExceptionDocumentInfoProcess(e, "系统在创建文档信息时发生异常！");
@@ -377,8 +376,8 @@ public class ActionPersistPublishByWorkFlow extends BaseAction {
 			}
 		}
 
-		//判断是否需要发送通知消息
-		if (check) {
+		//判断是否需要发送通知消息 后端业务接口不需要发送消息
+		/*if (check) {
 			try {
 				Boolean notify = false;
 				if( categoryInfo.getSendNotify() == null ) {
@@ -392,7 +391,7 @@ public class ActionPersistPublishByWorkFlow extends BaseAction {
 				}
 				if( notify ){
 					logger.info("try to add notify object to queue for document:" + document.getTitle() );
-					ThisApplication.queueSendDocumentNotify.send( document );
+					ThisApplication.queueSendDocumentNotify.send( document.getId() );
 				}
 			} catch (Exception e) {
 				check = false;
@@ -400,7 +399,7 @@ public class ActionPersistPublishByWorkFlow extends BaseAction {
 				result.error( exception );
 				logger.error( e, effectivePerson, request, null);
 			}
-		}
+		}*/
 
 		CacheManager.notify(Document.class);
 		return result;
@@ -464,7 +463,7 @@ public class ActionPersistPublishByWorkFlow extends BaseAction {
 		private String[] wf_attachmentIds = null;
 
 		@FieldDescribe( "文档数据." )
-		private Map<?, ?> docData = null;
+		private JsonElement docData = null;
 
 		@FieldDescribe( "文档读者." )
 		private List<PermissionInfo> readerList = null;
@@ -521,11 +520,11 @@ public class ActionPersistPublishByWorkFlow extends BaseAction {
 			this.dataPaths = dataPaths;
 		}
 
-		public Map<?, ?> getDocData() {
+		public JsonElement getDocData() {
 			return docData;
 		}
 
-		public void setDocData(Map<?, ?> docData) {
+		public void setDocData(JsonElement docData) {
 			this.docData = docData;
 		}
 

o2server/x_cms_assemble_control/src/main/java/com/x/cms/assemble/control/jaxrs/document/ActionPersistPublishContent.java

@@ -227,12 +227,11 @@ public class ActionPersistPublishContent extends BaseAction {
 
 		if (check) {
 			try {
-				JsonElement docData = XGsonBuilder.instance().toJsonTree(wi.getDocData(), Map.class);
 				wi.setDocStatus("published");
 				if( wi.getPublishTime() == null ) { wi.setPublishTime(new Date()); }
 				document =  wi.copier.copy(wi);
 				document.setId( wi.getId() );
-				document = documentPersistService.save( document, docData );
+				document = documentPersistService.save( document, wi.getDocData() );
 			} catch (Exception e) {
 				check = false;
 				Exception exception = new ExceptionDocumentInfoProcess(e, "系统在创建文档信息时发生异常！");
@@ -441,7 +440,7 @@ public class ActionPersistPublishContent extends BaseAction {
 				}
 				if( notify ){
 					logger.debug("try to add notify object to queue for document:" + document.getTitle() );
-					ThisApplication.queueSendDocumentNotify.send( document );
+					ThisApplication.queueSendDocumentNotify.send( document.getId() );
 				}
 			} catch (Exception e) {
 				check = false;
@@ -514,7 +513,7 @@ public class ActionPersistPublishContent extends BaseAction {
 		private String[] cms_attachmentIds = null;
 
 		@FieldDescribe( "文档数据JSON对象." )
-		private Map<?, ?> docData = null;
+		private JsonElement docData = null;
 
 		@FieldDescribe( "文档读者，Json数组，权限对象需要包含四个属性:<br/>permission权限类别：读者|阅读|作者|管理,  <br/>permissionObjectType使用者类别：所有人|组织|人员|群组, <br/>permissionObjectCode使用者编码：所有人|组织编码|人员UID|群组编码, <br/>permissionObjectName使用者名称：所有人|组织名称|人员名称|群组名称" )
 		private List<PermissionInfo> readerList = null;
@@ -1013,22 +1012,11 @@ public class ActionPersistPublishContent extends BaseAction {
 			this.readerList = readerList;
 		}
 
-//		public String[] getDataPaths() {
-//			if( dataPaths != null && dataPaths.length == 1 && dataPaths[0].equals("null")){
-//				return null;
-//			}
-//			return dataPaths;
-//		}
-//
-//		public void setDataPaths(String[] dataPaths) {
-//			this.dataPaths = dataPaths;
-//		}
-
-		public Map<?, ?> getDocData() {
+		public JsonElement getDocData() {
 			return docData;
 		}
 
-		public void setDocData(Map<?, ?> docData) {
+		public void setDocData(JsonElement docData) {
 			this.docData = docData;
 		}
 

o2server/x_cms_assemble_control/src/main/java/com/x/cms/assemble/control/jaxrs/document/ActionPersistSaveDocument.java

@@ -50,15 +50,15 @@ public class ActionPersistSaveDocument extends BaseAction {
 		try {
 			wi = this.convertToWrapIn( jsonElement, Wi.class );
 			document = Wi.copier.copy(wi);
-			document.setId( wi.getId() ); //继承传入的ID			
-			identity = wi.getIdentity();	
+			document.setId( wi.getId() ); //继承传入的ID
+			identity = wi.getIdentity();
 		} catch (Exception e ) {
 			check = false;
 			Exception exception = new ExceptionDocumentInfoProcess( e, "系统在将JSON信息转换为对象时发生异常。");
 			result.error( exception );
 			logger.error( e, effectivePerson, request, null);
 		}
-		
+
 		if (check) {
 			if( !"xadmin".equals( effectivePerson.getDistinguishedName() )) {
 				try {
@@ -166,15 +166,15 @@ public class ActionPersistSaveDocument extends BaseAction {
 
 		if (check) {
 			//补充部分信息
-//			document.setCategoryId(categoryInfo.getId());	
-			document.setAppId(appInfo.getId());					
-			document.setDocumentType( categoryInfo.getDocumentType() );	
+//			document.setCategoryId(categoryInfo.getId());
+			document.setAppId(appInfo.getId());
+			document.setDocumentType( categoryInfo.getDocumentType() );
 			document.setAppAlias( appInfo.getAppAlias());
 			document.setAppName(appInfo.getAppName());
 			document.setCategoryName(categoryInfo.getCategoryName());
 			document.setCategoryAlias(categoryInfo.getCategoryAlias());
 			document.setDocumentType( categoryInfo.getDocumentType() );
-			
+
 			if( !"信息".equals(document.getDocumentType()) && !"数据".equals( document.getDocumentType() )) {
 				document.setDocumentType( "信息" );
 			}
@@ -184,8 +184,8 @@ public class ActionPersistSaveDocument extends BaseAction {
 			try (EntityManagerContainer emc = EntityManagerContainerFactory.instance().create()) {
 				if( StringUtils.isNotEmpty( identity )) {
 					document.setCreatorIdentity( identity );
-				}				
-				
+				}
+
 				if (StringUtils.isEmpty( document.getCreatorIdentity() )) {
 					if( "cipher".equalsIgnoreCase( effectivePerson.getDistinguishedName() )) {
 						document.setCreatorIdentity("cipher");
@@ -202,7 +202,7 @@ public class ActionPersistSaveDocument extends BaseAction {
 						document.setCreatorIdentity(userManagerService.getMajorIdentityWithPerson( effectivePerson.getDistinguishedName()) );
 					}
 				}
-				
+
 				if ( !StringUtils.equals(  "cipher", document.getCreatorIdentity() ) && !StringUtils.equals(  "xadmin", document.getCreatorIdentity() )) {
 					//说明是指定的发布者，并不使用cipher和xadmin代替
 					if (StringUtils.isNotEmpty( document.getCreatorIdentity() )) {
@@ -219,7 +219,7 @@ public class ActionPersistSaveDocument extends BaseAction {
 				result.error(th);
 			}
 		}
-		
+
 //		if (check) {
 //			if ( StringUtils.isNotEmpty(identity)) {
 //				document.setCreatorIdentity( identity );
@@ -267,11 +267,7 @@ public class ActionPersistSaveDocument extends BaseAction {
 
 		if (check) {
 			try {
-				JsonElement dataJson = null;
-				if( wi.getDocData() != null ) {
-					dataJson = XGsonBuilder.instance().toJsonTree( wi.getDocData() );
-				}
-				document = documentPersistService.save( document, dataJson );
+				document = documentPersistService.save( document, wi.getDocData() );
 				CacheManager.notify(Document.class);
 
 				Wo wo = new Wo();
@@ -291,7 +287,7 @@ public class ActionPersistSaveDocument extends BaseAction {
 				logger.error(e, effectivePerson, request, null);
 			}
 		}
-		
+
 		// 处理文档的云文档图片信息
 		if (check) {
 			try {
@@ -363,7 +359,7 @@ public class ActionPersistSaveDocument extends BaseAction {
 				}
 			}
 		}
-		
+
 		if (check) {
 			try {//将读者以及作者信息持久化到数据库中
 				document = documentPersistService.refreshDocumentPermission( document.getId(), wi.getReaderList(), wi.getAuthorList() );
@@ -378,16 +374,16 @@ public class ActionPersistSaveDocument extends BaseAction {
 	}
 
 	public static class Wi{
-		
+
 		@FieldDescribe("ID，非必填，更新时必填写，不然就是新增文档")
 		private String id;
-		
+
 		@FieldDescribe("文档标题，<font style='color:red'>必填</font>")
 		private String title;
 
 		@FieldDescribe("分类ID，<font style='color:red'>必填</font>")
 		private String categoryId;
-		
+
 		@FieldDescribe( "文档操作者身份，如果不传入则取登录者信息。" )
 		private String identity = null;
 
@@ -396,42 +392,42 @@ public class ActionPersistSaveDocument extends BaseAction {
 
 		@FieldDescribe("文档状态: published | draft | checking | error，非必填，默认为draft")
 		private String docStatus = "draft";
-		
+
 		@FieldDescribe("文档发布时间")
 		private Date publishTime;
 
 		@FieldDescribe("首页图片列表，非必填")
-		private List<String> pictureList = null;		
-		
+		private List<String> pictureList = null;
+
 		@FieldDescribe( "数据的路径列表，非必填" )
 		private String[] dataPaths = null;
-		
+
 		@FieldDescribe( "启动流程的JobId，非必填" )
 		private String wf_jobId = null;
-		
+
 		@FieldDescribe( "启动流程的WorkId，非必填" )
 		private String wf_workId = null;
-		
+
 		@FieldDescribe( "启动流程的附件列表，非必填" )
-		private String[] wf_attachmentIds = null;	
-		
+		private String[] wf_attachmentIds = null;
+
 		@FieldDescribe( "文档数据，非必填" )
-		private Map<?, ?> docData = null;
-		
+		private JsonElement docData = null;
+
 		@FieldDescribe( "文档读者，非必填：{'permission':'读者', 'permissionObjectType':'组织', 'permissionObjectCode':'组织全称', 'permissionObjectName':'组织全称'}" )
 		private List<PermissionInfo> readerList = null;
-		
+
 		@FieldDescribe( "文档编辑者，非必填：{'permission':'读者', 'permissionObjectType':'组织', 'permissionObjectCode':'组织全称', 'permissionObjectName':'组织全称'}" )
 		private List<PermissionInfo> authorList = null;
-		
+
 		@FieldDescribe( "图片列表，非必填" )
 		private List<String> cloudPictures = null;
-		
+
 		@FieldDescribe( "不修改权限（跳过权限设置，保留原来的设置），非必填" )
-		private Boolean skipPermission  = false;	
+		private Boolean skipPermission  = false;
 
 		public static WrapCopier<Wi, Document> copier = WrapCopierFactory.wi( Wi.class, Document.class, null, JpaObject.FieldsUnmodifyExcludeId);
-		
+
 		public String getId() {
 			return id;
 		}
@@ -487,7 +483,7 @@ public class ActionPersistSaveDocument extends BaseAction {
 		public void setIdentity(String identity) {
 			this.identity = identity;
 		}
-		
+
 		public List<PermissionInfo> getReaderList() {
 			return readerList;
 		}
@@ -515,11 +511,11 @@ public class ActionPersistSaveDocument extends BaseAction {
 			this.dataPaths = dataPaths;
 		}
 
-		public Map<?, ?> getDocData() {
+		public JsonElement getDocData() {
 			return docData;
 		}
 
-		public void setDocData(Map<?, ?> docData) {
+		public void setDocData(JsonElement docData) {
 			this.docData = docData;
 		}
 
@@ -575,4 +571,4 @@ public class ActionPersistSaveDocument extends BaseAction {
 	public static class Wo extends WoId {
 
 	}
-}
+}

o2server/x_cms_assemble_control/src/main/java/com/x/cms/assemble/control/queue/QueueSendDocumentNotify.java

@@ -25,72 +25,72 @@ import java.util.List;
 /**
  * Document正式发布后，向所有的阅读者推送消息通知
  */
-public class QueueSendDocumentNotify extends AbstractQueue<Document> {
-	
+public class QueueSendDocumentNotify extends AbstractQueue<String> {
+
 	private static  Logger logger = LoggerFactory.getLogger( QueueSendDocumentNotify.class );
 	private UserManagerService userManagerService = new UserManagerService();
 
-	public void execute( Document document ) throws Exception {
-		logger.debug(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>QueueSendDocumentNotify:" + document.getTitle() );
-		if( document == null ) {
+	public void execute( String documentId ) throws Exception {
+		logger.debug(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>start QueueSendDocumentNotify:" + documentId );
+		if( StringUtils.isEmpty(documentId) ) {
 			logger.debug("can not send publish notify , document is NULL!" );
 			return;
 		}
-		if( !StringUtils.equalsIgnoreCase( "信息" , document.getDocumentType()) ) {
-			logger.debug("can not send publish notify , document is not '信息'!" );
-			return;
-		}
-		logger.debug("send publish notify for new document:" + document.getTitle() );
+		List<String> persons = null;
+		Document document = null;
 		try (EntityManagerContainer emc = EntityManagerContainerFactory.instance().create()) {
-			AppInfo appInfo = emc.find( document.getAppId(), AppInfo.class );
-			CategoryInfo category = emc.find( document.getCategoryId(), CategoryInfo.class );
-
-			if( appInfo != null && category != null ) {
-				ReviewService reviewService = new ReviewService();
-				List<String> persons = reviewService.listPermissionPersons( appInfo, category, document );
-				if( ListTools.isNotEmpty( persons )) {
-					//有可能是*， 一般是所有的人员标识列表
-					if( persons.contains( "*" )) {
-						String topUnitName = document.getCreatorTopUnitName();
-						logger.debug(">>>>>document.getCreatorTopUnitName()=" + topUnitName );
-						if( StringUtils.equalsAnyIgnoreCase("cipher",topUnitName ) || StringUtils.equalsAnyIgnoreCase("xadmin", topUnitName) ){
-							//取发起人所有顶层组织
-							if( !StringUtils.equalsAnyIgnoreCase("cipher",document.getCreatorIdentity() ) &&
-								!StringUtils.equalsAnyIgnoreCase("xadmin",document.getCreatorIdentity() )){
-								topUnitName = userManagerService.getTopUnitNameByIdentity(document.getCreatorIdentity());
-							}else if(!StringUtils.equalsAnyIgnoreCase("cipher",document.getCreatorPerson() ) &&
-									!StringUtils.equalsAnyIgnoreCase("xadmin",document.getCreatorPerson() )){
-								topUnitName = userManagerService.getTopUnitNameWithPerson(document.getCreatorPerson());
+			document = emc.find(documentId, Document.class);
+			if(document !=null && StringUtils.equals( "信息" , document.getDocumentType())) {
+				logger.debug("send publish notify for new document:" + document.getTitle() );
+				AppInfo appInfo = emc.find(document.getAppId(), AppInfo.class);
+				CategoryInfo category = emc.find(document.getCategoryId(), CategoryInfo.class);
+
+				if (appInfo != null && category != null) {
+					ReviewService reviewService = new ReviewService();
+					persons = reviewService.listPermissionPersons(appInfo, category, document);
+					if (ListTools.isNotEmpty(persons)) {
+						//有可能是*， 一般是所有的人员标识列表
+						if (persons.contains("*")) {
+							String topUnitName = document.getCreatorTopUnitName();
+							logger.debug(">>>>>document.getCreatorTopUnitName()=" + topUnitName);
+							if (StringUtils.equalsAnyIgnoreCase("cipher", topUnitName) || StringUtils.equalsAnyIgnoreCase("xadmin", topUnitName)) {
+								//取发起人所有顶层组织
+								if (!StringUtils.equalsAnyIgnoreCase("cipher", document.getCreatorIdentity()) &&
+										!StringUtils.equalsAnyIgnoreCase("xadmin", document.getCreatorIdentity())) {
+									topUnitName = userManagerService.getTopUnitNameByIdentity(document.getCreatorIdentity());
+								} else if (!StringUtils.equalsAnyIgnoreCase("cipher", document.getCreatorPerson()) &&
+										!StringUtils.equalsAnyIgnoreCase("xadmin", document.getCreatorPerson())) {
+									topUnitName = userManagerService.getTopUnitNameWithPerson(document.getCreatorPerson());
+								}
+							}
+							if (StringUtils.isNotEmpty(topUnitName)) {
+								//取顶层组织的所有人
+								persons = listPersonWithUnit(topUnitName);
+							} else {
+								persons = new ArrayList<>();
 							}
-						}
-						if( StringUtils.isNotEmpty( topUnitName )){
-							//取顶层组织的所有人
-							persons = listPersonWithUnit( topUnitName );
-						}else{
-							persons = new ArrayList<>();
 						}
 					}
+				} else {
+					logger.debug("can not send publish notify for document, category or  appinfo not exists! ID： " + document.getId());
 				}
-				if( ListTools.isNotEmpty( persons )) {
-					//去一下重复
-			        HashSet<String> set = new HashSet<String>( persons );
-			        persons.clear();
-			        persons.addAll(set);
-			        
-					MessageWo wo = MessageWo.copier.copy(document);
-					for( String person : persons ) {
-						if( !StringUtils.equals( "*", person  )) {
-							MessageFactory.cms_publish(person, wo);
-						}
-					}
-					logger.debug("cms send total count:" + persons.size()  );
+			}
+		}
+		if( ListTools.isNotEmpty( persons )) {
+			//去一下重复
+			HashSet<String> set = new HashSet<String>( persons );
+			persons.clear();
+			persons.addAll(set);
+
+			MessageWo wo = MessageWo.copier.copy(document);
+			for( String person : persons ) {
+				if( !StringUtils.equals( "*", person  )) {
+					MessageFactory.cms_publish(person, wo);
 				}
-				logger.debug("cms send publish notify for new document completed! " );
-				//}
-			}else{
-				logger.debug("can not send publish notify for document, category or  appinfo not exists! ID： " + document.getId() );
 			}
+			logger.debug(documentId +" cms send total count:" + persons.size()  );
 		}
+		logger.debug(documentId + " QueueSendDocumentNotify cms send publish notify for new document completed! " );
 	}
 
 	/**
@@ -108,7 +108,7 @@ public class QueueSendDocumentNotify extends AbstractQueue<Document> {
 		}
 		return persons;
 	}
-	
+
 	public static class MessageWo{
 
 		public static List<String> Excludes = new ArrayList<String>();