新增MQ消息忽略开关；优化JPush消息推送逻辑；

framework-base/src/main/java/com/chelvc/framework/base/context/LoggingContextHolder.java

@@ -5,13 +5,16 @@ import java.util.stream.Stream;
 import javax.servlet.ServletRequest;
 import javax.servlet.http.HttpServletRequest;
 
+import ch.qos.logback.classic.LoggerContext;
 import com.chelvc.framework.base.util.HttpUtils;
 import com.chelvc.framework.common.model.Platform;
 import com.chelvc.framework.common.model.Terminal;
 import com.chelvc.framework.common.util.ObjectUtils;
 import com.chelvc.framework.common.util.StringUtils;
 import lombok.NonNull;
+import org.slf4j.ILoggerFactory;
 import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * 日志上下文工具类
@@ -121,6 +124,17 @@ public final class LoggingContextHolder {
         return buffer.toString();
     }
 
+    /**
+     * 获取日志处理器
+     *
+     * @param name 处理器名称
+     * @return 处理器实例
+     */
+    public static Logger logger(@NonNull String name) {
+        ILoggerFactory factory = LoggerFactory.getILoggerFactory();
+        return factory instanceof LoggerContext ? ((LoggerContext) factory).exists(name) : factory.getLogger(name);
+    }
+
     /**
      * 打印正常日志
      *

framework-base/src/main/java/com/chelvc/framework/base/interceptor/ControllerAccessInterceptor.java

@@ -1,6 +1,5 @@
 package com.chelvc.framework.base.interceptor;
 
-import ch.qos.logback.classic.LoggerContext;
 import com.chelvc.framework.base.context.LoggingContextHolder;
 import com.chelvc.framework.base.context.SessionContextHolder;
 import lombok.extern.slf4j.Slf4j;
@@ -10,9 +9,7 @@ import org.aspectj.lang.Signature;
 import org.aspectj.lang.annotation.Around;
 import org.aspectj.lang.annotation.Aspect;
 import org.aspectj.lang.reflect.MethodSignature;
-import org.slf4j.ILoggerFactory;
 import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.boot.web.servlet.error.ErrorController;
 import org.springframework.core.Ordered;
 import org.springframework.core.annotation.Order;
@@ -50,9 +47,7 @@ public class ControllerAccessInterceptor {
         }
 
         // 获取Logger实例
-        ILoggerFactory factory = LoggerFactory.getILoggerFactory();
-        String name = signature.getDeclaringTypeName() + "." + signature.getName();
-        return factory instanceof LoggerContext ? ((LoggerContext) factory).exists(name) : factory.getLogger(name);
+        return LoggingContextHolder.logger(signature.getDeclaringTypeName() + "." + signature.getName());
     }
 
     /**

framework-kafka/src/main/java/com/chelvc/framework/kafka/config/KafkaConfigurer.java

@@ -9,7 +9,6 @@ import com.chelvc.framework.kafka.fallback.KafkaFallbackPolicy;
 import com.chelvc.framework.kafka.fallback.KafkaMemorySender;
 import com.chelvc.framework.kafka.fallback.KafkaPersistentSender;
 import com.chelvc.framework.kafka.fallback.KafkaStoreClient;
-import com.chelvc.framework.kafka.interceptor.KafkaSessionInterceptor;
 import com.chelvc.framework.kafka.producer.KafkaMessageSender;
 import com.chelvc.framework.kafka.producer.KafkaSenderWrapper;
 import com.chelvc.framework.kafka.producer.KafkaTransactionChecker;
@@ -66,19 +65,14 @@ public class KafkaConfigurer implements KafkaListenerConfigurer {
         };
     }
 
-    @Bean
-    public RecordInterceptor<Object, Object> recordInterceptor() {
-        return new KafkaSessionInterceptor();
-    }
-
     @Bean
     public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory(
             ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
-            ConsumerFactory<Object, Object> consumerFactory) {
+            ConsumerFactory<Object, Object> consumerFactory, RecordInterceptor<Object, Object> interceptor) {
         ConcurrentKafkaListenerContainerFactory<Object, Object> containerFactory =
                 new ConcurrentKafkaListenerContainerFactory<>();
         configurer.configure(containerFactory, consumerFactory);
-        containerFactory.setRecordInterceptor(this.recordInterceptor());
+        containerFactory.setRecordInterceptor(interceptor);
         containerFactory.getContainerProperties().setDeliveryAttemptHeader(true);
         return containerFactory;
     }
@@ -86,12 +80,12 @@ public class KafkaConfigurer implements KafkaListenerConfigurer {
     @Bean
     public KafkaListenerContainerFactory<?> batchListenerContainerFactory(
             ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
-            ConsumerFactory<Object, Object> consumerFactory) {
+            ConsumerFactory<Object, Object> consumerFactory, RecordInterceptor<Object, Object> interceptor) {
         ConcurrentKafkaListenerContainerFactory<Object, Object> containerFactory =
                 new ConcurrentKafkaListenerContainerFactory<>();
         configurer.configure(containerFactory, consumerFactory);
         containerFactory.setBatchListener(true);
-        containerFactory.setRecordInterceptor(this.recordInterceptor());
+        containerFactory.setRecordInterceptor(interceptor);
         containerFactory.getContainerProperties().setDeliveryAttemptHeader(true);
         containerFactory.setMessageConverter(new BatchMessagingMessageConverter(this.recordMessageConverter()));
         return containerFactory;

framework-kafka/src/main/java/com/chelvc/framework/kafka/config/KafkaProperties.java

@@ -6,7 +6,7 @@ import java.util.List;
 
 import com.chelvc.framework.common.util.JacksonUtils;
 import com.chelvc.framework.kafka.fallback.KafkaFallbackPolicy;
-import com.chelvc.framework.kafka.interceptor.KafkaSessionInterceptor;
+import com.chelvc.framework.kafka.interceptor.KafkaRecordInterceptor;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.ToString;
@@ -50,7 +50,7 @@ public class KafkaProperties extends org.springframework.boot.autoconfigure.kafk
         this.getProducer().setKeySerializer(StringSerializer.class);
         this.getProducer().setValueSerializer(JacksonSerializer.class);
         this.getProducer().getProperties().put(
-                ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, KafkaSessionInterceptor.class.getName()
+                ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, KafkaRecordInterceptor.class.getName()
         );
 
         // 初始化消费者默认配置

framework-kafka/src/main/java/com/chelvc/framework/kafka/interceptor/KafkaSessionInterceptor.java → framework-kafka/src/main/java/com/chelvc/framework/kafka/interceptor/KafkaRecordInterceptor.java

@@ -2,6 +2,7 @@ package com.chelvc.framework.kafka.interceptor;
 
 import java.util.Map;
 
+import com.chelvc.framework.base.context.ApplicationContextHolder;
 import com.chelvc.framework.base.context.SessionContextHolder;
 import com.chelvc.framework.kafka.context.KafkaContextHolder;
 import lombok.extern.slf4j.Slf4j;
@@ -10,16 +11,19 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerInterceptor;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
-import org.springframework.kafka.listener.RecordInterceptor;
+import org.springframework.kafka.listener.ConsumerAwareRecordInterceptor;
+import org.springframework.stereotype.Component;
 
 /**
- * Kafka消息会话处理拦截器
+ * Kafka消息处理拦截器
  *
  * @author Woody
  * @date 2024/1/30
  */
 @Slf4j
