• <code id="f6tm2"><nobr id="f6tm2"><track id="f6tm2"></track></nobr></code>
  • <th id="f6tm2"></th>

    <pre id="f6tm2"><nobr id="f6tm2"></nobr></pre>
    <big id="f6tm2"></big>

        <big id="f6tm2"><nobr id="f6tm2"></nobr></big>
        <th id="f6tm2"><video id="f6tm2"><span id="f6tm2"></span></video></th>

        全國咨詢熱線: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+人工智能

        香蕉电影