Kafka的api操作

电玩女神 2022-12-29 04:36 184阅读 0赞

## maven依赖： ##

<dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
                <version>2.6.0</version>
            </dependency>
            
            <dependency>
                <groupId>com.fasterxml.jackson.core</groupId>
                <artifactId>jackson-core</artifactId>
                <version>2.2.3</version>
            </dependency>
            <dependency>
                <groupId>com.fasterxml.jackson.core</groupId>
                <artifactId>jackson-databind</artifactId>
                <version>2.2.3</version>
            </dependency>
            <dependency>
                <groupId>com.fasterxml.jackson.core</groupId>
                <artifactId>jackson-annotations</artifactId>
                <version>2.2.3</version>
            </dependency>
            <dependency>
                <groupId>com.fasterxml.jackson.module</groupId>
                <artifactId>jackson-module-jaxb-annotations</artifactId>
                <version>2.2.3</version>
            </dependency>

## 简单模式 ##

Producer采用默认分区方式将消息散列的发送到各个分区当中：

package kafka;
    
    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;
    
    import java.util.Properties;
    import java.util.UUID;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/19 20:14
     * @Version: 1.0
     * @Description:
     */
    public class MineProducer {
        public static void main(String[] args) {
            Properties properties = new Properties();
            properties.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
            //ack模式，all是最慢但最安全的
            properties.put("acks","all");
            //失败重试次数
            properties.put("retries",0);
            //每个分区未发送消息总字节大小（单位：字节），超过设置的值就会提交数据到服务端
            properties.put("batch.size",100);
            //消息在缓冲区保留的时间，超过设置的值就会被提交到服务端
            properties.put("linger.ms", 10000);
            //整个Producer用到总内存的大小，如果缓冲区满了会提交数据到服务端
            //buffer.memory要大于batch.size，否则会报申请内存不足的错误
            //properties.put("buffer.memory", 10240);
            // 指定消息key的序列化器
            properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            // 指定消息value的序列化器
            properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            //props.put("transactional.id", "my_transactional_id");
            Producer<String, String> producer = new KafkaProducer<String, String>(properties);
            //producer.initTransactions();
            try {
                //数据发送必须在beginTransaction()和commitTransaction()中间，否则会报状态不对的异常
                //producer.beginTransaction();
                for (int i = 0; i < 100; i++)
                {
                    producer.send(new ProducerRecord<>("firstTopic", Integer.toString(i), UUID.randomUUID().toString() + (i+1)));
                }
                //producer.commitTransaction();
            } catch (Exception e) {
                // 这些异常不能被恢复，因此必须要关闭并退出Producer
                producer.close();
            }
            producer.close();
        }
    }

配置项