-public class KafkaSessionInterceptor implements ProducerInterceptor<Object, Object>, RecordInterceptor<Object, Object> {
+@Component
+public class KafkaRecordInterceptor implements ProducerInterceptor<Object, Object>,
+        ConsumerAwareRecordInterceptor<Object, Object> {
     @Override
     public ProducerRecord<Object, Object> onSend(ProducerRecord<Object, Object> record) {
         // 初始化会话信息到消息头
@@ -46,7 +50,16 @@ public class KafkaSessionInterceptor implements ProducerInterceptor<Object, Obje
     }
 
     @Override
-    public ConsumerRecord<Object, Object> intercept(ConsumerRecord<Object, Object> record) {
+    public ConsumerRecord<Object, Object> intercept(ConsumerRecord<Object, Object> record,
+                                                    Consumer<Object, Object> consumer) {
+        // 如果消费者组关闭则忽略该消息，返回null跳过该消息且不会执行afterRecord方法
+        if (!ApplicationContextHolder.getProperty(consumer.groupMetadata().groupId(), boolean.class, true)) {
+            if (log.isDebugEnabled()) {
+                log.debug("Kafka message skipping: {}, {}", consumer.groupMetadata().groupId(), record);
+            }
+            return null;
+        }
+
         // 设置当前会话信息
         SessionContextHolder.setSession(KafkaContextHolder.getSession(record.headers()));
         return record;
@@ -60,7 +73,6 @@ public class KafkaSessionInterceptor implements ProducerInterceptor<Object, Obje
 
     @Override
     public void failure(ConsumerRecord<Object, Object> record, Exception exception, Consumer<Object, Object> consumer) {
-        String topic = record.topic(), group = consumer.groupMetadata().groupId();
-        log.error("Kafka message consume failed: {}:{}, {}", topic, group, record.value(), exception);
+        log.error("Kafka message consume failed: {}, {}", consumer.groupMetadata().groupId(), record, exception);
     }
 }

framework-redis/src/main/java/com/chelvc/framework/redis/queue/MessageStreamListener.java

@@ -5,6 +5,7 @@ import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.Executor;
 
+import com.chelvc.framework.base.context.ApplicationContextHolder;
 import com.chelvc.framework.redis.context.RedisStreamHolder;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.connection.stream.Consumer;
@@ -63,7 +64,7 @@ public class MessageStreamListener<T> implements StreamListener<String, MapRecor
                     this.delayQueue.remove(message);
                 }
             } catch (Throwable t) {
-                log.error("RedisMQ message consume failed: {}, {}", this.consumer, message, t);
+                log.error("RedisMQ message consume failed: {}, {}", this.consumer.getGroup(), message, t);
             }
         }));
     }
