./kafka-topics.sh --create --zookeeper 192.168.25.128:2181 --replication-factor 1 --partitions 2 --topic testTopic2

kafka.test.topic1=testTopic1
kafka.test.topic2=testTopic2
kafka.test.topic3=testTopic3
kafka.broker=192.168.25.128:9092
auto.commit.interval.time=60000
#kafka.test.group=customer-test
kafka.test.group1=testGroup1
kafka.test.group2=testGroup2
kafka.test.group3=testGroup3
kafka.offset=earliest
kafka.auto.commit=false

session.timeout.time=10000
kafka.concurrency=2

package com.yin.customer.config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;

/**
 * @author yin
 * @Date 2019/11/24 15:54
 * @Method
 */
@Configuration
@Component
public class KafkaConfig {
    @Value("${kafka.broker}")
    private String broker;
    @Value("${kafka.auto.commit}")
    private String autoCommit;

   // @Value("${kafka.test.group}")
    //private String testGroup;

    @Value("${session.timeout.time}")
    private String sessionOutTime;

    @Value("${auto.commit.interval.time}")
    private String autoCommitTime;

    @Value("${kafka.offset}")
    private String offset;
    @Value("${kafka.concurrency}")
    private Integer concurrency;

   @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory(){
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        //監(jiān)聽(tīng)設(shè)置兩個(gè)個(gè)分區(qū)
        factory.setConcurrency(concurrency);
        //打開(kāi)批量拉取數(shù)據(jù)
        factory.setBatchListener(true);
        //這里設(shè)置的是心跳時(shí)間也是拉的時(shí)間，也就說(shuō)每間隔max.poll.interval.ms我們就調(diào)用一次poll,kafka默認(rèn)是300s,心跳只能在poll的時(shí)候發(fā)出，如果連續(xù)兩次poll的時(shí)候超過(guò)
        //max.poll.interval.ms 值就會(huì)導(dǎo)致rebalance
        //心跳導(dǎo)致GroupCoordinator以為本地consumer節(jié)點(diǎn)掛掉了，引發(fā)了partition在consumerGroup里的rebalance。
        // 當(dāng)rebalance后，之前該consumer擁有的分區(qū)和offset信息就失效了，同時(shí)導(dǎo)致不斷的報(bào)auto offset commit failed。
        factory.getContainerProperties().setPollTimeout(3000);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }

    private ConsumerFactory<String,String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<String, String>(consumerConfigs());
    }

   @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> propsMap = new HashMap<>();
        //kafka的地址
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, broker);
        //是否自動(dòng)提交 Offset
        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, autoCommit);
        // enable.auto.commit 設(shè)置成 false，那么 auto.commit.interval.ms 也就不被再考慮
        //默認(rèn)5秒鐘，一個(gè) Consumer 將會(huì)提交它的 Offset 給 Kafka
        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,  5000);

        //這個(gè)值必須設(shè)置在broker configuration中的group.min.session.timeout.ms 與 group.max.session.timeout.ms之間。
        //zookeeper.session.timeout.ms 默認(rèn)值：6000
        //ZooKeeper的session的超時(shí)時(shí)間，如果在這段時(shí)間內(nèi)沒(méi)有收到ZK的心跳，則會(huì)被認(rèn)為該Kafka server掛掉了。
        // 如果把這個(gè)值設(shè)置得過(guò)低可能被誤認(rèn)為掛掉，如果設(shè)置得過(guò)高，如果真的掛了，則需要很長(zhǎng)時(shí)間才能被server得知。
        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionOutTime);
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        //組與組間的消費(fèi)者是沒(méi)有關(guān)系的。
        //topic中已有分組消費(fèi)數(shù)據(jù)，新建其他分組ID的消費(fèi)者時(shí)，之前分組提交的offset對(duì)新建的分組消費(fèi)不起作用。
        //propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, testGroup);

        //當(dāng)創(chuàng)建一個(gè)新分組的消費(fèi)者時(shí)，auto.offset.reset值為latest時(shí)，
        // 表示消費(fèi)新的數(shù)據(jù)（從consumer創(chuàng)建開(kāi)始，后生產(chǎn)的數(shù)據(jù)），之前產(chǎn)生的數(shù)據(jù)不消費(fèi)。
        // https://blog.csdn.net/u012129558/article/details/80427016

        //earliest 當(dāng)分區(qū)下有已提交的offset時(shí)，從提交的offset開(kāi)始消費(fèi)；無(wú)提交的offset時(shí)，從頭開(kāi)始消費(fèi)。
       // latest 當(dāng)分區(qū)下有已提交的offset時(shí)，從提交的offset開(kāi)始消費(fèi)；無(wú)提交的offset時(shí)，消費(fèi)新產(chǎn)生的該分區(qū)下的數(shù)據(jù)。

        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, offset);
        //不是指每次都拉50條數(shù)據(jù)，而是一次最多拉50條數(shù)據(jù)()
        propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 5);
        return propsMap;
    }
}

@Component
public class KafkaConsumer {
    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);

    @KafkaListener(topics = "${kafka.test.topic1}",groupId = "${kafka.test.group1}",containerFactory = "kafkaListenerContainerFactory")
    public void listenPartition1(List<ConsumerRecord<?, ?>> records,Acknowledgment ack) {
        logger.info("testTopic1 recevice a message size :{}" , records.size());

        try {
            for (ConsumerRecord<?, ?> record : records) {
                Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                logger.info("received:{} " , record);
                if (kafkaMessage.isPresent()) {
                    Object message = record.value();
                    String topic = record.topic();
                    Thread.sleep(300);
                    logger.info("p1 topic is:{} received message={}",topic, message);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            ack.acknowledge();
        }
    }

    @KafkaListener(topics = "${kafka.test.topic2}",groupId = "${kafka.test.group2}",containerFactory = "kafkaListenerContainerFactory")
    public void listenPartition2(List<ConsumerRecord<?, ?>> records,Acknowledgment ack) {
        logger.info("testTopic2 recevice a message size :{}" , records.size());

        try {
            for (ConsumerRecord<?, ?> record : records) {
                Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                logger.info("received:{} " , record);
                if (kafkaMessage.isPresent()) {
                    Object message = record.value();
                    String topic = record.topic();
                    Thread.sleep(300);
                    logger.info("p2 topic :{},received message={}",topic, message);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            ack.acknowledge();
        }
    }

    @KafkaListener(topics = "${kafka.test.topic3}",groupId = "${kafka.test.group3}",containerFactory = "kafkaListenerContainerFactory")
    public void listenPartition3(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        logger.info("testTopic3 recevice a message size :{}" , records.size());

        try {
            for (ConsumerRecord<?, ?> record : records) {
                Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                logger.info("received:{} " , record);
                if (kafkaMessage.isPresent()) {
                    Object message = record.value();
                    String topic = record.topic();
                    logger.info("p3 topic :{},received message={}",topic, message);
                    Thread.sleep(300);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            ack.acknowledge();
        }
    }
}