接收kafka消息-蒲公英云

接收kafka消息

港控/mmm° 2022-12-10 02:28 341阅读 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();
    }
}