@Override

    @KafkaListener(id = ConsumerConst.LISTENER_ID_WARNING, topics = {"${kafka.app.topic.warning}"}, containerFactory = "ackContainerFactory", groupId = "warning")

    public void ackListener(ConsumerRecord record, Acknowledgment ack) {

        if (LOG.isInfoEnabled()) {

            LOG.info("###################预警ackListener接收到消息###################");

        }

　　　　　// todo 消费业务处理

    }

    @Bean("ackContainerFactory")

    public ConcurrentKafkaListenerContainerFactory ackContainerFactory(ConsumerFactory consumerFactory) {

        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();

        factory.setConsumerFactory(consumerFactory);

        //容器线程数：小于或等于Topic的分区数

        factory.setConcurrency(3);

        //设置提交偏移量的方式

        factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE);

        // 禁止自动启动

        factory.setAutoStartup(false);

        return factory;

    }