@@ -75,10 +76,19 @@ public class MessageStreamListener<T> implements StreamListener<String, MapRecor
      * @return true/false
      */
     private boolean processing(MapRecord<String, String, String> record) {
+        // 如果消费者组关闭则忽略该消息
+        if (!ApplicationContextHolder.getProperty(this.consumer.getGroup(), boolean.class, true)) {
+            if (log.isDebugEnabled()) {
+                log.debug("RedisMQ message skipping: {}, {}", this.consumer.getGroup(), record);
+            }
+            return true;
+        }
+
+        // 消息消费处理
         try {
             RedisStreamHolder.consume(record, this.type, this.listener::consume);
         } catch (Throwable t) {
-            log.error("RedisMQ message consume failed: {}, {}", this.consumer, record, t);
+            log.error("RedisMQ message consume failed: {}, {}", this.consumer.getGroup(), record, t);
             return false;
         }
         return true;
@@ -95,7 +105,7 @@ public class MessageStreamListener<T> implements StreamListener<String, MapRecor
         try {
             this.delayQueue.offer(RedisStreamHolder.serialize(record), timestamp);
         } catch (Throwable t) {
-            log.error("RedisMQ message delaying failed: {}, {}", this.consumer, record, t);
+            log.error("RedisMQ message delaying failed: {}, {}", this.consumer.getGroup(), record, t);
             return false;
         }
         return true;

framework-rocketmq/src/main/java/com/chelvc/framework/rocketmq/consumer/MultipleRocketMQListenerContainer.java

@@ -4,7 +4,6 @@ import java.lang.reflect.Type;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
-import java.util.stream.Collectors;
 
 import com.chelvc.framework.base.context.ApplicationContextHolder;
 import com.chelvc.framework.base.context.Session;
@@ -36,12 +35,11 @@ import org.springframework.core.env.Environment;
  */
 @Slf4j
 public class MultipleRocketMQListenerContainer<T> implements RocketMQListenerContainer<T> {
+    private int batch;
     private Type type;
-    private String topic;
     private String group;
-    private int batch;
-    private Duration duration;
     private Thread thread;
+    private Duration duration;
     private SimpleConsumer consumer;
     private RocketMQListener<T> listener;
     private volatile boolean running = true;
@@ -58,7 +56,7 @@ public class MultipleRocketMQListenerContainer<T> implements RocketMQListenerCon
             if (t.getCause() instanceof InterruptedException) {
                 Thread.currentThread().interrupt();
             } else {
-                log.error("RocketMQ message receive failed: {}:{}", this.topic, this.group, t);
+                log.error("RocketMQ message receive failed: {}", this.group, t);
             }
         }
         return Collections.emptyList();
@@ -72,15 +70,22 @@ public class MultipleRocketMQListenerContainer<T> implements RocketMQListenerCon
      */
     @SuppressWarnings("unchecked")
     private boolean consume(List<MessageView> messages) {
+        // 如果消费者组关闭则忽略该消息
+        if (!ApplicationContextHolder.getProperty(this.group, boolean.class, true)) {
+            if (log.isDebugEnabled()) {
+                log.debug("RocketMQ message skipping: {}, {}", this.group, messages);
+            }
+            return true;
+        }
+
+        // 消息消费处理
         try {
             List<Pair<T, Session>> payloads = RocketMQContextHolder.deserialize(
                     messages, this.type, (payload, session) -> Pair.of((T) payload, session)
             );
             return this.listener.consume(payloads);
         } catch (Throwable t) {
-            String ids = messages.stream().map(MessageView::getMessageId).map(String::valueOf)
-                    .collect(Collectors.joining(","));
-            log.error("RocketMQ message consume failed: {}:{}, {}", this.topic, this.group, ids, t);
+            log.error("RocketMQ message consume failed: {}, {}", this.group, messages, t);
         }
         return false;
     }
@@ -98,8 +103,7 @@ public class MultipleRocketMQListenerContainer<T> implements RocketMQListenerCon
                 if (t.getCause() instanceof InterruptedException) {
                     Thread.currentThread().interrupt();
                 } else {
-                    String id = String.valueOf(message.getMessageId());
-                    log.error("RocketMQ message acknowledge failed: {}:{}, {}", this.topic, this.group, id, t);
+                    log.error("RocketMQ message acknowledge failed: {}, {}", this.group, message, t);
                 }
             }
         }
