动态修改topics

Author: lyb-geek

springboot-mq-idempotent-consume/ReadME.md

@@ -7,4 +7,9 @@
 - 3、如果该ID已经被消费过，则不进行处理
 - 4、如果该ID没被消费过，则进行处理，处理结束后，把该ID存入redis，且value设置为已消费
 
+其他额外功能
+
+- 通过生产者拦截器，改写topics
+- 通过动态修改注解参数，改写@kafkaListener的topics
+
 > 注：消费端必须开启手动确认，不能是自动确认

springboot-mq-idempotent-consume/pom.xml

@@ -24,6 +24,15 @@
             <groupId>org.springframework.kafka</groupId>
             <artifactId>spring-kafka</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.reflections</groupId>
+            <artifactId>reflections</artifactId>
+        </dependency>
 
     </dependencies>
 

springboot-mq-idempotent-consume/src/main/java/com/github/lybgeek/kafka/config/KakfaConfig.java

@@ -1,6 +1,7 @@
 package com.github.lybgeek.kafka.config;
 
 
+import com.github.lybgeek.kafka.producer.interceptor.KafkaProducerInterceptor;
 import com.github.lybgeek.kafka.serialization.ObjectDeserializer;
 import com.github.lybgeek.kafka.serialization.ObjectSerializer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -19,7 +20,9 @@
 import org.springframework.kafka.core.*;
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 
@@ -82,6 +85,10 @@ private Map<String, Object> producerConfigs() {
         props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ObjectSerializer.class);
+        //定义拦截器
+        List<String> interceptors = new ArrayList<>();
+        interceptors.add(KafkaProducerInterceptor.class.getName());
+        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
         return props;
     }
 

springboot-mq-idempotent-consume/src/main/java/com/github/lybgeek/kafka/constant/Constant.java

@@ -3,4 +3,6 @@
 
 public class Constant {
     public static final String TOPIC = "msgTopic";
+
+    public static final String TOPIC_KEY_PREFIX = "test_";
 }

springboot-mq-idempotent-consume/src/main/java/com/github/lybgeek/kafka/consumer/factory/KafkaListenerFactoryBeanPostProcesser.java

@@ -0,0 +1,76 @@
+package com.github.lybgeek.kafka.consumer.factory;
+
+import com.github.lybgeek.kafka.constant.Constant;
+import com.google.common.collect.Lists;
+import lombok.SneakyThrows;
+import org.reflections.Reflections;
+import org.reflections.scanners.FieldAnnotationsScanner;
+import org.reflections.scanners.MethodAnnotationsScanner;
+import org.reflections.scanners.MethodParameterScanner;
+import org.reflections.scanners.SubTypesScanner;
+import org.reflections.util.ConfigurationBuilder;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.boot.autoconfigure.AutoConfigurationPackages;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor;
+import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ *  {@link KafkaListenerAnnotationBeanPostProcessor#postProcessAfterInitialization(java.lang.Object, java.lang.String)}
+ */
+@Component
+public class KafkaListenerFactoryBeanPostProcesser implements BeanFactoryPostProcessor {
+
+    @SneakyThrows
+    @Override
+    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
+
+        List<String> packageNames = AutoConfigurationPackages.get(beanFactory);
+
+        for (String packageName : packageNames) {
+            Reflections reflections = new Reflections(new ConfigurationBuilder()
+                    .forPackages(packageName) // 指定路径URL
+                    .addScanners(new SubTypesScanner()) // 添加子类扫描工具
+                    .addScanners(new FieldAnnotationsScanner()) // 添加 属性注解扫描工具
+                    .addScanners(new MethodAnnotationsScanner() ) // 添加 方法注解扫描工具
+                    .addScanners(new MethodParameterScanner() ) // 添加方法参数扫描工具
+            );
+
+            Set<Method> methodSet = reflections.getMethodsAnnotatedWith(KafkaListener.class);
+            if(!CollectionUtils.isEmpty(methodSet)){
+                for (Method method : methodSet) {
+                    KafkaListener kafkaListener = method.getAnnotation(KafkaListener.class);
+                    changeTopics(kafkaListener);
+                }
+            }
+        }
+
+    }
+
+
+    private void changeTopics(KafkaListener kafkaListener) throws Exception{
+        InvocationHandler invocationHandler = Proxy.getInvocationHandler(kafkaListener);
+        Field memberValuesField = invocationHandler.getClass().getDeclaredField("memberValues");
+        memberValuesField.setAccessible(true);
+        Map<String,Object> memberValues = (Map<String,Object>)memberValuesField.get(invocationHandler);
+        String[] topics = (String[])memberValues.get("topics");
+        System.out.println("修改前topics：" + Lists.newArrayList(topics));
+        for (int i = 0; i < topics.length; i++) {
+            topics[i] = Constant.TOPIC_KEY_PREFIX + topics[i];
+        }
+        memberValues.put("topics", topics);
+        System.out.println("修改后topics：" + Lists.newArrayList(kafkaListener.topics()));
+
+    }
+}

springboot-mq-idempotent-consume/src/main/java/com/github/lybgeek/kafka/producer/interceptor/KafkaProducerInterceptor.java

@@ -0,0 +1,63 @@
+package com.github.lybgeek.kafka.producer.interceptor;
+
+import com.alibaba.fastjson.JSON;
+import com.github.lybgeek.kafka.dto.MessageDTO;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.producer.ProducerInterceptor;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+
+import java.util.Map;
+
+import static com.github.lybgeek.kafka.constant.Constant.TOPIC_KEY_PREFIX;
+
+
+
+@Slf4j
+public class KafkaProducerInterceptor implements ProducerInterceptor<String, MessageDTO> {
+
+
+
+    /**
+     * 运行在用户主线程中，在消息被序列化之前调用
+     * @param record
+     * @return
+     */
+    @Override
+    public ProducerRecord<String, MessageDTO> onSend(ProducerRecord<String, MessageDTO> record) {
+        log.info("onSend,orginalRecord:{}", JSON.toJSONString(record));
+        return new ProducerRecord<String, MessageDTO>(TOPIC_KEY_PREFIX + record.topic(),
+                record.partition(),record.timestamp(),record.key(), record.value());
+    }
+
+
+
+
+    /**
+     * 在消息被应答之前或者消息发送失败时调用，通常在producer回调逻辑触发之前，运行在produer的io线程中
+     * @param metadata
+     * @param exception
+     */
+    @Override
+    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+
+    }
+
+
+    /**
+     *  清理工作
+     */
+    @Override
+    public void close() {
+    }
+
+
+    /**
+     * 初始化工作
+     * @param configs
+     */
+    @Override
+    public void configure(Map<String, ?> configs) {
+
+    }
+}

springboot-mq-idempotent-consume/src/main/resources/application.yml

@@ -33,6 +33,9 @@ spring:
       # acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
       # acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
       acks: 1
+      # 生产者拦截器配置
+      properties:
+        interceptor.classes: com.github.lybgeek.kafka.producer.interceptor.KafkaProducerInterceptor
     consumer:
       # 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D
       auto-commit-interval: 1S