Skip to content

iBiz4j Spring R7
iBiz4j Spring R7
提交 07a4e20c 编写于 作者: zhouweidong's avatar zhouweidong
浏览文件
操作

实体数据同步逻辑优化（remove）

上级 279993ca
显示空白字符变更
内嵌 并排
正在显示 包含 53 行增加 和 97 行删除
SLN/%PUBPRJ%-core/src/main/java/%SYS_PKGPATH%/core/util/config/RocketMQListenerProcessor.java.ftl
浏览文件 @ 07a4e20c
......@@ -32,21 +32,21 @@ TARGET=PSSYSTEM
</#if>
package ${pub.getPKGCodeName()}.core.util.config;
import lombok.extern.slf4j.Slf4j;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import java.io.ByteArrayInputStream;
import java.io.ObjectInputStream;
import java.util.List;
import java.util.Map;
/**
* MQ订阅消息处理
......@@ -82,7 +82,8 @@ public class RocketMQListenerProcessor implements MessageListenerOrderly {
try {
String topic = messageExt.getTopic();
String tags = messageExt.getTags();
Object body = null;
String body = new String(messageExt.getBody(),"utf-8");
log.info("MQ消息topic={}, tags={}, 消息内容={}", topic, tags, body);
<#list sys.getAllPSDataEntities() as dataEntity>
<#if dataEntity.getAllPSDEDataSyncs?? && dataEntity.getAllPSDEDataSyncs()??>
<#list dataEntity.getAllPSDEDataSyncs() as dataSync>
......@@ -92,18 +93,15 @@ public class RocketMQListenerProcessor implements MessageListenerOrderly {
<#assign inputAction=srfmethodname(dataSync.getInTestPSDEAction().codeName)>
<#assign columnname = dataEntity.getKeyPSDEField().getName()?lower_case>
if ("${dataSyncCodeName}".equalsIgnoreCase(tags)) {
<#if inputAction?lower_case == 'get' || inputAction?lower_case == 'remove'>
body = toObject(messageExt.getBody());
if (!ObjectUtils.isEmpty(body)) {
JSONObject obj = JSONObject.parseObject(body);
${pub.getPKGCodeName()}.core.${dataEntity.getPSSystemModule().getCodeName()?lower_case}.domain.${entityName} domain = new ${pub.getPKGCodeName()}.core.${dataEntity.getPSSystemModule().getCodeName()?lower_case}.domain.${entityName}();
domain.set("${columnname}",body);
${entityName}Service.${inputAction}(domain.get${srfcaseformat(dataEntity.getKeyPSDEField().codeName,'l_u2lC')?cap_first}());
for(Map.Entry<String, Object> entry : obj.entrySet()){
domain.set(entry.getKey(),entry.getValue());
}
<#if inputAction?lower_case == 'get' || inputAction?lower_case == 'remove'>
${entityName}Service.${inputAction}(domain.get${srfcaseformat(dataEntity.getKeyPSDEField().codeName,'l_u2lC')?cap_first}());
<#else>
String strBody = new String(messageExt.getBody(), "utf-8");
${pub.getPKGCodeName()}.core.${dataEntity.getPSSystemModule().getCodeName()?lower_case}.domain.${entityName} domain = JSON.parseObject(strBody,${pub.getPKGCodeName()}.core.${dataEntity.getPSSystemModule().getCodeName()?lower_case}.domain.${entityName}.class);
${entityName}Service.${inputAction}(domain);
body = strBody;
</#if>
}
<#else>
......@@ -113,7 +111,6 @@ public class RocketMQListenerProcessor implements MessageListenerOrderly {
</#list>
</#if>
</#list>
log.info("MQ消息topic={}, tags={}, 消息内容={}", topic, tags, body);
} catch (Exception e) {
log.error("获取MQ消息内容异常{}", e);
}
......@@ -121,34 +118,5 @@ public class RocketMQListenerProcessor implements MessageListenerOrderly {
}
return ConsumeOrderlyStatus.SUCCESS;
}
/**
* 数组转对象
*
* @param bytes
* @return
*/
public Object toObject(byte[] bytes) {
Object obj = null;
ByteArrayInputStream bis = null;
ObjectInputStream ois = null;
try {
bis = new ByteArrayInputStream(bytes);
ois = new ObjectInputStream(bis);
obj = ois.readObject();
} catch (Exception ex) {
} finally {
try {
if (ois != null) {
ois.close();
}
if (bis != null) {
bis.close();
}
} catch (Exception e) {
}
}
return obj;
}
}
</#if>
\ No newline at end of file
SLN/%PUBPRJ%-util/src/main/java/%SYS_PKGPATH%/util/aspect/RocketMQAspect.java.ftl
浏览文件 @ 07a4e20c
......@@ -31,12 +31,15 @@ TARGET=PSSYSTEM
</#if>
package ${pub.getPKGCodeName()}.util.aspect;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import ${pub.getPKGCodeName()}.util.domain.EntityBase;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.AfterReturning;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
......@@ -47,14 +50,10 @@ import org.springframework.expression.Expression;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import java.io.ByteArrayOutputStream;
import java.io.ObjectOutputStream;
/**
* rocketMQ消息切面
*/
......@@ -78,14 +77,15 @@ public class RocketMQAspect
<#list dataEntity.getAllPSDEDataSyncs() as dataSync>
<#if dataSync.getOutPSSysDataSyncAgent?? && dataSync.getOutPSSysDataSyncAgent()??>
<#assign producer=dataEntity.codeName+dataSync.codeName>
@AfterReturning(value = "${getPointCut(dataEntity,dataSync)}")
@Async
public void ${srfmethodname(producer)}(JoinPoint point) {
@Around(value = "${getPointCut(dataEntity,dataSync)}")
public void ${srfmethodname(producer)}(ProceedingJoinPoint point) {
Object entity = getEntity(point);
Object result = point.proceed();
<#if dataSync.getOutTestPSDEAction?? && dataSync.getOutTestPSDEAction()??>
<#assign actionName=srfmethodname(dataSync.getOutTestPSDEAction().codeName)>
outputAction(point, "${actionName}");
outputAction(point, "${actionName}", entity);
</#if>
sendMsg(topic, "${dataSync.codeName?lower_case}", getEntity(point));
sendMsg(topic, "${dataSync.codeName?lower_case}", entity);
}
</#if>
</#list>
......@@ -96,17 +96,17 @@ public class RocketMQAspect
* 输出过滤行为
* @param point
* @param actionName
* @param entity
*/
private void outputAction(JoinPoint point,String actionName) {
Object [] args = point.getArgs();
if(ObjectUtils.isEmpty(args) || args.length==0) {
private void outputAction(JoinPoint point, String actionName, Object entity) {
Object[] args = point.getArgs();
if (ObjectUtils.isEmpty(args) || args.length == 0 || ObjectUtils.isEmpty(entity)) {
return;
}
Object arg = args[0];
Object service = point.getTarget();
EvaluationContext serviceCtx = new StandardEvaluationContext();
serviceCtx.setVariable("service", service);
serviceCtx.setVariable("arg", arg);
serviceCtx.setVariable("arg", entity);
Expression serviceExp = parser.parseExpression(String.format("#service.%s(#arg)", actionName));
serviceExp.getValue(serviceCtx);
}
......@@ -117,13 +117,31 @@ public class RocketMQAspect
* @param point
* @return
*/
private Object getEntity(JoinPoint point) {
private Object getEntity(ProceedingJoinPoint point) {
Object entity = null;
try {
String action = point.getSignature().getName();
Object[] args = point.getArgs();
Object serviceObj = point.getTarget();
if (ObjectUtils.isEmpty(args) || args.length == 0 || StringUtils.isEmpty(action)) {
return null;
return entity;
}
if ("remove".equalsIgnoreCase(action) || "get".equalsIgnoreCase(action)) {
Object idValue = args[0];
if (!ObjectUtils.isEmpty(serviceObj)) {
EvaluationContext oldContext = new StandardEvaluationContext();
oldContext.setVariable("service", serviceObj);
oldContext.setVariable("id", idValue);
Expression oldExp = parser.parseExpression("#service.get(#id)");
entity = oldExp.getValue(oldContext, EntityBase.class);
}
} else {
entity = args[0];
}
return args[0];
} catch (Exception e) {
log.error("发送消息失败，无法获取实体对象" + e);
}
return entity;
}
/**
......@@ -139,43 +157,13 @@ public class RocketMQAspect
return;
}
try {
Message message = new Message(topic, tag, toByte(body));
Message message = new Message(topic, tag, JSON.toJSONString(body).getBytes());
SendResult sendResult = defaultMQProducer.send(message);
log.info("消息发送响应：" + sendResult.toString());
} catch (Exception e) {
log.error("消息发送异常，" + e);
}
}
/**
* 转byte数组
*
* @param object
* @return
*/
public byte[] toByte(Object object) {
byte[] bytes = null;
ObjectOutputStream oos = null;
ByteArrayOutputStream bos = new ByteArrayOutputStream();
try {
oos = new ObjectOutputStream(bos);
oos.writeObject(object);
oos.flush();
bytes = bos.toByteArray();
} catch (Exception ex) {
} finally {
try {
if (oos != null) {
oos.close();
}
if (bos != null) {
bos.close();
}
} catch (Exception e) {
}
}
return bytes;
}
}
</#if>
......
Markdown 格式
0% or
您添加了 0 人 到此讨论。请谨慎行事。
先完成此消息的编辑！
想要评论请 注册 或