java本身手動控制kafka的offset

以前使用kafka的KafkaStream：http://blog.csdn.net/qq_20641565/article/details/60810174，讓每一個消費者和對應的patition創建對應的流來讀取kafka上面的數據，若是comsumer獲得數據，那麼kafka就會自動去維護該comsumer的offset，例如在獲取到kafka的消息後正準備入庫（未入庫），可是消費者掛了，那麼若是讓kafka自動去維護offset，它就會認爲這條數據已經被消費了，那麼會形成數據丟失。

可是kafka可讓你本身去手動提交，若是在上面的場景中，那麼須要咱們手動commit，若是comsumer掛了 那麼程序就不會執行commit這樣的話 其餘同group的消費者又能夠消費這條數據，保證數據不丟，先要作以下設置：

//設置不自動提交，本身手動更新offset
properties.put("enable.auto.commit", "false");

•  

使用以下api提交：

consumer.commitSync();

•  

注意：

剛作了個測試，若是我從kafka中取出5條數據，分別爲1,2,3,4,5，若是消費者在執行一些邏輯在執行1,2,3，4的時候都失敗了未提交commit，而後消費5作邏輯成功了提交了commit，那麼offset也會被移動到5那一條數據那裏，1,2,3,4 至關於也會丟失

若是是作消費者取出數據執行一些操做，所有都失敗的話，而後重啓消費者，這些數據會從失敗的時候從新開始讀取

因此消費者仍是應該本身作容錯機制

測試項目結構以下:

其中ConsumerThreadNew類：

package com.lijie.kafka;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 *                       
 * @Filename ConsumerThreadNew.java
 *
 * @Description 
 *
 * @Version 1.0
 *
 * @Author Lijie
 *
 * @Email lijiewj39069@touna.cn
 *       
 * @History
 *<li>Author: Lijie</li>
 *<li>Date: 2017年3月21日</li>
 *<li>Version: 1.0</li>
 *<li>Content: create</li>
 *
 */
public class ConsumerThreadNew implements Runnable {
    private static Logger                   LOG = LoggerFactory.getLogger(ConsumerThreadNew.class);

    //KafkaConsumer kafka生產者
    private KafkaConsumer<String, String>   consumer;

    //消費者名字
    private String                          name;

    //消費的topic組
    private List<String>                    topics;

    //構造函數
    public ConsumerThreadNew(KafkaConsumer<String, String> consumer, String topic, String name) {
        super();
        this.consumer = consumer;
        this.name = name;
        this.topics = Arrays.asList(topic);
    }

    @Override
    public void run() {
        consumer.subscribe(topics);
        List<ConsumerRecord<String, String>> buffer = new ArrayList<>();

        // 批量提交數量
        final int minBatchSize = 1; 
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                LOG.info("消費者的名字爲:" + name + ",消費的消息爲：" + record.value());
                buffer.add(record);
            }
            if (buffer.size() >= minBatchSize) {
                //這裏就是處理成功了而後本身手動提交
                consumer.commitSync();
                LOG.info("提交完畢");
                buffer.clear();
            }
        }
    }

}

•  

MyConsume類以下：

package com.lijie.kafka;

import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 *                       
 * @Filename MyConsume.java
 *
 * @Description 
 *
 * @Version 1.0
 *
 * @Author Lijie
 *
 * @Email lijiewj39069@touna.cn
 *       
 * @History
 *<li>Author: Lijie</li>
 *<li>Date: 2017年3月21日</li>
 *<li>Version: 1.0</li>
 *<li>Content: create</li>
 *
 */
public class MyConsume {
    private static Logger   LOG = LoggerFactory.getLogger(MyConsume.class);

    public MyConsume() {
        // TODO Auto-generated constructor stub
    }

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "10.0.4.141:19093,10.0.4.142:19093,10.0.4.143:19093");
        //設置不自動提交，本身手動更新offset
        properties.put("enable.auto.commit", "false");
        properties.put("auto.offset.reset", "latest");
        properties.put("zookeeper.connect", "10.0.4.141:2181,10.0.4.142:2181,10.0.4.143:2181");
        properties.put("session.timeout.ms", "30000");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("group.id", "lijieGroup");
        properties.put("zookeeper.connect", "192.168.80.123:2181");
        properties.put("auto.commit.interval.ms", "1000");
        ExecutorService executor = Executors.newFixedThreadPool(5);

        //執行消費
        for (int i = 0; i < 7; i++) {
            executor.execute(new ConsumerThreadNew(new KafkaConsumer<String, String>(properties),
                "lijietest", "消費者" + (i + 1)));
        }
    }
}

•  

MyProducer類以下：

package com.lijie.kafka;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 * 
 *                       
 * @Filename MyProducer.java
 *
 * @Description 
 *
 * @Version 1.0
 *
 * @Author Lijie
 *
 * @Email lijiewj39069@touna.cn
 *       
 * @History
 *<li>Author: Lijie</li>
 *<li>Date: 2017年3月21日</li>
 *<li>Version: 1.0</li>
 *<li>Content: create</li>
 *
 */
public class MyProducer {

    private static Properties                       properties;
    private static KafkaProducer<String, String>    pro;
    static {
        //配置
        properties = new Properties();
        properties.put("bootstrap.servers", "10.0.4.141:19093,10.0.4.142:19093,10.0.4.143:19093");

        //序列化類型
        properties
            .put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        //建立生產者
        pro = new KafkaProducer<>(properties);
    }

    public static void main(String[] args) throws Exception {
        produce("lijietest");
    }

    public static void produce(String topic) throws Exception {

        //模擬message
        //          String value = UUID.randomUUID().toString();
        for (int i = 0; i < 10000; i++) {
            //封裝message
            ProducerRecord<String, String> pr = new ProducerRecord<String, String>(topic, i + "");

            //發送消息
            pro.send(pr);
            Thread.sleep(1000);
        }

    }
}

•  

pom文件以下：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>lijie-kafka-offset</groupId>
    <artifactId>lijie-kafka-offset</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.11</artifactId>
            <version>0.10.1.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.2.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>2.2.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.2.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client</artifactId>
            <version>1.0.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-server</artifactId>
            <version>1.0.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>2.2.0</version>
        </dependency>
        <dependency>
            <groupId>jdk.tools</groupId>
            <artifactId>jdk.tools</artifactId>
            <version>1.7</version>
            <scope>system</scope>
            <systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>
        </dependency>
        <dependency>
            <groupId>org.apache.httpcomponents</groupId>
            <artifactId>httpclient</artifactId>
            <version>4.3.6</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>1.7</source>
                    <target>1.7</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>