關於SpringKafka消費者的幾個監聽器:[一次處理單條消息和一次處理一批消息]以及[自動提交offset和手動提交offset]

自己在使用Spring Kafka 的消費者消費消息的時候的實踐總結：

接口 KafkaDataListener 是spring-kafka提供的一個供消費者接受消息的頂層接口,也是一個空接口;
public interface KafkaDataListener<T> {}

對於消費端接收消息的時候,spring-kafka的設計思路是,提供一個頂層接口,提供兩個子類,一個子類是自動提交offset的,另一個子類是手動提交offset的.
無論是自動提交offset還是手動提交offset,又各分為兩種,一種是一次只處理一條消息,另一種是一次可以處理一批消息.

該 KafkaDataListener 頂層接口有兩個實現類:GenericMessageListener 和 GenericAcknowledgingMessageListener,
二者的區別是，前者是自動提交offset，后者是手動提交offset。

1、 GenericMessageListener 
    該接口是自動提交offset,它的onMessage方法的參數只有一個,就是傳遞過來的一條消息;
    public interface GenericMessageListener<T> extends KafkaDataListener<T> {void onMessage(T data);}
    這個接口又有兩個子接口:MessageListener 和 BatchMessageListener
    這兩個接口也都是空接口,二者的區別是,前者一次只處理一條消息,后者一次處理一批消息.
    
    //一次處理一條消息
    //消費者如果實現該接口的話,如果配置中設置max.poll.records參數大於1的話是無效的,因為它一次只處理一條消息
    public interface MessageListener<K, V> extends GenericMessageListener<ConsumerRecord<K, V>> {}
    //一次可以處理一批消息,每一批次的消息總條數是隨機的,但可以在消費者的配置中設置一個最大值（max.poll.records,
    //比如設置了最大拉取的消息條數為100,那么onMessage方法每次接受到的消息條數是隨機的,但最大不會超過100）
    public interface BatchMessageListener<K, V> extends GenericMessageListener<List<ConsumerRecord<K, V>>> {}
    
2、 GenericAcknowledgingMessageListener
    該接口是手動提交offset,它的onMessage方法的參數有兩個,第一個是傳遞過來的一條消息,第二個參數是用於提交offset的對象
    public interface GenericAcknowledgingMessageListener<T> extends KafkaDataListener<T> {void onMessage(T data, Acknowledgment acknowledgment);}
    
    這個接口也有兩個子接口:AcknowledgingMessageListener 和 BatchAcknowledgingMessageListener,這兩個接口也都是空接口.
    //一次只處理一條消息,並手動提交offset,需要在消費者的配置中設置<property name="ackMode" value="MANUAL_IMMEDIATE"/>
    public interface AcknowledgingMessageListener<K, V> extends GenericAcknowledgingMessageListener<ConsumerRecord<K, V>> {}
    //一次處理一批消息,處理完這一批消息之后，在批量提交offset,需要在消費者的配置中設置<property name="ackMode" value="MANUAL"/>
    public interface BatchAcknowledgingMessageListener<K, V> extends GenericAcknowledgingMessageListener<List<ConsumerRecord<K, V>>> {}

 

下面的消費者繼承的是MessageListener這個監聽器，就是一次處理一條消息，而且是自動提交offset：

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.xxxxxx.consumer.dto.FriendRelationDto;
import com.xxxxxx.consumer.dto.MessageDto;
import com.xxxxxx.consumer.service.FriendRelationService;
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.MessageListener;
import org.springframework.stereotype.Service;
import java.io.IOException;

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

    private static final Logger logger = LoggerFactory.getLogger(ConsumerService.class);
    @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();
        }
    }
}

 

下面的消費者實現的BatchMessageListener這個監聽器，就是一次接受一批消息,消息的數量是隨機的，但最大不會超過"max.poll.records"參數配置的數量：

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.xxxxxx.consumer.dto.FriendRelationDto;
import com.xxxxxx.consumer.dto.MessageDto;
import com.xxxxxx.consumer.service.FriendRelationService;
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.BatchMessageListener;
import org.springframework.stereotype.Service;

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

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

    private static final Logger logger = LoggerFactory.getLogger(ConsumerService.class);
    @Autowired
    private FriendRelationService friendRelationService;

    @Override
    public void onMessage(List<ConsumerRecord<Integer, String>> recordList) {
        for (ConsumerRecord<Integer, String> record : recordList) {
            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();
            }
        }

    }
}

下面的消費者實現的是AcknowledgingMessageListener這個監聽器，它的特點是一次接收一條消息，可以通過acknowledgment來手動提交offset，需要在消費者的配置中指定<property name="ackMode" value="MANUAL_IMMEDIATE"/>：

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.xxxxxx.consumer.dto.FriendRelationDto;
import com.xxxxxx.consumer.dto.MessageDto;
import com.xxxxxx.consumer.service.FriendRelationService;
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;

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

    private static final Logger logger = LoggerFactory.getLogger(ConsumerService.class);
    @Autowired
    private FriendRelationService friendRelationService;

    @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());
            logger.info("===========開始提交offset=============");
            acknowledgment.acknowledge();//提交offset
            logger.info("===========已經提交offset=============");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

下面的消費者實現的是BatchAcknowledgingMessageListener這個監聽器，它的特點是一次可以處理一批消息，並且可以在處理完這一批消息之后提交offset，需要在消費者的配置文件中配置"max.poll.records"參數指定本批消息可以達到的最大值，並指定<property name="ackMode" value="MANUAL"/>：

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.xxxxxx.consumer.dto.FriendRelationDto;
import com.xxxxxx.consumer.dto.MessageDto;
import com.xxxxxx.consumer.service.FriendRelationService;
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.BatchAcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

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

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

    private static final Logger logger = LoggerFactory.getLogger(ConsumerService.class);
    @Autowired
    private FriendRelationService friendRelationService;
    

    @Override
    public void onMessage(List<ConsumerRecord<Integer, String>> recordList, Acknowledgment acknowledgment) {
        logger.info("Before receiving:" + recordList.toString());
        logger.info("本次消息總數:" + recordList.size());
        for (ConsumerRecord<Integer, String> record : recordList) {
            String value = record.value();
            MessageDto<FriendRelationDto> message = JSON.parseObject(value, new TypeReference<MessageDto<FriendRelationDto>>() {
            });
            try {
                friendRelationService.process(message.getData());
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        logger.info("===========開始提交offset=============");
        acknowledgment.acknowledge();//提交offset
        logger.info("===========已經提交offset=============");
    }
}

 

下面是spring-kafka消費端的配置文件示例：

<?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"/>
                <!-- 當使用批量處理消息的時候,每次onMessage方法獲取到的消息總條數雖然是隨機的,但是不會超過此最大值 -->
                <entry key="max.poll.records" value="50"/>
                <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.xxxxxxx.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"/>
        <!-- 提交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>-->

    <!-- 多線程消息監聽容器,每啟動一個消費者客戶端,可以開啟多個線程,開啟多少個線程自己可以通過concurrency來指定 -->
    <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>