kafka生產者Producer、消費者Consumer的攔截器interceptor

1、Producer的攔截器interceptor，和consumer端的攔截器interceptor是在kafka0.10版本被引入的，主要用於實現clients端的定制化控制邏輯，生產者攔截器可以用在消息發送前做一些准備工作，使用場景，如下所示：

　　1）、按照某個規則過濾掉不符合要求的消息。
　　2）、修改消息的內容。
　　3）、統計類需求。

package com.demo.kafka.listener;

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;

/**
 * 生產者攔截器
 * 
 * @author 生產者攔截器
 *
 */

public class ProducerInterceptorPrefix implements ProducerInterceptor<String, String> {

    // 發送成功計數
    private volatile long sendSuccess = 0;

    // 發送失敗計數
    private volatile long sendFailure = 0;

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

    }

    /**
     * 發送消息已經操作消息的方法
     */
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        String modifiedValue = "前綴prefix : " + record.value();
        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(
                record.topic(), // 主題
                record.partition(), // 分區
                record.timestamp(), // 時間戳
                record.key(), // key值
                modifiedValue,  // value值
                record.headers()); // 消息頭
        return producerRecord;
    }

    /**
     * ack確認的方法
     */
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if(exception == null) {
            sendSuccess++;
        }else {
            sendFailure++;
        }
    }

    /**
     * 關閉的方法，發送成功之后會將攔截器關閉，調用此方法
     */
    @Override
    public void close() {
        double successRation = (double)sendSuccess / (sendSuccess + sendFailure);
        System.out.println("【INFO 】 發送成功率： " + String.format("%f", successRation * 100) + "%");
    }

}

生產者客戶端要配置一下Producer的攔截器interceptor，如下所示：

package com.demo.kafka.producer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import com.demo.kafka.listener.ProducerInterceptorPrefix;

public class KafkaProducerSimple {

    // 設置服務器地址
    private static final String brokerList = "192.168.110.142:9092";

    // 設置主題
    private static final String topic = "topic-demo";

    public static void main(String[] args) {
        Properties properties = new Properties();
        // 設置key的序列化器
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 設置重試次數
        properties.put(ProducerConfig.RETRIES_CONFIG, 10);

        // 設置值的序列化器
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 打印輸出序列化器的路徑信息
        System.err.println(StringSerializer.class.getName());

        // 設置集群地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);

        // 自定義攔截器使用，可以計算發送成功率或者失敗率，進行消息的拼接或者過濾操作
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptorPrefix.class.getName());

        // 將參數配置到生產者對象中
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

        for (int i = 0; i < 100000; i++) {
            // 生產者消息記錄
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, "hello world!!!" + i);
            // 同步獲取消息
//            RecordMetadata recordMetadata = producer.send(record).get();
            producer.send(record);
        }

        // 關閉
        producer.close();
    }

}

消費者代碼，如下所示：

package com.demo.kafka.consumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

public class KafkaConsumerSimple {

    // 設置服務器地址
    private static final String bootstrapServer = "192.168.110.142:9092";

    // 設置主題
    private static final String topic = "topic-demo";

    // 設置消費者組
    private static final String groupId = "group.demo";

    public static void main(String[] args) {
        Properties properties = new Properties();
        // 設置反序列化key參數信息
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 設置反序列化value參數信息
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 設置服務器列表信息
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);

        // 設置消費者組信息
        properties.put("group.id", groupId);

        // 將參數設置到消費者參數中
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

        // 消息訂閱
        consumer.subscribe(Collections.singletonList(topic));

        while (true) {
            // 每隔一秒監聽一次
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            // 獲取到消息信息
            for (ConsumerRecord<String, String> record : records) {
                System.err.println(record.toString());
            }
        }

    }

}

 

2、生產者的acks參數，這個參數用來指定分區中必須有多少副本來收到這條消息，之后生產者才會認為這條消息寫入成功的。acks是生產者客戶端中非常重要的一個參數，它涉及到消息的可靠性和吞吐量之間的權衡。

　　1）、ack等於0，生產者在成功寫入消息之前不會等待任何來自服務器的響應。如果出現問題生產者是感知不到的，消息就丟失了，不過因為生產者不需要等待服務器響應，所以他可以以網絡能夠支持的最大速度發送消息，從而達到很高的吞吐量。
　　2）、acks等於1，默認值為1，只要集群的首領節點收到消息，生產者就會收到一個來自服務器的成功響應。如果消息無法達到首領節點，比如首領節點崩潰，新的首領節點還沒有被選舉出來，生產者會收到一個錯誤響應，為了避免數據丟失，生產者會重發消息。但是這樣還有可能會導致數據丟失，如果收到寫成功通知，此時首領節點還沒有來的及同步數據到follower節點，首領節點崩潰，就會導致數據丟失。
　　3）、acks等於-1，只有當所有參與復制的節點收到消息時候，生產者會收到一個來自服務器額成功響應，這種模式 最安全的，他可以保證不止一個服務器收到消息。

　　注意，acks參數配置的是一個字符串類型，而不是整數類型，如果配置為整數類型會拋出異常信息。

 

3、kafka消費者訂閱主題和分區，創建完消費者后我們便可以訂閱主題了，只需要調用subscribe方法即可，這個方法會接受一個主題列表，如下所示：

　　另外，我們也可以使用正則表達式來匹配多個主題，而且訂閱之后如果又有匹配的新主題，那么這個消費組立即對其進行消費。正則表達式在連接kafka與其他系統非常有用。比如訂閱所有的測試主題。

package com.demo.kafka.consumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;
import java.util.regex.Pattern;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

public class KafkaConsumerSimple {

    // 設置服務器地址
    private static final String bootstrapServer = "192.168.110.142:9092";

    // 設置主題
    private static final String topic = "topic-demo";

    // 設置主題
    private static final String topic2 = "topic-demo2";

    // 設置消費者組
    private static final String groupId = "group.demo";

    public static void main(String[] args) {
        Properties properties = new Properties();
        // 設置反序列化key參數信息
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 設置反序列化value參數信息
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 設置服務器列表信息，必填參數，該參數和生產者相同，，制定鏈接kafka集群所需的broker地址清單，可以設置一個或者多個
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);

        // 設置消費者組信息，消費者隸屬的消費組，默認為空，如果設置為空，則會拋出異常，這個參數要設置成具有一定業務含義的名稱
        properties.put("group.id", groupId);

        // 制定kafka消費者對應的客戶端id，默認為空，如果不設置kafka消費者會自動生成一個非空字符串。
        properties.put("client.id", "consumer.client.id.demo");

        // 將參數設置到消費者參數中
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

        // 消息訂閱
        consumer.subscribe(Collections.singletonList(topic));
        // 可以訂閱多個主題
        consumer.subscribe(Arrays.asList(topic, topic2));
        // 可以使用正則表達式進行訂閱
        consumer.subscribe(Pattern.compile("topic-demo*"));

        // 指定訂閱的分區
        consumer.assign(Arrays.asList(new TopicPartition(topic, 0)));

        while (true) {
            // 每隔一秒監聽一次
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            // 獲取到消息信息
            for (ConsumerRecord<String, String> record : records) {
                System.err.println(record.toString());
            }
        }

    }

}