//package com.base.biz.mq;
//
//import com.base.biz.mq.factory.HandlerMqFactory;
//import com.base.orbit.mq.MqServer;
//import com.base.orbit.mq.handler.MessageHandler;
//import com.base.orbit.mq.kafka.KafkaMqServer;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.kafka.clients.consumer.KafkaConsumer;
//import org.apache.kafka.common.serialization.StringDeserializer;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.boot.context.event.ApplicationStartedEvent;
//import org.springframework.context.ApplicationListener;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.Resource;
//import java.util.List;
//import java.util.Objects;
//import java.util.Properties;
//
///**
// * @author jhzhu
// * @date 2023/2/6 10:46
// * @Description: mq接收者
// */
//@Component
//@Slf4j
//public class MqReceiver implements ApplicationListener<ApplicationStartedEvent> {
//
//    @Value("${kafka.service.url.list}")
//    private String kafkaServer;
//
//    @Value("${kafka.service.groupId:YKFZ_GROUP}")
//    private String groupId;
//
//    //  是否自动提交
//    @Value("${kafka.service.enableAutoCommit:true}")
//    private String enableAutoCommit = "true";
//
//    //  自动提交间隔时间
//    @Value("${kafka.service.autoCommitIntervalMs:3000}")
//    private String autoCommitIntervalMs = "3000";
//
//    @Value("${kafka.service.topics:YKSC,YKSC_SCOUTREQ,TASKISSUED-yksufz}")
//    private List<String> testTopics;
//
//    @Override
//    public void onApplicationEvent(ApplicationStartedEvent applicationStartedEvent) {
//        Properties properties = new Properties();
//        properties.setProperty("bootstrap.servers", kafkaServer);
//        properties.setProperty("group.id", groupId);
//        properties.setProperty("key.deserializer", StringDeserializer.class.getName()); //  消息格式均为byte[]数组，key和value需要进行反序列化
//        properties.setProperty("value.deserializer", StringDeserializer.class.getName());
//        properties.setProperty("enable.auto.commit", enableAutoCommit);
//        properties.setProperty("auto.commit.interval.ms", autoCommitIntervalMs);
//        properties.setProperty("auto.offset.reset", "earliest");
//
//        testTopics.stream().forEach(testTopic -> {
//            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
//            MessageHandler testHandler = HandlerMqFactory.getInvokeHandlerByTopic(testTopic);
//            Objects.requireNonNull(testHandler, "handler not found by topic is " + testTopics + "");
//            MqServer mqServer = new KafkaMqServer(consumer, testHandler, testTopic);
//            mqServer.start();
//        });
//    }
//}