control.command.kafka.enabled=true
control.command.kafka.bootstrap-servers=172.0.0.1:9092
control.command.kafka.command-topics=lastTopic
control.command.kafka.consumer.group-id=consumer-eslink-iwater-control-command
control.command.kafka.consumer.properties.session.timeout.ms=30000
control.command.kafka.consumer.properties.request.timeout.ms=90000
control.command.kafka.consumer.fetch-min-size=10KB
control.command.kafka.consumer.fetch-max-wait=500
control.command.kafka.consumer.max-poll-records=1000
control.command.kafka.consumer.auto-offset-reset=earliest
control.command.kafka.listener.ack-mode=MANUAL_IMMEDIATE
control.command.kafka.listener.concurrency=1
control.command.kafka.listener.type=SINGLE
control.command.kafka.producer.acks=all
control.command.kafka.producer.batchSize=4096
control.command.kafka.producer.bufferMemory=40960
control.command.kafka.producer.linger=10
control.command.kafka.producer.retries=3

@Configuration
@ConditionalOnExpression("${control.command.kafka.enabled:false}")
@EnableKafka
public class ControlCommandKafkaConfig {

    @Bean("controlCommandKafkaProperties")
    @ConfigurationProperties("control.command.kafka")
    @Primary
    public KafkaProperties kafkaProperties() {
        return new KafkaProperties();
    }

    @Bean("controlCommandKafkaConsumerFactory")
    public ConsumerFactory<Object, Object> kafkaConsumerFactory(
            @Qualifier("controlCommandKafkaProperties") KafkaProperties inProps) {
        DefaultKafkaConsumerFactory<Object, Object> consumerFactory = new DefaultKafkaConsumerFactory<>(inProps.buildConsumerProperties());
        return consumerFactory;
    }

    @Bean("controlCommandBatchFactory")
    @DependsOn("controlCommandKafkaProperties")
    public KafkaListenerContainerFactory<?> egBatchFactory(@Qualifier("controlCommandKafkaProperties") KafkaProperties inProps,
                                                           @Qualifier("controlCommandKafkaConsumerFactory") ConsumerFactory consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> listenerFactory =
                new ConcurrentKafkaListenerContainerFactory<>();
        listenerFactory.setConsumerFactory(consumerFactory);
        configureListenerFactory(listenerFactory, inProps);
        configureContainer(listenerFactory.getContainerProperties(), inProps);
        return listenerFactory;
    }

    @Bean("controlCommandKafkaTemplate")
    public KafkaTemplate<String, String> kafkaTemplate(@Qualifier("controlCommandKafkaProperties") KafkaProperties inProps) {
        return new KafkaTemplate(new DefaultKafkaProducerFactory(inProps.buildProducerProperties()));
    }

    private void configureListenerFactory(ConcurrentKafkaListenerContainerFactory<Object, Object> factory, KafkaProperties inProps) {
        PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
        KafkaProperties.Listener properties = inProps.getListener();
        // 設(shè)置監(jiān)聽線程并發(fā)數(shù) control.command.kafka.listener.concurrency
        map.from(properties::getConcurrency).to(factory::setConcurrency);
        // control.kafka.listener.type=batch 時(shí) 批量監(jiān)聽
        if (properties.getType().equals(KafkaProperties.Listener.Type.BATCH)) {
            factory.setBatchListener(true);
        }
    }

    private void configureContainer(ContainerProperties container, KafkaProperties inProps) {
        PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
        KafkaProperties.Listener properties = inProps.getListener();
        map.from(properties::getAckMode).to(container::setAckMode);
        map.from(properties::getClientId).to(container::setClientId);
        map.from(properties::getAckCount).to(container::setAckCount);
        map.from(properties::getAckTime).as(Duration::toMillis).to(container::setAckTime);
        map.from(properties::getPollTimeout).as(Duration::toMillis).to(container::setPollTimeout);
        map.from(properties::getNoPollThreshold).to(container::setNoPollThreshold);
        map.from(properties::getIdleEventInterval).as(Duration::toMillis).to(container::setIdleEventInterval);
        map.from(properties::getMonitorInterval).as(Duration::getSeconds).as(Number::intValue)
                .to(container::setMonitorInterval);
        map.from(properties::getLogContainerConfig).to(container::setLogContainerConfig);
        map.from(properties::isMissingTopicsFatal).to(container::setMissingTopicsFatal);
    }
}

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;

