目 录CONTENT

文章目录

多topic注入kafka消费者

Carol
2023-02-02 / 0 评论 / 0 点赞 / 88 阅读 / 0 字 / 正在检测是否收录...

😋特惠💰

多个付费教程捆绑包限时特惠,8折打包购买: 详情点击爱发电 ~~~

多topic注入kafka消费者

@KafkaListener() 两种方式加载多个topic

# application.properties
spring.kafkaListenerList = topic1,topic2

第一种

通过EL表达式进行注入

@KafkaListener(topics = "#{'${spring.kafkaListenerList}'.split(',')}")

第二种

通过解析数据生成对象进行注入

KafkaListenerReceiver是自定义的消费对象

  • KafkaListenerConfig.java
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.Ordered;
import org.springframework.core.PriorityOrdered;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;


@Configuration
@Slf4j
public class KafkaListenerConfig implements BeanDefinitionRegistryPostProcessor, PriorityOrdered {

    private List<String> listenerList;

    KafkaListenerConfig() throws IOException {
        Properties pro = new Properties();
        File f;
        f =new File( java.net.URLDecoder.decode(this.getClass().getResource("/application.properties").getPath(),"utf-8"));
        FileInputStream in = new FileInputStream(f);
        pro.load(in);
        in.close();
        String listString = pro.getProperty("spring.verify.kafkaListenerList");
        if (StringUtils.isEmpty(listString)){
            log.error("spring.verify.kafkaListenerList is empty or null");
            System.exit(0);
        }
        log.info("spring.kafkaListenerList: "+listString);
        listenerList = Arrays.asList(listString.split(","));
    }

    public List<String> getListenerList() {
        return listenerList;
    }

    public void setListenerList(List<String> listenerList) {
        this.listenerList = listenerList;
    }

    @Override
    public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry beanDefinitionRegistry) throws BeansException {
        for (String listener :  listenerList) {
            BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(KafkaListenerReceiver.class);
            builder.addConstructorArgValue(listener);
            beanDefinitionRegistry.registerBeanDefinition(listener, builder.getBeanDefinition());
        }
    }

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory configurableListableBeanFactory) throws BeansException {

    }

    @Override
    public int getOrder() {
        return Ordered.HIGHEST_PRECEDENCE;
    }
}
  • KafkaListenerReceiver.java
@KafkaListener(topics = "#{__listener.topic}")
0
  1. 支付宝打赏

    qrcode alipay
  2. 微信打赏

    qrcode weixin
🔥🔥🔥阿里云特惠购买,点击链接更享折上折~ 详情点击链接查看活动 ~~~

评论区