merge from 4.7.2

Author: gaoyf

mq-client-common-open/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>com.sohu.tv</groupId>
 		<artifactId>mq</artifactId>
-		<version>4.7.1</version>
+		<version>4.7.2</version>
 	</parent>
 
 	<artifactId>mq-client-common-open</artifactId>

mq-client-common-open/src/main/java/com/sohu/index/tv/mq/common/BatchConsumerCallback.java

@@ -10,46 +10,12 @@
  * @param <T> msg obj
  * @param MessageExt
  */
-public interface BatchConsumerCallback<T, MessageExt> {
+public interface BatchConsumerCallback<T> {
 
     /**
      * 订阅回调方法
      *
      * @return
      */
-    void call(List<MQMessage<T, MessageExt>> batchMessage) throws Exception;
-
-    /**
-     * 批量消息
-     * 
-     * @author yongfeigao
-     * @date 2019年10月18日
-     * @param <T>
-     * @param <MessageExt>
-     */
-    public class MQMessage<T, MessageExt> {
-        private T message;
-        private MessageExt messageExt;
-
-        public MQMessage(T message, MessageExt messageExt) {
-            this.message = message;
-            this.messageExt = messageExt;
-        }
-
-        public T getMessage() {
-            return message;
-        }
-
-        public void setMessage(T message) {
-            this.message = message;
-        }
-
-        public MessageExt getMessageExt() {
-            return messageExt;
-        }
-
-        public void setMessageExt(MessageExt messageExt) {
-            this.messageExt = messageExt;
-        }
-    }
+    void call(List<MQMessage<T>> batchMessage) throws Exception;
 }

mq-client-common-open/src/main/java/com/sohu/index/tv/mq/common/MQMessage.java

@@ -0,0 +1,53 @@
+package com.sohu.index.tv.mq.common;
+
+import java.nio.ByteBuffer;
+
+import org.apache.rocketmq.common.message.MessageDecoder;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.sysflag.MessageSysFlag;
+
+/**
+ * 批量消息
+ * 
+ * @author yongfeigao
+ * @date 2019年10月18日
+ * @param <T>
+ * @param <MessageExt>
+ */
+public class MQMessage<T> {
+    private T message;
+    private MessageExt messageExt;
+
+    public MQMessage(T message, MessageExt messageExt) {
+        this.message = message;
+        this.messageExt = messageExt;
+    }
+
+    public T getMessage() {
+        return message;
+    }
+
+    public void setMessage(T message) {
+        this.message = message;
+    }
+
+    public MessageExt getMessageExt() {
+        return messageExt;
+    }
+
+    public void setMessageExt(MessageExt messageExt) {
+        this.messageExt = messageExt;
+    }
+    
+    /**
+     * 构建offsetMsgId
+     * @return
+     */
+    public String buildOffsetMsgId() {
+        int msgIdLength = (messageExt.getSysFlag() & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 + 8
+                : 16 + 4 + 8;
+        ByteBuffer byteBufferMsgId = ByteBuffer.allocate(msgIdLength);
+        return MessageDecoder.createMessageId(byteBufferMsgId, messageExt.getStoreHostBytes(),
+                messageExt.getCommitLogOffset());
+    }
+}

mq-client-common-open/src/main/java/com/sohu/tv/mq/metric/ConsumeFailedStat.java

@@ -0,0 +1,54 @@
+package com.sohu.tv.mq.metric;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * 消费失败统计
+ * 
+ * @author yongfeigao
+ * @date 2021年4月29日
+ */
+public class ConsumeFailedStat {
+    // 计数数组的下标
+    private volatile AtomicLong indexer = new AtomicLong();
+
+    private MessageExceptionMetric[] messageExceptionMetricArray;
+
+    public ConsumeFailedStat(int size) {
+        messageExceptionMetricArray = new MessageExceptionMetric[size];
+    }
+
+    public void set(MessageExceptionMetric messageExceptionMetric) {
+        int index = (int) (indexer.getAndIncrement() % messageExceptionMetricArray.length);
+        // 溢出重置
+        if (index < 0) {
+            indexer.set(0);
+            index = 0;
+        }
+        messageExceptionMetricArray[index] = messageExceptionMetric;
+    }
+
+    /**
+     * 获取所有统计
+     * 
+     * @return
+     */
+    public List<StackTraceMetric> getAll() {
+        List<StackTraceMetric> list = new LinkedList<>();
+        for (MessageExceptionMetric metric : messageExceptionMetricArray) {
+            if (metric == null) {
+                continue;
+            }
+            StackTraceMetric threadMetric = new StackTraceMetric(metric.getStartTime(), metric.getMsgIdList());
+            threadMetric.setId(metric.getThreadId());
+            threadMetric.setName(metric.getThreadName());
+            threadMetric.setStackTraceArray(metric.getException().getStackTrace());
+            threadMetric.setErrorClass(metric.getException().getClass().toString());
+            threadMetric.setMessage(metric.getException().getMessage());
+            list.add(threadMetric);
+        }
+        return list;
+    }
+}

mq-client-common-open/src/main/java/com/sohu/tv/mq/metric/ConsumeStatManager.java

@@ -0,0 +1,44 @@
+package com.sohu.tv.mq.metric;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * 消费统计管理
+ * 
+ * @author yongfeigao
+ * @date 2021年4月14日
+ */
+public class ConsumeStatManager {
+
+    private static ConsumeStatManager instance = new ConsumeStatManager();
+
+    private Map<String, ConsumeThreadStat> consumeThreadMetricsMap = new HashMap<>();
+
+    private Map<String, ConsumeFailedStat> consumeFailedMetricsMap = new HashMap<>();
+
+    private ConsumeStatManager() {
+    }
+
+    public static ConsumeStatManager getInstance() {
+        return instance;
+    }
+
+    /**
+     * 注册
+     * 
+     * @param consuemrGroup
+     */
+    public void register(String consuemrGroup) {
+        consumeThreadMetricsMap.put(consuemrGroup, new ConsumeThreadStat());
+        consumeFailedMetricsMap.put(consuemrGroup, new ConsumeFailedStat(10));
+    }
+
+    public ConsumeThreadStat getConsumeThreadMetrics(String consuemrGroup) {
+        return consumeThreadMetricsMap.get(consuemrGroup);
+    }
+
+    public ConsumeFailedStat getConsumeFailedMetrics(String consuemrGroup) {
+        return consumeFailedMetricsMap.get(consuemrGroup);
+    }
+}

mq-client-common-open/src/main/java/com/sohu/tv/mq/metric/ConsumeThreadStat.java

@@ -0,0 +1,55 @@
+package com.sohu.tv.mq.metric;
+/**
+ * 消费线程统计
+ * 
+ * @author yongfeigao
+ * @date 2021年4月14日
+ */
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * 消费线程统计
+ * 
+ * @author yongfeigao
+ * @date 2021年4月14日
+ */
+public class ConsumeThreadStat {
+
+    // 线程消息统计
+    private ConcurrentMap<Thread, MessageMetric> threadMessageMetricMap = new ConcurrentHashMap<>();
+
+    /**
+     * 设置消息统计
+     * 
+     * @param messageMetric
+     */
+    public void set(MessageMetric messageMetric) {
+        threadMessageMetricMap.put(Thread.currentThread(), messageMetric);
+    }
+
+    /**
+     * 移除
+     */
+    public void remove() {
+        threadMessageMetricMap.remove(Thread.currentThread());
+    }
+
+    /**
+     * 获取所有统计
+     * 
+     * @return
+     */
+    public List<StackTraceMetric> getAll() {
+        List<StackTraceMetric> list = new LinkedList<>();
+        threadMessageMetricMap.forEach((thread, messageMetric) -> {
+            StackTraceMetric threadMetric = new StackTraceMetric(messageMetric.getStartTime(), messageMetric.getMsgIdList());
+            threadMetric.initThreadMetric(thread);
+            list.add(threadMetric);
+        });
+        return list;
+    }
+}

mq-client-common-open/src/main/java/com/sohu/tv/mq/metric/MessageExceptionMetric.java

@@ -0,0 +1,39 @@
+package com.sohu.tv.mq.metric;
+/**
+ * 消息异常统计
+ * 
+ * @author yongfeigao
+ * @date 2021年4月29日
+ */
+public class MessageExceptionMetric extends MessageMetric {
+    // 异常
+    private Throwable exception;
+    // 线程id
+    private long threadId;
+    // 线程名
+    private String threadName;
+
+    public Throwable getException() {
+        return exception;
+    }
+
+    public void setException(Throwable exception) {
+        this.exception = exception;
+    }
+
+    public long getThreadId() {
+        return threadId;
+    }
+
+    public void setThreadId(long threadId) {
+        this.threadId = threadId;
+    }
+
+    public String getThreadName() {
+        return threadName;
+    }
+
+    public void setThreadName(String threadName) {
+        this.threadName = threadName;
+    }
+}

mq-client-common-open/src/main/java/com/sohu/tv/mq/metric/MessageMetric.java

@@ -0,0 +1,32 @@
+package com.sohu.tv.mq.metric;
+
+import java.util.List;
+
+/**
+ * 消息统计
+ * 
+ * @author yongfeigao
+ * @date 2021年4月14日
+ */
+public class MessageMetric {
+    // 开始时间
+    private long startTime;
+    // 消费的消息id
+    private List<String> msgIdList;
+
+    public long getStartTime() {
+        return startTime;
+    }
+
+    public void setStartTime(long startTime) {
+        this.startTime = startTime;
+    }
+
+    public List<String> getMsgIdList() {
+        return msgIdList;
+    }
+
+    public void setMsgIdList(List<String> msgIdList) {
+        this.msgIdList = msgIdList;
+    }
+}