@SpringBootApplication
@EnableKafka
public class MyKafkaApplication {
    public static void main(String[] args) {
        SpringApplication.run(MyKafkaApplication.class, args);
    }
}

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MyConfiguration {

    @Bean
    @ConditionalOnExpression("${myapp.feature.enabled:true}")
    public MyBean myBean() {
        // 返回需要?jiǎng)?chuàng)建的 Bean 實(shí)例
        return new MyBean();
    }
}

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

@KafkaListener(topics = "my_topic", groupId = "my_group", ackMode = "MANUAL")
public void processMessage(String message, Acknowledgment acknowledgment) {
    // 處理消息的邏輯
    // 手動(dòng)確認(rèn)消息
    acknowledgment.acknowledge();
}

@Component
@RefreshScope
@ConditionalOnExpression("${control.command.kafka.enabled:false}")
@DependsOn(value = {"controlCommandKafkaTemplate"})
public class ControlCommandKafkaProducer extends BaseLogable {

    @Resource(name = "controlCommandKafkaTemplate")
    private KafkaTemplate<String, String> controlCommandKafkaTemplate;

    public void send(String key, String topic, String msg) {
        bizLogger.info("sent kafka msg, topic:{}, key: {}, msg: {}", topic, key, msg);
        ListenableFuture<SendResult<String, String>> listenableFuture = controlCommandKafkaTemplate.send(topic, key, msg);
        listenableFuture.addCallback(result -> bizLogger.info("sent msg success：{}", msg),
                e -> {
                    bizLogger.error("sent msg failure, msg: {}", msg, e);
                });
    }

}

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService {
    
    private final KafkaTemplate<String, String> kafkaTemplate;
    
    public KafkaProducerService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
    
    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
}

public void sendMessageToPartition(String topic, int partition, String message) {
    kafkaTemplate.send(topic, partition, null, message);
}

public void sendMessageWithKey(String topic, String key, String message) {
    kafkaTemplate.send(topic, key, message);
}

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.support.SendResult;
import org.springframework.kafka.support.future.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

public void sendMessageAndWaitForConfirmation(String topic, String message) {
    ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);
    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(record);
    
    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
        @Override
        public void onSuccess(SendResult<String, String> result) {
            // 消息發(fā)送成功的處理邏輯
        }

        @Override
        public void onFailure(Throwable ex) {
            // 消息發(fā)送失敗的處理邏輯
        }
    });
}

@Component
@RefreshScope
@ConditionalOnExpression("${control.command.kafka.enabled:false}")
@DependsOn(value = {"controlCommandBatchFactory"})
public class ControlCommandKafkaConsumer extends BaseLogable {


    /**
     * 監(jiān)聽主題，單條消費(fèi)
     */
    @KafkaListener(id = "${control.command.kafka.consumer.groupId}", topics = "${kafka.command.topic.ecgs-command-up}", containerFactory = "controlCommandBatchFactory")
    public void listen(ConsumerRecord<String, String> record, Acknowledgment ack) {
        bizLogger.info("kafka msg::: " + record.value() + ", key: " + record.key());
        //接收Kafka消息
        String data = record.value();

        //接收消息確認(rèn)
        ack.acknowledge();
    }

}

package cc.eslink.yq.iwater.kafka;

import cc.eslink.common.base.BaseLogable;
import cc.eslink.yq.iwater.dto.schedule.AlarmInfoDTO;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.annotation.DependsOn;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;


@Component
@RefreshScope
@ConditionalOnExpression("${control.command.kafka.enabled:false}")
@DependsOn(value = {"controlCommandBatchFactory"})
public class ControlCommandKafkaConsumer extends BaseLogable {


    @KafkaListener(id = "${control.command.kafka.consumer.groupId}", topics = "${kafka.command.topic.ecgs-command-up}", containerFactory = "controlCommandBatchFactory")
    public void listen(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        bizLogger.info("== 接收到 kafka msg 條數(shù):::{} ==", records.size());
        //接收消息確認(rèn)
        ack.acknowledge();
        //接收Kafka消息
        for (ConsumerRecord<String, String> record : records) {
            try {
                final AlarmInfoDTO alarmInfoDTO = JSON.parseObject(record.value(), new TypeReference<AlarmInfoDTO>() {});
            } catch (Exception e) {
                expLogger.error("alarm.dispose error", record, e);
            }
        }


    }

}