Kafka Java Producer代碼實(shí)例的詳細(xì)解析

import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;

/**
 * Created by gerry on 12/21.
 */
public class JavaKafkaProducerPartitioner implements Partitioner {

  /**
   * 無參構(gòu)造函數(shù)
   */
  public JavaKafkaProducerPartitioner() {
    this(new VerifiableProperties());
  }

  /**
   * 構(gòu)造函數(shù)，必須給定
   *
   * @param properties 上下文
   */
  public JavaKafkaProducerPartitioner(VerifiableProperties properties) {
    // nothings
  }

  @Override
  public int partition(Object key, int numPartitions) {
    int num = Integer.valueOf(((String) key).replaceAll("key_", "").trim());
    return num % numPartitions;
  }
}

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.apache.log4j.Logger;

import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.ThreadLocalRandom;

/**
 * Created by gerry on 12/21.
 */
public class JavaKafkaProducer {
  private Logger logger = Logger.getLogger(JavaKafkaProducer.class);
  public static final String TOPIC_NAME = "test";
  public static final char[] charts = "qazwsxedcrfvtgbyhnujmikolp1234567890".toCharArray();
  public static final int chartsLength = charts.length;


  public static void main(String[] args) {
    String brokerList = "192.168.187.149:9092";
    brokerList = "192.168.187.149:9092,192.168.187.149:9093,192.168.187.149:9094,192.168.187.149:9095";
    brokerList = "192.168.187.146:9092";
    Properties props = new Properties();
    props.put("metadata.broker.list", brokerList);
    /**
     * 0表示不等待結(jié)果返回<br/>
     * 1表示等待至少有一個(gè)服務(wù)器返回?cái)?shù)據(jù)接收標(biāo)識(shí)<br/>
     * -1表示必須接收到所有的服務(wù)器返回標(biāo)識(shí)，及同步寫入<br/>
     * */
    props.put("request.required.acks", "0");
    /**
     * 內(nèi)部發(fā)送數(shù)據(jù)是異步還是同步
     * sync：同步, 默認(rèn)
     * async：異步
     */
    props.put("producer.type", "async");
    /**
     * 設(shè)置序列化的類
     * 可選：kafka.serializer.StringEncoder
     * 默認(rèn)：kafka.serializer.DefaultEncoder
     */
    props.put("serializer.class", "kafka.serializer.StringEncoder");
    /**
     * 設(shè)置分區(qū)類
     * 根據(jù)key進(jìn)行數(shù)據(jù)分區(qū)
     * 默認(rèn)是：kafka.producer.DefaultPartitioner ==> 安裝key的hash進(jìn)行分區(qū)
     * 可選:kafka.serializer.ByteArrayPartitioner ==> 轉(zhuǎn)換為字節(jié)數(shù)組后進(jìn)行hash分區(qū)
     */
    props.put("partitioner.class", "JavaKafkaProducerPartitioner");

    // 重試次數(shù)
    props.put("message.send.max.retries", "3");

    // 異步提交的時(shí)候(async)，并發(fā)提交的記錄數(shù)
    props.put("batch.num.messages", "200");

    // 設(shè)置緩沖區(qū)大小，默認(rèn)10KB
    props.put("send.buffer.bytes", "102400");

    // 2. 構(gòu)建Kafka Producer Configuration上下文
    ProducerConfig config = new ProducerConfig(props);

    // 3. 構(gòu)建Producer對象
    final Producer<String, String> producer = new Producer<String, String>(config);

    // 4. 發(fā)送數(shù)據(jù)到服務(wù)器，并發(fā)線程發(fā)送
    final AtomicBoolean flag = new AtomicBoolean(true);
    int numThreads = 50;
    ExecutorService pool = Executors.newFixedThreadPool(numThreads);
    for (int i = 0; i < 5; i++) {
      pool.submit(new Thread(new Runnable() {
        @Override
        public void run() {
          while (flag.get()) {
            // 發(fā)送數(shù)據(jù)
            KeyedMessage message = generateKeyedMessage();
            producer.send(message);
            System.out.println("發(fā)送數(shù)據(jù):" + message);

            // 休眠一下
            try {
              int least = 10;
              int bound = 100;
              Thread.sleep(ThreadLocalRandom.current().nextInt(least, bound));
            } catch (InterruptedException e) {
              e.printStackTrace();
            }
          }

          System.out.println(Thread.currentThread().getName() + " shutdown....");
        }
      }, "Thread-" + i));

    }

    // 5. 等待執(zhí)行完成
    long sleepMillis = 600000;
    try {
      Thread.sleep(sleepMillis);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    flag.set(false);

    // 6. 關(guān)閉資源

    pool.shutdown();
    try {
      pool.awaitTermination(6, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
    } finally {
      producer.close(); // 最后之后調(diào)用
    }
  }

  /**
   * 產(chǎn)生一個(gè)消息
   *
   * @return
   */
  private static KeyedMessage<String, String> generateKeyedMessage() {
    String key = "key_" + ThreadLocalRandom.current().nextInt(10, 99);
    StringBuilder sb = new StringBuilder();
    int num = ThreadLocalRandom.current().nextInt(1, 5);
    for (int i = 0; i < num; i++) {
      sb.append(generateStringMessage(ThreadLocalRandom.current().nextInt(3, 20))).append(" ");
    }
    String message = sb.toString().trim();
    return new KeyedMessage(TOPIC_NAME, key, message);
  }

  /**
   * 產(chǎn)生一個(gè)給定長度的字符串
   *
   * @param numItems
   * @return
   */
  private static String generateStringMessage(int numItems) {
    StringBuilder sb = new StringBuilder();
    for (int i = 0; i < numItems; i++) {
      sb.append(charts[ThreadLocalRandom.current().nextInt(chartsLength)]);
    }
    return sb.toString();
  }
}

<properties>
  <kafka.version>0.8.2.1</kafka.version>
</properties>

<dependencies>
  <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.10</artifactId>
    <version>${kafka.version}</version>
  </dependency>
</dependencies>