spring-kafka手动提交offset

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
         http://www.springframework.org/schema/beans/spring-beans.xsd">


    <bean id="consumerProperties" class="java.util.HashMap">
        <constructor-arg>
            <map>
                <entry key="bootstrap.servers" value="${bootstrap.servers}"/>
                <!-- 指定消费组名 -->
                <entry key="group.id" value="friend-group"/>
                <entry key="enable.auto.commit" value="false"/>
                <entry key="auto.commit.interval.ms" value="1000"/>
                <entry key="session.timeout.ms" value="15000"/>
                <entry key="max.poll.records" value="1"/>
                <entry key="key.deserializer" value="org.apache.kafka.common.serialization.IntegerDeserializer"/>
                <!--<entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>-->
                <entry key="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
            </map>
        </constructor-arg>
    </bean>

    <bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
        <constructor-arg>
            <ref bean="consumerProperties"/>
        </constructor-arg>
    </bean>

    <!-- 消费消息的服务类 -->
    <bean id="messageListernerConsumerService" class="com.zhaopin.consumer.ConsumerService"/>

    <!-- 消费者容器配置信息 -->
    <bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties">
        <constructor-arg value="friend"/>
        <!--<constructor-arg>
            <list>
                <value>zptopic</value>
                <value>ssmk</value>
                <value>friend</value>
            </list>
        </constructor-arg>-->
        <property name="messageListener" ref="messageListernerConsumerService"/>

        <!-- 设置如何提交offset -->
        <property name="ackMode" value="MANUAL_IMMEDIATE"/>
    </bean>

    <!-- 单线程消息监听容器 -->
    <!--<bean id="messageListenerContainer" class="org.springframework.kafka.listener.KafkaMessageListenerContainer" init-method="doStart">
        <constructor-arg ref="consumerFactory"/>
        <constructor-arg ref="containerProperties"/>
    </bean>-->

    <!-- 多线程消息监听容器 -->
    <bean id="messageListenerContainer" class="org.springframework.kafka.listener.ConcurrentMessageListenerContainer" init-method="doStart">
        <constructor-arg ref="consumerFactory"/>
        <constructor-arg ref="containerProperties"/>
        <property name="concurrency" value="5"/>
    </bean>

</beans>

消费者监听类实现AcknowledgingMessageListener这个监听器，可以实现手动提交offset：

package com.zhaopin.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.zhaopin.consumer.dto.FriendRelationDto;
import com.zhaopin.consumer.dto.MessageDto;
import com.zhaopin.consumer.service.FriendRelationService;
import com.zhaopin.pojo.TbPerson;
import com.zhaopin.service.PersonService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by SYJ on 2017/3/21.
 */
@Service
public class ConsumerService implements AcknowledgingMessageListener<Integer, String> {

    private static final Logger logger = LoggerFactory.getLogger(ConsumerService.class);
    private static List<TbPerson> personList = new ArrayList<TbPerson>();
    private static final Integer INSERT_BATCH_COUNT = 50;

    @Autowired
    private PersonService personService;

    @Autowired
    private FriendRelationService friendRelationService;

    /**
     * 消息监听方法
     * @param record
     */
    /*@Override
    public void onMessage(ConsumerRecord<Integer, String> record) {
        logger.info("Before receiving:" + record.toString());
        String value = record.value();
        MessageDto<FriendRelationDto> message = JSON.parseObject(value, new TypeReference<MessageDto<FriendRelationDto>>(){});
        try {
            friendRelationService.process(message.getData());
        } catch (IOException e) {
            e.printStackTrace();
        }
        //insert(record);
        //insertBatch(record);
    }*/

    /**
     * 单个TbPerson入库
     * @param record
     */
    public void insert(ConsumerRecord<Integer, String> record){
        String value = record.value();
        TbPerson person = JSON.parseObject(value, TbPerson.class);
        personService.insert(person);
        System.out.println("Single data writing to the database:" + record);
    }

    /**
     * 批量TbPerson入库
     * @param record
     */
    public void insertBatch(ConsumerRecord<Integer, String> record){
        String value = record.value();
        TbPerson person = JSON.parseObject(value, TbPerson.class);
        personList.add(person);
        if (personList.size() == INSERT_BATCH_COUNT) {
            personService.insertBatch(personList);
            System.out.println("Batch data writing to the database:" + personList);
            personList.clear();
        }
    }

    @Override
    public void onMessage(ConsumerRecord<Integer, String> record, Acknowledgment acknowledgment) {
        logger.info("Before receiving:" + record.toString());
        String value = record.value();
        MessageDto<FriendRelationDto> message = JSON.parseObject(value, new TypeReference<MessageDto<FriendRelationDto>>(){});
        try {
            friendRelationService.process(message.getData());
            acknowledgment.acknowledge();//提交offset
        } catch (IOException e) {
            e.printStackTrace();
        }
        //insert(record);
        //insertBatch(record);
    }
}