|
@@ -1,48 +1,64 @@
|
|
package com.iamberry.wechat.handles.mq;
|
|
package com.iamberry.wechat.handles.mq;
|
|
|
|
|
|
-import java.util.Timer;
|
|
|
|
-import java.util.TimerTask;
|
|
|
|
-import java.util.concurrent.ArrayBlockingQueue;
|
|
|
|
-import java.util.concurrent.BlockingQueue;
|
|
|
|
-
|
|
|
|
-import com.iamberry.app.tool.log.RatFWLogger;
|
|
|
|
import com.iamberry.wechat.core.entity.mq.MQMessage;
|
|
import com.iamberry.wechat.core.entity.mq.MQMessage;
|
|
import com.iamberry.wechat.face.mq.MQSerivce;
|
|
import com.iamberry.wechat.face.mq.MQSerivce;
|
|
|
|
+import com.iamberry.zk.SpringContextHolder;
|
|
|
|
+import org.slf4j.Logger;
|
|
|
|
+import org.slf4j.LoggerFactory;
|
|
|
|
+import org.springframework.beans.factory.annotation.Autowired;
|
|
|
|
+import org.springframework.context.annotation.Lazy;
|
|
|
|
+import org.springframework.context.event.EventListener;
|
|
|
|
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
|
|
|
|
+import org.springframework.scheduling.annotation.Async;
|
|
|
|
+import org.springframework.scheduling.annotation.EnableAsync;
|
|
|
|
+import org.springframework.stereotype.Component;
|
|
|
|
+import org.springframework.stereotype.Service;
|
|
|
|
+import org.springframework.transaction.TransactionDefinition;
|
|
|
|
+import org.springframework.transaction.TransactionStatus;
|
|
|
|
+import org.springframework.transaction.support.DefaultTransactionDefinition;
|
|
|
|
|
|
-public class MQTimer extends TimerTask {
|
|
|
|
|
|
+import java.lang.reflect.Method;
|
|
|
|
|
|
- // 队列
|
|
|
|
- public static BlockingQueue<MQMessage> messageStaticQueue = new ArrayBlockingQueue<MQMessage>(500);
|
|
|
|
-
|
|
|
|
- private MQSerivce mQSerivce;
|
|
|
|
-
|
|
|
|
- private RatFWLogger ratFWLogger;
|
|
|
|
-
|
|
|
|
- public MQTimer(MQSerivce mqSerivce, RatFWLogger logger) {
|
|
|
|
- this.mQSerivce = mqSerivce;
|
|
|
|
- this.ratFWLogger = logger;
|
|
|
|
- }
|
|
|
|
|
|
+@Component
|
|
|
|
+@EnableAsync
|
|
|
|
+public class MQTimer {
|
|
|
|
+
|
|
|
|
+ @Autowired
|
|
|
|
+ private DataSourceTransactionManager transactionManager;
|
|
|
|
|
|
- @Override
|
|
|
|
- public void run() {
|
|
|
|
- ratFWLogger.info("启动订阅程序...");
|
|
|
|
|
|
+ private static final Logger LOGGER = LoggerFactory.getLogger(MQTimer.class);
|
|
|
|
+
|
|
|
|
+ @Async
|
|
|
|
+ @EventListener
|
|
|
|
+ public void run(MQMessage mqMessage) {
|
|
// 订阅
|
|
// 订阅
|
|
try {
|
|
try {
|
|
- while (true) {
|
|
|
|
- // 如果空闲,阻塞,让出系统资源
|
|
|
|
- MQMessage mqMessage = messageStaticQueue.take();
|
|
|
|
- ratFWLogger.info("收到订阅请求...");
|
|
|
|
- // 远程调用 Service
|
|
|
|
- boolean flag = mQSerivce.updateMessageOne(mqMessage);
|
|
|
|
- if (flag) {
|
|
|
|
- ratFWLogger.info(mqMessage + " execution true");
|
|
|
|
- } else {
|
|
|
|
- ratFWLogger.error(this, mqMessage + " execution error");
|
|
|
|
|
|
+ LOGGER.info("收到订阅请求...");
|
|
|
|
+ // 远程调用 Service
|
|
|
|
+ Object object = SpringContextHolder.getBean(mqMessage.getServiceHandlerObjectName());
|
|
|
|
+ if (object == null) {
|
|
|
|
+ throw new RuntimeException(mqMessage.getServiceHandlerObjectName() + "{} 对象不存在,请设置");
|
|
|
|
+ }
|
|
|
|
+ Class classes = object.getClass();
|
|
|
|
+ // 获取方法
|
|
|
|
+ Method method = classes.getMethod(mqMessage.getServiceHandlerMethodName(), MQMessage.class);
|
|
|
|
+ if (method == null) {
|
|
|
|
+ throw new RuntimeException(mqMessage.getServiceHandlerMethodName() + "{} 方法不存在,请设置");
|
|
|
|
+ } else {
|
|
|
|
+ // 执行方法,对方法启动事务
|
|
|
|
+ DefaultTransactionDefinition def = new DefaultTransactionDefinition();
|
|
|
|
+ def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
|
|
|
|
+ TransactionStatus status = transactionManager.getTransaction(def);
|
|
|
|
+ try {
|
|
|
|
+ //逻辑代码
|
|
|
|
+ method.invoke(object, mqMessage);
|
|
|
|
+ transactionManager.commit(status);
|
|
|
|
+ } catch (Exception e) {
|
|
|
|
+ transactionManager.rollback(status);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
} catch (Exception e) {
|
|
} catch (Exception e) {
|
|
- ratFWLogger.error(this, e.getMessage());
|
|
|
|
- new Timer().schedule(new MQTimer(mQSerivce, ratFWLogger), 1000);
|
|
|
|
|
|
+ LOGGER.error("{}", e.getMessage());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
|