From ef6693aabbbf12e83d09ad8749f6e60faacc012d Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期二, 29 十一月 2022 15:18:56 +0800
Subject: [PATCH] Merge branch 'wvp-28181-2.0' into Zafu-Dev-1127
---
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamResponseListener.java | 30 +++++++++++++++---------------
1 files changed, 15 insertions(+), 15 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamResponseListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamResponseListener.java
index 05d662d..33eae1e 100644
--- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamResponseListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamResponseListener.java
@@ -1,7 +1,6 @@
package com.genersoft.iot.vmp.service.redisMsg;
import com.alibaba.fastjson2.JSON;
-import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannelResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -26,8 +25,6 @@
private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamResponseListener.class);
- private boolean taskQueueHandlerRun = false;
-
private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor")
@@ -43,24 +40,27 @@
@Override
public void onMessage(Message message, byte[] bytes) {
- logger.warn("[REDIS娑堟伅-璇锋眰鎺ㄦ祦缁撴灉]锛� {}", new String(message.getBody()));
+ logger.info("[REDIS娑堟伅-璇锋眰鎺ㄦ祦缁撴灉]锛� {}", new String(message.getBody()));
+ boolean isEmpty = taskQueue.isEmpty();
taskQueue.offer(message);
- if (!taskQueueHandlerRun) {
- taskQueueHandlerRun = true;
+ if (isEmpty) {
taskExecutor.execute(() -> {
while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll();
- MessageForPushChannelResponse response = JSON.parseObject(new String(msg.getBody()), MessageForPushChannelResponse.class);
- if (response == null || ObjectUtils.isEmpty(response.getApp()) || ObjectUtils.isEmpty(response.getStream())){
- logger.info("[REDIS娑堟伅-璇锋眰鎺ㄦ祦缁撴灉]锛氬弬鏁颁笉鍏�");
- continue;
- }
- // 鏌ョ湅姝ｅ湪绛夊緟鐨刬nvite娑堟伅
- if (responseEvents.get(response.getApp() + response.getStream()) != null) {
- responseEvents.get(response.getApp() + response.getStream()).run(response);
+ try {
+ MessageForPushChannelResponse response = JSON.parseObject(new String(msg.getBody()), MessageForPushChannelResponse.class);
+ if (response == null || ObjectUtils.isEmpty(response.getApp()) || ObjectUtils.isEmpty(response.getStream())){
+ logger.info("[REDIS娑堟伅-璇锋眰鎺ㄦ祦缁撴灉]锛氬弬鏁颁笉鍏�");
+ continue;
+ }
+ // 鏌ョ湅姝ｅ湪绛夊緟鐨刬nvite娑堟伅
+ if (responseEvents.get(response.getApp() + response.getStream()) != null) {
+ responseEvents.get(response.getApp() + response.getStream()).run(response);
+ }
+ }catch (Exception e) {
+ logger.warn("[REDIS鐨凙LARM閫氱煡] 鍙戠幇鏈鐞嗙殑寮傚父, {}",e.getMessage());
}
}
- taskQueueHandlerRun = false;
});
}
}
--
Gitblit v1.8.0