package com.yeejoin.amos.api.alarm.service.impl;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;



/**
 * @description: 监听设备告警信息
 * @author: tw
 * @createDate: 2023/6/27
 */

@Service
public class AlarmKafkaConsumer {

    @Autowired
    PointSystemServiceImpl pointSystemServiceImpl;

    //消费者来处理消息
    @KafkaListener(id="user1" , topics ={"${kafka.equipment.alarm}"})
    public void message1( String record, Acknowledgment ack){
        // 处理业务
        String date=record;
        //异步触发预警
        pointSystemServiceImpl.sendWarningAsync(date);
        //手动提交
        ack.acknowledge();

    }


//    public void message1( ConsumerRecord<?, ?> record, Acknowledgment ack){
//        // 消费的哪个topic、partition的消息,打印出消息内容
//
//        StringBuffer sb = new StringBuffer();
//        // 主题
//        sb.append(record.topic() + "-");
//        // 分区
//        sb.append(record.partition() + "-");
//        // 需要消费的值
//        sb.append(record.value() + "-");
//        // 位移
//        sb.append(record.offset());
//
//        System.out.println( "消费者进行消费："+ sb);
//        ack.acknowledge();
//
//    }
//    // 简单消费者，groupId可以任意起
//    @KafkaListener(id = "Consumer0", groupId = "jf0-group", topics = "jf1", topicPartitions = {
//            @TopicPartition(topic = "jf1", partitions = {"0"}),
//    }, containerFactory = "kafkaListenerContainerFactory")
//    public void consumer0(ConsumerRecord<String, String> records, Acknowledgment ack) {
//        this.message1(records,ack);
//    }
//
//    @KafkaListener(id = "Consumer1", groupId = "jf1-group", topics = "jf1", topicPartitions = {
//            @TopicPartition(topic = "jf1", partitions = {"1"}),
//    }, containerFactory = "kafkaListenerContainerFactory")
//    public void consumer1(ConsumerRecord<String, String> records, Acknowledgment ack) {
//        this.message1(records,ack);
//    }
//
//    @KafkaListener(id = "Consumer2", groupId = "jf2-group", topics = "jf1", topicPartitions = {
//            @TopicPartition(topic = "jf1", partitions = {"2"}),
//    }, containerFactory = "kafkaListenerContainerFactory")
//    public void consumer3(ConsumerRecord<String, String> records, Acknowledgment ack) {
//        this.message1(records,ack);
//    }

}