<table> 
 <tbody> 
  <tr> 
   <th>名称</th> 
   <th>说明</th> 
   <th>默认值</th> 
   <th>有效值</th> 
   <th>重要性</th> 
  </tr> 
 </tbody> 
 <tbody> 
  <tr> 
   <td>bootstrap.servers</td> 
   <td>kafka集群的broker-list，如：&lt;br&gt;hadoop01:9092,hadoop02:9092</td> 
   <td>无</td> 
   <td>&nbsp;</td> 
   <td>必选</td> 
  </tr> 
  <tr> 
   <td>acks</td> 
   <td>确保生产者可靠性设置，有三个选项：&lt;br&gt;acks=0:不等待成功返回&lt;br&gt;acks=1:等Leader写成功返回&lt;br&gt;acks=all:等Leader和所有ISR中的Follower写成功返回,all也可以用-1代替</td> 
   <td>-1</td> 
   <td>0,1,-1,all</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>key.serializer</td> 
   <td>key的序列化器</td> 
   <td>&nbsp;</td> 
   <td>ByteArraySerializer&lt;br&gt;StringSerializer</td> 
   <td>必选</td> 
  </tr> 
  <tr> 
   <td>value.serializer</td> 
   <td>value的序列化器</td> 
   <td>&nbsp;</td> 
   <td>ByteArraySerializer&lt;br&gt;StringSerializer</td> 
   <td>必选</td> 
  </tr> 
  <tr> 
   <td>buffer.memory</td> 
   <td>Producer总体内存大小</td> 
   <td>33554432</td> 
   <td>不要超过物理内存，根据实际情况调整</td> 
   <td>建议必选</td> 
  </tr> 
  <tr> 
   <td>compression.type</td> 
   <td>压缩类型&lt;br&gt;压缩最好用于批量处理，批量处理消息越多，压缩性能越好</td> 
   <td>无</td> 
   <td>none、gzip、snappy</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>retries</td> 
   <td>发送失败尝试重发次数</td> 
   <td>0</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>batch.size</td> 
   <td>每个partition的未发送消息大小</td> 
   <td>16384</td> 
   <td>根据实际情况调整</td> 
   <td>建议必选</td> 
  </tr> 
  <tr> 
   <td>client.id</td> 
   <td>附着在每个请求的后面，用于标识请求是从什么地方发送过来的</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>connections.max&lt;br&gt;.idle.ms</td> 
   <td>连接空闲时间超过过久自动关闭（单位毫秒）</td> 
   <td>540000</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>linger.ms</td> 
   <td>数据在缓冲区中保留的时长,0表示立即发送&lt;br&gt;为了减少网络耗时，需要设置这个值&lt;br&gt;太大可能容易导致缓冲区满，阻塞消费者&lt;br&gt;太小容易频繁请求服务端</td> 
   <td>0</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>max.block.ms</td> 
   <td>最大阻塞时长</td> 
   <td>60000</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>max.request.size</td> 
   <td>请求的最大字节数，该值要比batch.size大&lt;br&gt;不建议去更改这个值，如果设置不好会导致程序不报错，但消息又没有发送成功</td> 
   <td>1048576</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>partitioner.class</td> 
   <td>分区类，可以自定义分区类，实现partitioner接口</td> 
   <td>默认是哈希值%partitions</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>receive.buffer.bytes</td> 
   <td>socket的接收缓存空间大小,当阅读数据时使用</td> 
   <td>32768</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>request.timeout.ms</td> 
   <td>等待请求响应的最大时间,超时则重发请求,超过重试次数将抛异常</td> 
   <td>3000</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>send.buffer.bytes</td> 
   <td>发送数据时的缓存空间大小</td> 
   <td>131072</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>timeout.ms</td> 
   <td>控制server等待来自followers的确认的最大时间</td> 
   <td>30000</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>max.in.flight.&lt;br&gt;requests.per.&lt;br&gt;connection</td> 
   <td>kafka可以在一个connection中发送多个请求，叫作一个flight,这样可以减少开销，但是如果产生错误，可能会造成数据的发送顺序改变。</td> 
   <td>5</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>metadata.fetch&lt;br&gt;.timeout.ms</td> 
   <td>从ZK中获取元数据超时时间&lt;br&gt;比如topic\host\partitions</td> 
   <td>60000</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>metadata.max.age.ms</td> 
   <td>即使没有任何partition leader 改变，强制更新metadata的时间间隔</td> 
   <td>300000</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>metric.reporters</td> 
   <td>类的列表，用于衡量指标。实现MetricReporter接口，将允许增加一些类，这些类在新的衡量指标产生时就会改变。JmxReporter总会包含用于注册JMX统计</td> 
   <td>none</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>metrics.num.samples</td> 
   <td>用于维护metrics的样本数</td> 
   <td>2</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>metrics.sample.window.ms</td> 
   <td>metrics系统维护可配置的样本数量，在一个可修正的window size。这项配置配置了窗口大小，例如。我们可能在30s的期间维护两个样本。当一个窗口推出后，我们会擦除并重写最老的窗口</td> 
   <td>30000</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>reconnect.backoff.ms</td> 
   <td>连接失败时，当我们重新连接时的等待时间。这避免了客户端反复重连</td> 
   <td>10</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>retry.backoff.ms</td> 
   <td>在试图重试失败的produce请求之前的等待时间。避免陷入发送-失败的死循环中</td> 
   <td>100</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

