package com.itheima.kafka.producer;

import com.itheima.kafka.producer.mq.KFProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;

import java.util.concurrent.ExecutionException;

/**
 * Kafka生产者启动类
 */
@SpringBootApplication
public class KafkaProducerApplication {

    // 定义日志记录器
    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProducerApplication.class) ;

    // 定义发送消息的模板类
    private static KafkaTemplate kafkaTemplate ;

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        // 初始化环境
        ConfigurableApplicationContext applicationContext = SpringApplication.run(KafkaProducerApplication.class, args);
        kafkaTemplate = applicationContext.getBean(KafkaTemplate.class) ;

        // 测试Kafka环境
        // environmentTest() ;

        // 消息的异步发送
        // sendMessageMethod03();

        // 消息顺序性保障
        // sendMessageToDestPartition() ;

        // 事务消息的测试
        // sendTransactionMessageMethod01() ;
        sendMessageTx(applicationContext) ;
    }

    private static void sendMessageTx(ConfigurableApplicationContext applicationContext) {
        KFProducer kfProducer = applicationContext.getBean(KFProducer.class);
        kfProducer.sendTransactionMessage();
    }


    // Kafka单节点环境测试
    public static void environmentTest(){
        kafkaTemplate.send("itheima" , "Kakfa环境测试....") ;
    }

    // 演示消息发送： 异步发送
    public static void sendMessageMethod03() throws ExecutionException, InterruptedException {

        for(int x = 0 ; x < 5 ; x++) {

            // Kafka消息的异步发送
            String msg = "Kakfa环境测试...." + x ;
            kafkaTemplate.send("itheima" , msg).addCallback((obj) -> {
                LOGGER.info("send msg to kafka broker success ---> {} " , ((SendResult)obj).getProducerRecord().value());
            } , (t) -> {
                t.printStackTrace();
            });
            LOGGER.info("send msg to local cache success ---> {} " , msg);
        }

    }

    // 发送消息到指定的分区，保证分区的消息顺序性
    public static void sendMessageToDestPartition() {

        for(int x = 0 ; x < 5 ; x++) {

            // Kafka消息的异步发送
            String msg = "Kakfa环境测试...." + x ;
            kafkaTemplate.send("itcast01" , 0 , "order"  , msg).addCallback((obj) -> {
                LOGGER.info("send msg to kafka broker success ---> {} " , ((SendResult)obj).getProducerRecord().value());
            } , (t) -> {
                t.printStackTrace();
            });

            LOGGER.info("send msg to local cache success ---> {} " , msg);
        }

    }

    // 事务消息发送的第一种方式
    public static void sendTransactionMessageMethod01() {

        // 发送事务消息
        kafkaTemplate.executeInTransaction((operations) -> {

            // 发送消息
            kafkaTemplate.send("itcast01" , 0 , "order" , "事务消息 ----> 1") ;
            kafkaTemplate.send("itcast01" , 0 , "order" , "事务消息 ----> 2") ;

            // 产生异常代码
            int a = 1 / 0 ;

            kafkaTemplate.send("itcast01" , 0 , "order" , "事务消息 ----> 3") ;

            // 返回true，表示发送成功
            return true ;

        }) ;

        LOGGER.info("send transaction message to local cache success ");
    }

}
