Kafka學習筆記之掃盲

時間 2019-11-07

標籤 kafka 學習筆記掃盲欄目 Kafka 简体版

原文原文鏈接

本文使用的Kafka版本0.11html

先思考些問題:java

我想分析一下用戶行爲（pageviews），以便我能設計出更好的廣告位git
我想對用戶的搜索關鍵詞進行統計，分析出當前的流行趨勢。這個頗有意思，在經濟學上有個長裙理論，就是說，若是長裙的銷量高了，說明經濟不景氣了，由於姑娘們沒錢買各類絲襪了。github
有些數據，我以爲存數據庫浪費，直接存硬盤又怕到時候操做效率低。數據庫

這個時候，咱們就能夠用到分佈式消息系統了。雖然上面的描述更偏向於一個日誌系統，但確實kafka在實際應用中被大量的用於日誌系統。
這些場景都有一個共同點：數據是由上游模塊產生，上游模塊，使用上游模塊的數據計算、統計、分析，這個時候就可使用消息系統，尤爲是分佈式消息系統！ apache

Kafka是一個分佈式消息系統，由linkedin使用scala編寫. Kafka的動態擴容是經過Zookeeper來實現的。
Zookeeper是一種在分佈式系統中被普遍用來做爲：分佈式狀態管理、分佈式協調管理、分佈式配置管理、和分佈式鎖服務的集羣。kafka增長和減小服務器都會在Zookeeper節點上觸發相應的事件。編程

編程

APIDOC:http://kafka.apache.org/0110/...
官方github例子： https://github.com/apache/kaf...

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.11.0.0</version>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>0.11.0.0</version>
</dependency>

首先貼一下官方例子：

Producer:

public class MyKafkaProducer {

    public static void main(String[] args) {
        /**
         * 這個例子中，每次調用都會建立一個Producer實例，但此處只是爲了演示方便，實際使用中，請將Producer做爲單例使用，它是線程安全的。

         * 從Kafka 0.11 開始，KafkaProducer支持兩種額外的模式：冪等(idempotent)與事務(transactional)。冪等使得以前的at least once變成exactly once傳送
         * 冪等Producer的重試再也不會致使重複消息。事務容許應用程序以原子方式將消息發送到多個分區（和主題！）

         * 開啓idempotence冪等:props.put("enable.idempotence", true);設置以後retries屬性自動被設爲Integer.MAX_VALUE;;acks屬性自動設爲all;;max.inflight.requests.per.connection屬性自動設爲1.其他同樣。

         * 開啓事務性： props.put("transactional.id", "my-transactional-id");一旦這個屬性被設置，那麼冪等也會自動開啓。而後使用事務API操做便可
         */
    }
    private static void send(){
        Properties props = new Properties();
         props.put("bootstrap.servers", "localhost:9092");
         props.put("enable.idempotence", true);//開啓idempotence冪等 extract-once
//         props.put("acks", "all");//acks配置控制請求被認爲完成的條件
//         props.put("retries", 0);重試次數
//         props.put("batch.size", 16384);
//         props.put("linger.ms", 1);
//         props.put("buffer.memory", 33554432);
         props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
         props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

         Producer<String, String> producer = new KafkaProducer<>(props);
         for (int i = 0; i < 100; i++)
             producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));

         producer.close();
    }
    private static void sendInTx(){
         Properties props = new Properties();
         props.put("bootstrap.servers", "localhost:9092");
         props.put("transactional.id", "my-transactional-id");//要啓用事務，必須配置一個惟一的事務id

         /**
          * http://kafka.apache.org/0110/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html
          * KafkaProducer類是線程安全的，能夠在多線程之間共享。
          */
         Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());

         producer.initTransactions();

         try {
             producer.beginTransaction();
             for (int i = 0; i < 100; i++){
                 // send()是異步的，會當即返回，內部是緩存到producer的buffer中，以便於生產者能夠批量提交， 你也能夠傳遞一個回調send(ProducerRecord<K,V> record, Callback callback)
                 producer.send(new ProducerRecord<>("my-topic", Integer.toString(i), Integer.toString(i)));
             }
             producer.commitTransaction();
         } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
             //沒法恢復的異常，咱們只能關閉producer 
             producer.close();
         } catch (KafkaException e) {
             // 可恢復的異常，終止事務而後重試便可。
             producer.abortTransaction();
         }
         producer.close();
    }
}

發送完以後，咱們能夠用bin目錄下的kafka-console-consumer來看發送的結果（固然如今用的topic是test）。能夠用命令：

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