更全的配置参考官方文档：[http://kafka.apache.org/documentation/\#producerconfigs][http_kafka.apache.org_documentation_producerconfigs]

## 幂等模式 ##

幂等性：客户端一次或多次操作，最终数据是一致的，比如购买火车票支付时可能显示网络异常，但其实已经扣款成功，用户再次发起扣款不会再触发真正的扣款Kafka只能保证在一个会话中的幂等性

幂等模式只需要将enable.idempotence设置为true，一旦设置了该属性，那么retries默认是Integer.MAX\_VALUE ，acks默认是all。代码的写法和前面例子没什么区别

## 事务模式 ##

事务模式要求数据发送必须包含在事务中，在事务中可以向多个topic发送数据，消费者端最好也使用事务模式读，保证一次能将整个事务的数据全部读取过来。当然消费者也可以不设置为事务读的模式。

package kafka;
    
    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;
    
    import java.util.Properties;
    import java.util.UUID;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/19 20:14
     * @Version: 1.0
     * @Description:
     */
    public class MineProducer {
        public static void main(String[] args) {
            Properties properties = new Properties();
            properties.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
            //ack模式，all是最慢但最安全的
            properties.put("acks","all");
            //失败重试次数
            properties.put("retries",0);
            //每个分区未发送消息总字节大小（单位：字节），超过设置的值就会提交数据到服务端
            properties.put("batch.size",100);
            //消息在缓冲区保留的时间，超过设置的值就会被提交到服务端
            properties.put("linger.ms", 10000);
            //整个Producer用到总内存的大小，如果缓冲区满了会提交数据到服务端
            //buffer.memory要大于batch.size，否则会报申请内存不足的错误
            //properties.put("buffer.memory", 10240);
            // 指定消息key的序列化器
            properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            // 指定消息value的序列化器
            properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            properties.put("transactional.id", "my_transactional_id");
            Producer<String, String> producer = new KafkaProducer<String, String>(properties);
            producer.initTransactions();
            try {
                //数据发送必须在beginTransaction()和commitTransaction()中间，否则会报状态不对的异常
                producer.beginTransaction();
                for (int i = 0; i < 100; i++)
                {
                    producer.send(new ProducerRecord<>("firstTopic", Integer.toString(i), UUID.randomUUID().toString() + (i+1)));
                }
                producer.commitTransaction();
            } catch (Exception e) {
                // 这些异常不能被恢复，因此必须要关闭并退出Producer
                producer.close();
            }
            producer.close();
        }
    }

### 自定义分区： ###

package kafka;
    
    import org.apache.kafka.clients.producer.Partitioner;
    import org.apache.kafka.common.Cluster;
    
    import java.util.Map;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/20 7:09
     * @Version: 1.0
     * @Description:
     */
    public class MinePartitioner implements Partitioner {
        @Override
        public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
            return 1;
        }
    
        @Override
        public void close() {
    
        }
    
        @Override
        public void configure(Map<String, ?> map) {
    
        }
    }

package kafka;
    
    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;
    
    import java.util.Properties;
    import java.util.UUID;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/19 20:14
     * @Version: 1.0
     * @Description:
     */
    public class MineProducer {
        public static void main(String[] args) {
            Properties properties = new Properties();
            properties.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
            //ack模式，all是最慢但最安全的
            properties.put("acks","all");
            //失败重试次数
            properties.put("retries",1);
            //每个分区未发送消息总字节大小（单位：字节），超过设置的值就会提交数据到服务端
            properties.put("batch.size",100);
            //消息在缓冲区保留的时间，超过设置的值就会被提交到服务端
            properties.put("linger.ms", 10000);
            //整个Producer用到总内存的大小，如果缓冲区满了会提交数据到服务端
            //buffer.memory要大于batch.size，否则会报申请内存不足的错误
            //properties.put("buffer.memory", 10240);
            // 指定消息key的序列化器
            properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            // 指定消息value的序列化器
            properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    
            properties.put("partitioner.class","kafka.MinePartitioner");
    
            properties.put("transactional.id", "my_transactional_id");
            Producer<String, String> producer = new KafkaProducer<String, String>(properties);
            producer.initTransactions();
            try {
                //数据发送必须在beginTransaction()和commitTransaction()中间，否则会报状态不对的异常
                producer.beginTransaction();
                for (int i = 0; i < 100; i++)
                {
                    producer.send(new ProducerRecord<>("firstTopic", Integer.toString(i), UUID.randomUUID().toString() + "------" + (i+1)));
                }
                producer.commitTransaction();
            } catch (Exception e) {
                // 这些异常不能被恢复，因此必须要关闭并退出Producer
                producer.close();
            }
            producer.close();
        }
    }

### 消费者代码： ###

package kafka;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    
    import java.time.Duration;
    import java.util.Arrays;
    import java.util.Collections;
    import java.util.Properties;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/20 7:26
     * @Version: 1.0
     * @Description:
     */
    public class MineConsumer {
        public static void main(String[] args) {
            //配置项
            Properties props = new Properties();
            //kafka的服务器地址
            props.setProperty("bootstrap.servers", "hadoop01:9092");
            //group.id
            props.setProperty("group.id", "test");
            //自动提交(offset)
            props.setProperty("enable.auto.commit", "true");
            //自定义处理间隔时间(1000ms)
            props.setProperty("auto.commit.interval.ms", "1000");
            props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
            //定义消费者的topic
            consumer.subscribe(Collections.singletonList("firstTopic"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                }
            }
        }
    }

