接收kafka消息

港控/mmm° 2022-12-10 02:28 257阅读 0赞

kafka:
      server: 127.0.0.1:8081,127.0.0.1:8082,127.0.0.1:8083,127.0.0.1:8084
      topics: ESEAL-HR-TEST-2
    gourp:
      id: HTHR

@Value("${kafka.server}")
    String servers;
    @Value("${group.id:HR}")
    String groupId;
    @Value("${enable.auto.commit:true}")
    Boolean autoCommit;
    @Value("${auto.commit.interval.ms:1000}")
    int autoCommitInterval;
    @Value("${session.timeout.ms:10000}")
    int sessionTimeout;
        @Value("${key.deserializer:org.apache.kafka.common.serialization.StringDeserializer}")
    String key;
        @Value("${value.deserializer:org.apache.kafka.common.serialization.StringDeserializer}")
    String value;
    @Value("${kafka.topics}")
    String topic;
    @Value("${auto.offset.reset:earliest}")
    String resetOffset;
    
    @PostConstruct
    public void initKafka(){
        Properties props = new Properties();
        // Kafka集群，多台服务器地址之间用逗号隔开
        props.put("bootstrap.servers", servers);
        // 消费组ID
        props.put("group.id", groupId);
        // Consumer的offset是否自动提交
        props.put("enable.auto.commit", autoCommit);
        props.put("session.timeout.ms",sessionTimeout);
        // 自动提交offset到zk的时间间隔，时间单位是毫秒
        props.put("auto.commit.interval.ms", autoCommitInterval);
        props.put("auto.offset.reset", resetOffset);
        // 消息的反序列化类型
        props.put("key.deserializer", key);
        props.put("value.deserializer", value);
        KafkaConsumer<String, String> kafkaConsumer ;
        kafkaConsumer = new KafkaConsumer<>(props);
        // 订阅的话题
        kafkaConsumer.subscribe(Arrays.asList(topic));
        Thread t = new Thread(new Runnable(){
            public void run(){
                // Consumer调用poll方法来轮询Kafka集群的消息，一直等到Kafka集群中没有消息或者达到超时时间100ms为止
                while (true) {
                    ConsumerRecords<String, String> records = kafkaConsumer.poll(0);
                    try {
                        for (ConsumerRecord<String, String> record : records) {
                            log.info("offset = {}, key = {}, value = {}", record.offset(), record.key(), record.value());
                            MessageNoticeDto dto = JSONUtil.toBean(record.value(), MessageNoticeDto.class);
                            log.info("返回信息："+JSONUtil.toJson(dto));
                            if(null != dto){
                                //todo 返回信息处理
                                
                            }
                        }
                    }catch (Exception e){
                        log.error("异常信息：" + e.toString());
                    }
                }
            }
        });
        //启动kafka增加开关控制
        String kafkaFlag = profileClient.getProfileValueByOptions("HHR.ONBOARD.KAFKA.EXC.FLAG");
        if(StringUtil.equalsIgnoreCase(kafkaFlag,"Y")){
            log.info("开始消费kafka的消息");
            t.start();
        }
    }

@PostConstruct该注解被用来修饰一个非静态的void（）方法。被@PostConstruct修饰的方法会在服务器加载Servlet的时候运行，并且只会被服务器执行一次。PostConstruct在构造函数之后执行，init（）方法之前执行。

通常我们会是在Spring框架中使用到@PostConstruct注解 该注解的方法在整个Bean初始化中的执行顺序：

Constructor(构造方法) -> @Autowired(依赖注入) -> @PostConstruct(注释的方法)