import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;

public class CustomPartitioner implements Partitioner {

  /**
   *
   * @param topic 當(dāng)前的發(fā)送的topic
   * @param key  當(dāng)前的key值
   * @param keyBytes 當(dāng)前的key的字節(jié)數(shù)組
   * @param value 當(dāng)前的value值
   * @param valueBytes 當(dāng)前的value的字節(jié)數(shù)組
   * @param cluster
   * @return
   */
  @Override
  public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    //這邊根據(jù)返回值就是分區(qū)號, 這邊就是固定發(fā)送到三號分區(qū)
    return 3;
  }

  @Override
  public void close() {

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

  }

}

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;
import java.util.UUID;

public class MessageInterceptor implements ProducerInterceptor<String, String> {

  @Override
  public void configure(Map<String, ?> configs) {
    System.out.println("這是MessageInterceptor的configure方法");
  }

  /**
   * 這個是消息發(fā)送之前進(jìn)行處理
   *
   * @param record
   * @return
   */
  @Override
  public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
    // 創(chuàng)建一個新的record，把uuid入消息體的最前部
    System.out.println("為消息添加uuid");
    return new ProducerRecord(record.topic(), record.partition(), record.timestamp(), record.key(),
        UUID.randomUUID().toString().replace("-", "") + "," + record.value());
  }

  /**
   * 這個是生產(chǎn)者回調(diào)函數(shù)調(diào)用之前處理
   * @param metadata
   * @param exception
   */
  @Override
  public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
    System.out.println("MessageInterceptor攔截器的onAcknowledgement方法");
  }

  @Override
  public void close() {
    System.out.println("MessageInterceptor close 方法");
  }
}

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) {
    System.out.println("這是CounterInterceptor的configure方法");
  }

  @Override
  public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
    System.out.println("CounterInterceptor計數(shù)過濾器不對消息做任何操作");
    return record;
  }

  @Override
  public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
    // 統(tǒng)計成功和失敗的次數(shù)
    System.out.println("CounterInterceptor過濾器執(zhí)行統(tǒng)計失敗和成功數(shù)量");
    if (exception == null) {
      successCounter++;
    } else {
      errorCounter++;
    }
  }

  @Override
  public void close() {
    // 保存結(jié)果
    System.out.println("Successful sent: " + successCounter);
    System.out.println("Failed sent: " + errorCounter);
  }
}

import org.apache.kafka.clients.producer.*;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class Producer1 {
  public static void main(String[] args) throws Exception {
    Properties props = new Properties();
    // Kafka服務(wù)端的主機名和端口號
    props.put("bootstrap.servers", "localhost:9092");
    // 等待所有副本節(jié)點的應(yīng)答
    props.put("acks", "all");
    // 消息發(fā)送最大嘗試次數(shù)
    props.put("retries", 0);
    // 一批消息處理大小
    props.put("batch.size", 16384);
    // 請求延時，可能生產(chǎn)數(shù)據(jù)太快了
    props.put("linger.ms", 1);
    // 發(fā)送緩存區(qū)內(nèi)存大小，數(shù)據(jù)是先放到生產(chǎn)者的緩沖區(qū)
    props.put("buffer.memory", 33554432);
    // key序列化
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    // value序列化
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    // 具體的分區(qū)類
    props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "kafka.CustomPartitioner");
    //定義攔截器
    List<String> interceptors = new ArrayList<>();
    interceptors.add("kafka.MessageInterceptor");
    interceptors.add("kafka.CounterInterceptor");
    props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);

    Producer<String, String> producer = new KafkaProducer<>(props);
    for (int i = 0; i < 1; i++) {
      producer.send(new ProducerRecord<String, String>("test_0515", i + "", "xxx-" + i), new Callback() {
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
          System.out.println("這是producer回調(diào)函數(shù)");
        }
      });
    }
    /*System.out.println("現(xiàn)在執(zhí)行關(guān)閉producer");
    producer.close();*/
    producer.close();
  }
}