<i id="ufncm"></i>

    <wbr id="ufncm"></wbr>
    1. 全國咨詢熱線:400-618-4000

      kafka自定義攔截器實例教程[傳智播客]

      創建時間:2019年09月17日15時32分

      1、攔截器原理

      Producer攔截器(interceptor)是在Kafka 0.10版本被引入的,主要用于實現clients端的定制化控制邏輯。

      對于producer而言,interceptor使得用戶在消息發送前以及producer回調邏輯前有機會對消息做一些定制化需求,比如修改消息等。同時,producer允許用戶指定多個interceptor按序作用于同一條消息從而形成一個攔截鏈(interceptor chain)。Intercetpor的實現接口是org.apache.kafka.clients.producer.ProducerInterceptor,其定義的方法包括:

      (1)configure(configs)

      獲取配置信息和初始化數據時調用。

      (2)onSend(ProducerRecord):

      該方法封裝進KafkaProducer.send方法中,即它運行在用戶主線程中。Producer確保在消息被序列化以及計算分區前調用該方法。用戶可以在該方法中對消息做任何操作,但最好保證不要修改消息所屬的topic和分區,否則會影響目標分區的計算。

      (3)onAcknowledgement(RecordMetadata, Exception):

      該方法會在消息從RecordAccumulator成功發送到Kafka Broker之后,或者在發送過程中失敗時調用。并且通常都是在producer回調邏輯觸發之前。onAcknowledgement運行在producer的IO線程中,因此不要在該方法中放入很重的邏輯,否則會拖慢producer的消息發送效率。

      (4)close:

      關閉interceptor,主要用于執行一些資源清理工作

      如前所述,interceptor可能被運行在多個線程中,因此在具體實現時用戶需要自行確保線程安全。另外倘若指定了多個interceptor,則producer將按照指定順序調用它們,并僅僅是捕獲每個interceptor可能拋出的異常記錄到錯誤日志中而非在向上傳遞。這在使用過程中要特別留意。

      2、攔截器案例

      1)需求:

      實現一個簡單的雙interceptor組成的攔截鏈。第一個interceptor會在消息發送前將時間戳信息加到消息value的最前部;第二個interceptor會在消息發送后更新成功發送消息數或失敗發送消息數。

      2)案例實操

      (1)增加時間戳攔截器

      package com.heima.kafka.interceptor;
      import java.util.Map;
      import org.apache.kafka.clients.producer.ProducerInterceptor;
      import org.apache.kafka.clients.producer.ProducerRecord;
      import org.apache.kafka.clients.producer.RecordMetadata;
      public class TimeInterceptor implements ProducerInterceptor<String, String> {
      @Override
      public void configure(Map<String, ?> configs) {
      }
      @Override
      public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
      // 創建一個新的record,把時間戳寫入消息體的最前部
      return new ProducerRecord(record.topic(), record.partition(), record.timestamp(), record.key(),
      System.currentTimeMillis() + "," + record.value().toString());
      }
      @Override
      public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
      }
      @Override
      public void close() {
      }
      }


      kafka自定義攔截器

      (2)統計發送消息成功和發送失敗消息數,并在producer關閉時打印這兩個計數器

      package com.heima.kafka.interceptor;
      import java.util.Map;
      import org.apache.kafka.clients.producer.ProducerInterceptor;
      import org.apache.kafka.clients.producer.ProducerRecord;
      import org.apache.kafka.clients.producer.RecordMetadata;
      public class CounterInterceptor implements ProducerInterceptor<String, String>{
          private int errorCounter = 0;
          private int successCounter = 0;
      @Override
      public void configure(Map<String, ?> configs) {
      }
      @Override
      public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
      return record;
      }
      @Override
      public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
      // 統計成功和失敗的次數
              if (exception == null) {
                  successCounter++;
              } else {
                  errorCounter++;
              }
      }
      @Override
      public void close() {
              // 保存結果
              System.out.println("Successful sent: " + successCounter);
              System.out.println("Failed sent: " + errorCounter);
      }
      }

      (3)producer主程序

      package com.heima.kafka.interceptor;

      import java.util.ArrayList;
      import java.util.List;
      import java.util.Properties;
      import org.apache.kafka.clients.producer.KafkaProducer;
      import org.apache.kafka.clients.producer.Producer;
      import org.apache.kafka.clients.producer.ProducerConfig;
      import org.apache.kafka.clients.producer.ProducerRecord;
      public class InterceptorProducer {
      public static void main(String[] args) throws Exception {
      // 1 設置配置信息
      Properties props = new Properties();
      props.put("bootstrap.servers", "hadoop102:9092");
      props.put("acks", "all");
      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");
      // 2 構建攔截鏈
      List<String> interceptors = new ArrayList<>();
      interceptors.add("com.heima.kafka.interceptor.TimeInterceptor"); interceptors.add("com.heima.kafka.interceptor.CounterInterceptor"); 
      props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
      String topic = "first";
      Producer<String, String> producer = new KafkaProducer<>(props);
      // 3 發送消息
      for (int i = 0; i < 10; i++) {
          ProducerRecord<String, String> record = new ProducerRecord<>(topic, "message" + i);
          producer.send(record);
      }
      // 4 一定要關閉producer,這樣才會調用interceptor的close方法
      producer.close();
      }
      }
      3)測試
      (1)在kafka上啟動消費者,然后運行客戶端java程序。

      [root@hadoop102 kafka]$ bin/kafka-console-consumer.sh \
      --bootstrap-server hadoop102:9092 --from-beginning --topic first
      1501904047034,message0
      1501904047225,message1
      1501904047230,message2
      1501904047234,message3
      1501904047236,message4
      1501904047240,message5
      1501904047243,message6
      1501904047246,message7
      1501904047249,message8
      1501904047252,message9
      推薦了解:
      大數據培訓
      web前端開發
      python+人工智能

      香蕉电影