package com.zzsn.job;


import com.alibaba.fastjson.JSON;
import com.zzsn.crawler.WeixinDetailThread;
import com.zzsn.crawler.WeixinSiteThread;
import com.zzsn.entity.SiteMsgTemple;
import com.zzsn.util.Constants;
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.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
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.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Component
@EnableScheduling
@Slf4j
public class KafkaConsumerJob {

    private static KafkaConsumer<String, String> createConsumer() {
        Properties properties = new Properties();
        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);
    }




    /**
     * 从kafka中获取公众号信息进行发送获取列表内容提取链接
     */
    @Scheduled(cron = "0 0/3 * * * ?")
    @Async("asyncTaskExecutor")
    public void wxOfficialConsumer (){
        ExecutorService threadPool = Executors.newSingleThreadExecutor();
        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){
                    SiteMsgTemple siteMsgTemple = JSON.parseObject(record.value().toString(), SiteMsgTemple.class);
                    WeixinSiteThread siteThread=new WeixinSiteThread();
                    siteThread.siteMsgTemple=siteMsgTemple;
                    //创建使用固定线程数的线程池
                    threadPool.execute(siteThread);
                    TimeUnit.SECONDS.sleep(20);
                }
            }
        }catch (Exception e){
            consumer = createConsumer();
            consumer.subscribe(Arrays.asList(Constants.KAFKA_CONSUMER_TOPIC));
        }
        threadPool.shutdown();
        while(true)
        {
            boolean isfinished = threadPool.isTerminated();
            if(isfinished)
                break;
        }
    }
    ExecutorService serviceDetail = Executors.newFixedThreadPool(1);
    /**
     * 从kafka中获取微信资讯url进行解析详情
     */
//    @Scheduled(cron = "0 0/2 * * * ?")
    public void wxDetailconsumer (){
        log.info("定时获取mq消息");
        //1.创建消费者
        KafkaConsumer<String, String> consumer = createConsumer();
        consumer.subscribe(Arrays.asList(Constants.KAFKA_WXDETAILURL_TOPIC));

        try{
            while(true){
                //消费者是一个长期运行的程序，通过持续轮询向Kafka请求数据。在其他线程中调用consumer.wakeup()可以退出循环
                //在0ms内等待Kafka的broker返回数据.超时参数指定poll在多久之后可以返回，不管有没有可用的数据都要返回
                ConsumerRecords<String, String> records = consumer.poll(0);
                consumer.commitSync();

                for(ConsumerRecord record : records){
                    SiteMsgTemple siteMsgTemple = JSON.parseObject(record.value().toString(), SiteMsgTemple.class);
                    WeixinDetailThread siteThread=new WeixinDetailThread();
                    siteThread.siteMsgTemple=siteMsgTemple;
//                    siteThread.start();
                    //创建使用固定线程数的线程池
                    serviceDetail.execute(()->{
                        String threadName= Thread.currentThread().getName();
                        System.out.println(threadName+"开始执行");
                        try {
                            siteThread.start();
                            TimeUnit.SECONDS.sleep(2);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println(threadName+"执行结束");
                    });
                    TimeUnit.SECONDS.sleep(10);
                }
            }
        }catch (Exception e){
            consumer = createConsumer();
            consumer.subscribe(Arrays.asList(Constants.KAFKA_WXDETAILURL_TOPIC));
        }

    }

    private static KafkaConsumer<String, String> create2Consumer() {
        Properties props = new Properties();
        // 必须设置的属性
        props.put("bootstrap.servers", "114.115.159.144:9092");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("group.id", "group1");
        // 可选设置属性
        props.put("enable.auto.commit", "true");
        // 自动提交offset,每1s提交一次
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset","earliest ");
        props.put("client.id", "es-sync");
        return new KafkaConsumer<>(props);
    }
}