Consumer:

/**
 *與producer不一樣，Kafka consumer不是線程安全的。
 */
public class MyKafkaConsumer {
    /**
     * 經過配置enable.auto.commit,auto.commit.interval.ms來按期自動提交消費的偏移量
     */
    private  void recieveByAutoCommitOffset(){
        Properties props = new Properties();
         props.put("bootstrap.servers", "localhost:9092");
         props.put("group.id", "test");
         props.put("enable.auto.commit", "true");
         props.put("auto.commit.interval.ms", "1000");
         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
         consumer.subscribe(Arrays.asList("foo", "bar"));
         while (true) {
             ConsumerRecords<String, String> records = consumer.poll(100);
             for (ConsumerRecord<String, String> record : records)
                 System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
         }
//         consumer.wakeup();
    }
    /**
     * 手動提交消費的偏移量,這樣用戶能夠控制記錄什麼時候被視爲已消費，從而提交其偏移量。 當消息的消耗與一些處理邏輯相結合時，這是有用的，由於在完成處理以前不該將消息視爲已消費。
     */
    private void recieveByManualCommitOffset(){
        Properties props = new Properties();
         props.put("bootstrap.servers", "localhost:9092");
         props.put("group.id", "test");
         props.put("enable.auto.commit", "false");//手動提交offset
         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
         consumer.subscribe(Arrays.asList("foo", "bar"));
         final int minBatchSize = 200;
         List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
         while (true) {
             ConsumerRecords<String, String> records = consumer.poll(100);
             for (ConsumerRecord<String, String> record : records) {
                 buffer.add(record);
             }
             if (buffer.size() >= minBatchSize) {
//                 insertIntoDb(buffer); 執行相關邏輯
                 consumer.commitSync();//提交offset
                 buffer.clear();
             }
         }
    }
}

Streams:

public class MyKafkaStreams {
    public void test(){
        Map<String, Object> props = new HashMap<>();
         props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-stream-processing-application");
         props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
         props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
         props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
         StreamsConfig config = new StreamsConfig(props);

         KStreamBuilder builder = new KStreamBuilder();
         builder.stream("my-input-topic").mapValues(value -> value.toString()+"!!!").to("my-output-topic");

         KafkaStreams streams = new KafkaStreams(builder, config);
         streams.start();
    }
}

注意點：

將producer寫成單例模式，有助於減小zookeeper端佔用的資源。Producer自身是線程安全的類，只要封裝得當就能最恰當的發揮好producer的做用。(ZkClient去鏈接zookeeper的server時候都會建立sendThread和eventThread兩個線程，其中sendThread主要用於client與server端之間的網絡鏈接，真正的處理線程由eventThread來執行。Zookeeper是一個分佈式的協調框架，而分佈式應用中常常會出現動態的增長或刪除節點的操做，因此爲了實時瞭解分佈式整個節點的數量和基本信息，就有必要維護一個長鏈接的線程與服務端保持鏈接。另外zookeeper鏈接時佔用的時間也比較長，若是每次生產數據時都鏈接發起一次鏈接勢必形成了大量時間的耗費。)
kafka是將消息按照topic的形式存儲，一個topic會按照partition存在同一個文件夾下，目錄在config/server.properties中指定：

# The directory under which to store log files
log.dir=/tmp/kafka-logs

在消息系統中都會有這樣一個問題存在，數據消費狀態這個信息到底存哪裏。是存在consumer端，仍是存在broker端。對於這樣的爭論，通常會出現三種狀況：

At most once ：消息一旦發出就立馬標記已消費，不會再有第二發生即便失敗了，缺點是容易丟失消息。
At least once ：消息至少發送一次,若是消息未能接受成功,可能會重發,直到接收成功.
Exactly once ：每一個消息僅發生一次，並且一次就能確保到達。這是理想狀態。(kafka0.11支持冪等以後，在開啓冪等的狀況下，就是這種模式)

at most once: 消費者fetch消息,而後保存offset,而後處理消息;當client保存offset以後,可是在消息處理過程當中出現了異常,致使部分消息未能繼續處理.那麼此後"未處理"的消息將不能被fetch到,這就是"atmost once".

at least once: 消費者fetch消息,而後處理消息,而後保存offset.若是消息處理成功以後,可是在保存offset階段zookeeper異常致使保存操做未能執行成功,這就致使接下來再次fetch時可能得到上次已經處理過的消息,這就是"at least once"，緣由offset沒有及時的提交給zookeeper，zookeeper恢復正常仍是以前offset狀態.