Kafka producer 自定義攔截器(interceptor)

一、攔截器原理

• 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 可能拋出的例外記錄到錯誤日志中而非在向上傳遞，這在使用程序中要特別留意，

二、攔截器案例

1、準備作業

• 在IDE上創建maven專案，pom檔案添加依賴

<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>1.1.1</version>
</dependency>

• 啟動zookeeper集群

bin/zkServer.sh start

• 啟動kafka集群

bin/kafka-server-start.sh -daemon config/server.properties

• kafka集群開一個消費者

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

2、需求分析

實作一個簡單的雙 interceptor 組成的攔截器鏈，第一個 interceptor 會在訊息發送前將時間戳資訊加到訊息 value 的最前部；第二個 interceptor 會在訊息發送后更新成功發送訊息數和失敗發送訊息數，

3、代碼展示

	3.1、 增加時間戳攔截器

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Map;

public class MyTimeInterceptor implements ProducerInterceptor<String,String> {

    @Override
    public void configure(Map<String, ?> configs) {

    }

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        /* 取出value值 */
        String value = record.value();

        /* 創建一個新的 record，把時間戳寫入訊息體的最前部 */
        return new ProducerRecord(record.topic(),record.partition(),record.key(),System.currentTimeMillis() + "---" + value);
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {

    }

    @Override
    public void close() {

    }
}

	3.2、增加統計次數攔截器，統計發送訊息成功和發送失敗訊息數，并在 producer 關閉時列印這兩個計數器

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Map;

public class MyCountInterceptor implements ProducerInterceptor<String,String> {
    private int success;
    private int error;

    @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 (metadata != null){
            success ++;
        }else {
            error ++ ;
        }
    }

    @Override
    public void close() {
        /* 保存結果 */
        System.out.println("success: " + success);
        System.out.println("error: " + error);
    }
}

	3.3、producer 主程式

import java.util.ArrayList;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class MyProducerInterceptor {
    public static void main(String[] args) {

        /* 相當于map操作 */
        Properties properties = new Properties();

        /* kafka 集群，broker-list */
        properties.put("bootstrap.servers", "centos7-1:9092");

        /* 等待所有副本節點的應答 */
        properties.put("acks", "all");

        /* 重試次數 */
        properties.put("retries", Integer.valueOf(3));

        /* 批次大小 */
        properties.put("batch.size", Integer.valueOf(16384));

        /* 等待時間 */
        properties.put("linger.ms", Integer.valueOf(1));

        /* RecordAccumulator 緩沖區大小 */
        properties.put("buffer.memory", Integer.valueOf(33554432));

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

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

        /* 構建攔截器鏈 */
        ArrayList<String> interceptors = new ArrayList<>();
        interceptors.add("com.jh.interceptor.MyTimeInterceptor ");
        interceptors.add("com.jh.interceptor.MyCountInterceptor ");
        /* 添加攔截器鏈 */
        properties.put("interceptor.classes", interceptors);

        /* 創建生產者物件 */
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer(properties);

        /*  發送訊息 */
        for (int i = 0; i < 10000; i++){
            /* 發送的主題、key、value */
            kafkaProducer.send(new ProducerRecord("bigdata", "jh","jh==" + i));
        }

        /* 一定要關閉 producer，這樣才會呼叫 interceptor的 close 方法 */
        kafkaProducer.close();
    }
}

4、運行效果展示