package com.zzsn.job;

import com.google.gson.Gson;
import com.zzsn.search.entity.KeywordMsg;
import com.zzsn.search.googleThread.DetailGoogleSearchThread;
import com.zzsn.search.googleThread.WebGoogleSearchThread;
import com.zzsn.search.yaooThread.DetailYahooSearchThread;
import com.zzsn.search.yaooThread.WebYahooSearchThread;
import com.zzsn.utility.index.Constants;
import com.zzsn.utility.model.CatchWebByMetaSearch;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
@EnableScheduling
@Slf4j
public class KafkaConsumerYahooTask {


    private static KafkaConsumer<String, String> createConsumer() {
        Properties properties = new Properties();
        System.out.println(Constants.KAFKA_CONSUMER_SERVERS);
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, Constants.KAFKA_CONSUMER_SERVERS);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, Constants.KAFKA_CONSUMER_GROUP_ID);
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        //kafka数据的读取方式
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,Constants.KAFKA_CONSUMER_AUTO_OFFSET_RESET);
//         latest earliest
        //时间间隔设置为1h
        properties.put("max.poll.interval.ms", 60*60*1000);
        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
        return new KafkaConsumer<>(properties);
    }

    //打包编译时修改定时启动的任务 如果是搜索关键词放开consumerKeyword 如果是内容解析放开consumerDetailUrl 上面的定时任务
    @Scheduled(cron = "0 0/5 * * * ?")
    @Async("webExecutor")
    public void consumerKeyword (){
        log.info("定时获取mq消息");
        //1.创建消费者
        KafkaConsumer<String, String> consumer = createConsumer();
        consumer.subscribe(Arrays.asList(Constants.KAFKA_CONSUMER_TOPIC));
        try{
            while(true){
                //消费者是一个长期运行的程序，通过持续轮询向Kafka请求数据。在其他线程中调用consumer.wakeup()可以退出循环
                //在0ms内等待Kafka的broker返回数据.超时参数指定poll在多久之后可以返回，不管有没有可用的数据都要返回
                ConsumerRecords<String, String> records = consumer.poll(0);
                consumer.commitSync();
                for(ConsumerRecord record : records){
                    KeywordMsg keywordMsg = new Gson().fromJson(record.value().toString(), KeywordMsg.class);
                    WebYahooSearchThread metaSearchThread=new WebYahooSearchThread();
                    metaSearchThread.keywordMsg=keywordMsg;
                    metaSearchThread.crawler();
                }
            }
        }catch (Exception e){
            consumer = createConsumer();
            consumer.subscribe(Arrays.asList(Constants.KAFKA_CONSUMER_TOPIC));
        }

    }

//    @Scheduled(cron = "0 0/5 * * * ?")
    @Async("detailExecutor")
    public void consumerDetailUrl (){
        log.info("定时获取mq2消息");
        //1.创建消费者
        KafkaConsumer<String, String> consumer = createConsumer();
        consumer.subscribe(Arrays.asList(Constants.KAFKA_PRODUCT_GOOGLE_URLLIST_TOPIC));
        ExecutorService poolExecuter = Executors.newFixedThreadPool(Constants.THREAD_SIZE);
        try{
            while(true){
                //消费者是一个长期运行的程序，通过持续轮询向Kafka请求数据。在其他线程中调用consumer.wakeup()可以退出循环
                //在0ms内等待Kafka的broker返回数据.超时参数指定poll在多久之后可以返回，不管有没有可用的数据都要返回
                ConsumerRecords<String, String> records = consumer.poll(0);
                consumer.commitSync();
                for(ConsumerRecord record : records){
                    CatchWebByMetaSearch catchWebByMetaSearch = new Gson().fromJson(record.value().toString(), CatchWebByMetaSearch.class);
                    DetailYahooSearchThread detailBaiduSearchThread=new DetailYahooSearchThread();
                    detailBaiduSearchThread.catchWebByMetaSearch=catchWebByMetaSearch;
                    detailBaiduSearchThread.crawler();
                }
            }
        }catch (Exception e){
            consumer = createConsumer();
            consumer.subscribe(Arrays.asList(Constants.KAFKA_PRODUCT_URLLIST_TOPIC));
        }

    }


}
