public class KonkaKafkaListener {

    private final static Logger LOGGER = LoggerFactory.getLogger(KonkaKafkaListener.class);


    @Autowired
    private RouterService routerService;


    @KafkaListener(containerFactory = "kafkaListenerContainerFactory", topics = "test")
    public void consumerListener(List<ConsumerRecord> consumerRecords, Acknowledgment ack) {
        ack.acknowledge();//直接提交offset
        if (consumerRecords.size() > 0) {
            PartitionCounter.addCounter(consumerRecords.get(0).partition(), consumerRecords.size());
        }
        Iterator<ConsumerRecord> iterator = consumerRecords.iterator();
        while (iterator.hasNext()) {
            ConsumerRecord consumerRecord = iterator.next();
            String key = consumerRecord.key().toString();
            KafkaLogMessage kafkaLogMessage = (KafkaLogMessage) consumerRecord.value();
            if (kafkaLogMessage == null) {
                continue;
            }
            routerService.handleKafkaMessage(key, kafkaLogMessage);
        }
    }

#消费者并发启动个数（对应分区个数）每个listener方法
kafka.concurrency=10  

将启动器的并发提高到和分区数一致

kafka 消费能力的提高

1、自动提交的实现

2、autoCommitIntervalMs 设置每次隔多久自动提交offset

3、kafka.max.poll.interval.ms 和 sessionTimeout

max.poll.interval.ms ，它表示最大的poll数据间隔，如果超过这个间隔没有发起pool请求，但heartbeat仍旧在发，就认为该consumer处于 livelock状态。就会将该consumer退出consumer group

之后就会触发导致reblance

·heartbeat.interval.ms

心跳间隔。心跳是在consumer与coordinator之间进行的。心跳是确定consumer存活，加入或者退出group的有效手段。

这个值必须设置的小于session.timeout.ms，因为：

当Consumer由于某种原因不能发Heartbeat到coordinator时，并且时间超过session.timeout.ms时，就会认为该consumer已退出，它所订阅的partition会分配到同一group 内的其它的consumer上。

通常设置的值要低于session.timeout.ms的1/3。

默认值是：3000 （3s）

·session.timeout.ms

Consumer session 过期时间。这个值必须设置在broker configuration中的group.min.session.timeout.ms 与 group.max.session.timeout.ms之间。

其默认值是：10000 （10 s）