@@ -113,17 +117,8 @@ public class MultipleRocketMQListenerContainer<T> implements RocketMQListenerCon
     private Thread startup() {
         return ThreadUtils.run(() -> {
             while (this.running && !Thread.currentThread().isInterrupted()) {
-                // 批量接收消息
                 List<MessageView> messages = this.receive();
-                if (ObjectUtils.isEmpty(messages)) {
-                    continue;
-                }
-
-                // 消费消息
-                boolean success = this.consume(messages);
-
-                // 回执消息
-                if (success) {
+                if (ObjectUtils.notEmpty(messages) && this.consume(messages)) {
                     this.acknowledge(messages);
                 }
             }
@@ -144,17 +139,17 @@ public class MultipleRocketMQListenerContainer<T> implements RocketMQListenerCon
 
         // 初始化消费者实例
         this.type = type;
-        Environment environment = ApplicationContextHolder.getEnvironment();
-        this.topic = RocketMQContextHolder.isolate(environment.resolvePlaceholders(annotation.topic()));
-        this.group = RocketMQContextHolder.isolate(environment.resolvePlaceholders(annotation.group()));
         this.batch = annotation.batch();
-        this.duration = Duration.ofSeconds(annotation.duration());
         this.listener = listener;
+        this.duration = Duration.ofSeconds(annotation.duration());
+        Environment environment = ApplicationContextHolder.getEnvironment();
+        this.group = RocketMQContextHolder.isolate(environment.resolvePlaceholders(annotation.group()));
+        String topic = RocketMQContextHolder.isolate(environment.resolvePlaceholders(annotation.topic()));
         Duration interval = Duration.ofSeconds(annotation.interval());
         FilterExpression filter = new FilterExpression(annotation.tag(), FilterExpressionType.TAG);
         SimpleConsumerBuilder builder = provider.newSimpleConsumerBuilder().setClientConfiguration(configuration)
                 .setConsumerGroup(this.group).setAwaitDuration(interval)
-                .setSubscriptionExpressions(Collections.singletonMap(this.topic, filter));
+                .setSubscriptionExpressions(Collections.singletonMap(topic, filter));
         try {
             this.consumer = builder.build();
         } catch (ClientException e) {
@@ -177,7 +172,7 @@ public class MultipleRocketMQListenerContainer<T> implements RocketMQListenerCon
                 try {
                     this.consumer.close();
                 } catch (Throwable t) {
-                    log.warn("RocketMQ consumer shutdown failed: {}:{}, {}", this.topic, this.group, t.getMessage());
+                    log.warn("RocketMQ consumer shutdown failed: {}, {}", this.group, t.getMessage());
                 }
             }
         }

framework-rocketmq/src/main/java/com/chelvc/framework/rocketmq/consumer/SingleRocketMQListenerContainer.java

@@ -31,7 +31,6 @@ import org.springframework.core.env.Environment;
 @Slf4j
 public class SingleRocketMQListenerContainer<T> implements RocketMQListenerContainer<T> {
     private Type type;
-    private String topic;
     private String group;
     private PushConsumer consumer;
     private RocketMQListener<T> listener;
@@ -44,14 +43,22 @@ public class SingleRocketMQListenerContainer<T> implements RocketMQListenerConta
      */
     @SuppressWarnings("unchecked")
     private ConsumeResult consume(MessageView message) {
+        // 如果消费者组关闭则忽略该消息
+        if (!ApplicationContextHolder.getProperty(this.group, boolean.class, true)) {
+            if (log.isDebugEnabled()) {
+                log.debug("RocketMQ message skipping: {}, {}", this.group, message);
+            }
+            return ConsumeResult.SUCCESS;
+        }
+
+        // 消息消费处理
         try {
             return RocketMQContextHolder.deserialize(message, this.type, payload -> {
                 this.listener.consume((T) payload);
                 return ConsumeResult.SUCCESS;
             });
         } catch (Throwable t) {
-            String id = String.valueOf(message.getMessageId());
-            log.error("RocketMQ message consume failed: {}:{}, {}", this.topic, this.group, id, t);
+            log.error("RocketMQ message consume failed: {}, {}", this.group, message, t);
             return ConsumeResult.FAILURE;
         }
     }
@@ -68,15 +75,14 @@ public class SingleRocketMQListenerContainer<T> implements RocketMQListenerConta
 
         // 初始化消费者实例
         this.type = type;
+        this.listener = listener;
         Environment environment = ApplicationContextHolder.getEnvironment();
-        this.topic = RocketMQContextHolder.isolate(environment.resolvePlaceholders(annotation.topic()));
         this.group = RocketMQContextHolder.isolate(environment.resolvePlaceholders(annotation.group()));
-        this.listener = listener;
+        String topic = RocketMQContextHolder.isolate(environment.resolvePlaceholders(annotation.topic()));
         FilterExpression filter = new FilterExpression(annotation.tag(), FilterExpressionType.TAG);
         PushConsumerBuilder builder = provider.newPushConsumerBuilder().setClientConfiguration(configuration)
                 .setConsumerGroup(this.group).setConsumptionThreadCount(annotation.concurrency())
-                .setSubscriptionExpressions(Collections.singletonMap(this.topic, filter))
-                .setMessageListener(this::consume);
+                .setSubscriptionExpressions(Collections.singletonMap(topic, filter)).setMessageListener(this::consume);
         try {
             this.consumer = builder.build();
         } catch (ClientException e) {
@@ -90,7 +96,7 @@ public class SingleRocketMQListenerContainer<T> implements RocketMQListenerConta
             try {
                 this.consumer.close();
             } catch (Throwable t) {
-                log.warn("RocketMQ consumer shutdown failed: {}:{}, {}", this.topic, this.group, t.getMessage());
+                log.warn("RocketMQ consumer shutdown failed: {}, {}", this.group, t.getMessage());
             }
         }
     }