# **[Kafka producer拦截器(interceptor)][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)：

该方法会在消息被应答之前或消息发送失败时调用，并且通常都是在producer回调逻辑触发之前。onAcknowledgement运行在producer的IO线程中，因此不要在该方法中放入很重的逻辑，否则会拖慢producer的消息发送效率

（4）close：

关闭interceptor，主要用于执行一些资源清理工作

如前所述，interceptor可能被运行在多个线程中，因此在具体实现时用户需要自行确保线程安全。另外倘若指定了多个interceptor，则producer将按照指定顺序调用它们，并仅仅是捕获每个interceptor可能抛出的异常记录到错误日志中而非在向上传递。这在使用过程中要特别留意。

**TimeInterceptor.java：**

package kafka;
    
    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;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/20 7:45
     * @Version: 1.0
     * @Description:
     */
    public class TimeInterceptor implements ProducerInterceptor<String,String> {
        @Override
        public void configure(Map<String, ?> configs) {
    
        }
    
        @Override
        public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
            return new ProducerRecord<String, String>(record.topic(),record.partition(),record.timestamp()
            ,record.key(),record.value() +  "---" + System.currentTimeMillis());
        }
    
        @Override
        public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
    
        }
    
        @Override
        public void close() {
    
        }
    }

**ContentInterceptor.java： **

package kafka;
    
    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;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/20 7:51
     * @Version: 1.0
     * @Description:
     */
    public class ContentInterceptor implements ProducerInterceptor<String,String> {
    
        private static int successCount = 0;
        private static int failedCount = 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){
                failedCount ++;
            }
            else{
                successCount ++;
            }
        }
    
        @Override
        public void close() {
            System.out.println("successCount = " + successCount + "; failedCount = " + failedCount);
        }
    }

**InterceptorProducer.java： **

package kafka;
    
    import org.apache.kafka.clients.producer.*;
    
    import java.util.Arrays;
    import java.util.Properties;
    import java.util.UUID;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/19 20:14
     * @Version: 1.0
     * @Description:
     */
    public class InterceptorProducer {
        public static void main(String[] args) {
            Properties properties = new Properties();
            properties.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
            //ack模式，all是最慢但最安全的
            properties.put("acks","all");
            //失败重试次数
            properties.put("retries",1);
            //每个分区未发送消息总字节大小（单位：字节），超过设置的值就会提交数据到服务端
            properties.put("batch.size",100);
            //消息在缓冲区保留的时间，超过设置的值就会被提交到服务端
            properties.put("linger.ms", 10000);
            //整个Producer用到总内存的大小，如果缓冲区满了会提交数据到服务端
            //buffer.memory要大于batch.size，否则会报申请内存不足的错误
            //properties.put("buffer.memory", 10240);
            // 指定消息key的序列化器
            properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            // 指定消息value的序列化器
            properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    
            properties.put("partitioner.class","kafka.MinePartitioner");
    
            String [] interceptors = {"kafka.TimeInterceptor","kafka.ContentInterceptor"};
    
            properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, Arrays.asList(interceptors));
    
            properties.put("transactional.id", "my_transactional_id");
            Producer<String, String> producer = new KafkaProducer<String, String>(properties);
            producer.initTransactions();
            try {
                //数据发送必须在beginTransaction()和commitTransaction()中间，否则会报状态不对的异常
                producer.beginTransaction();
                for (int i = 0; i < 100; i++)
                {
                    producer.send(new ProducerRecord<>("firstTopic", Integer.toString(i), UUID.randomUUID().toString() + "------" + (i + 1))
                            , new Callback() {
                                @Override
                                public void onCompletion(RecordMetadata metadata, Exception exception) {
                                    System.out.println("发送成功 ===" + metadata.offset());
                                }
                            });
                }
                producer.commitTransaction();
            } catch (Exception e) {
                // 这些异常不能被恢复，因此必须要关闭并退出Producer
                e.printStackTrace();
                producer.close();
            }
            producer.close();
        }
    }

# Stream： #

[http_kafka.apache.org_documentation_producerconfigs]: http://kafka.apache.org/documentation/#producerconfigs
[Kafka producer_interceptor]: http://www.cnblogs.com/huxi2b/p/7072447.html