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());
            }
        }

    }

}