Kafka Producer接口-白红宇

参考，

, 0.8版本，关于producer，consumer，broker所有的配置

因为Producer相对于consumer比较简单，直接看代码，需要注意的点

1. 配置参数，详细参考上面链接

1.1 metadata.broker.list，不同于0.7，不需要给出zk的地址，而是给出一些broker地址，不用全部，这里建议给两个防止一个不可用

Kafka会自己找到相应topic，partition的leader broker

1.2 serializer.class，需要给出message的序列化的encoder，这里使用的是简单的StringEncoder

并且对于key还可以单独的设定，"key.serializer.class"

注意，除非明确知道message编码，否则不要直接使用StringEncoder，

因为源码中的逻辑是如果没有在初始化时指定编码会默认按UTF8转码，会导致乱码

所以不明确的时候，不要指定serializer.class，默认的encoder逻辑是直接将byte[]放入broker，不会做转码

1.3 partitioner.class，可以不设置，默认就是random partition，当然这里可以自定义，如何根据key来选择partition

1.4 request.required.acks，是否要求broker给出ack，如果不设置默认是'fire and forget'，会丢数据

默认为0，即和0.7一样，发完不会管是否成功，lowest latency but the weakest durability

1, 等待leader replica的ack，否则重发，折中的方案，当leader在同步数据前dead，会丢数据

-1，等待all in-sync replicas的ack，只要有一个replica活着，就不会丢数据

1.5 producer.type，

sync，单条发送

async，buffer一堆请求后，再一起发送

如果不是对丢数据非常敏感，请设为async，因为对throughput帮助很大，但是当client crash时，会丢数据

1.6 compression.codec
支持"none", "gzip" and "snappy"

可以通过，compressed.topics，来指定压缩的topic

当producer.type选择async的时候，需要关注如下配置

queue.buffering.max.ms (5000), 最大buffer数据的时间，默认是5秒

batch.num.messages (200), batch发送的数目，默认是200，producer会等待buffer的messages数目达到200或时间超过5秒，才发送数据

queue.buffering.max.messages (10000), 最多可以buffer的message数目，超过要么producer block或把数据丢掉

queue.enqueue.timeout.ms (-1), 默认是-1，即达到buffer最大meessage数目时，producer会block

设为0，达到buffer最大meessage数目时会丢掉数据

2. Producer发送的是kv数据

无论Producer或KeyedMessage都是<String, String>的泛型，这里是指key和value的类型

import java.util.*; import kafka.javaapi.producer.Producer;import kafka.producer.KeyedMessage;import kafka.producer.ProducerConfig; public class TestProducer {    public static void main(String[] args) {        long events = Long.parseLong(args[0]);        Random rnd = new Random();         Properties props = new Properties();        props.put("metadata.broker.list", "host1:9092, host2:9092 "); //        props.put("serializer.class", "kafka.serializer.StringEncoder");        props.put("partitioner.class", "example.producer.SimplePartitioner"); //可以不设置        props.put("request.required.acks", "1");         ProducerConfig config = new ProducerConfig(props);         Producer
     
       producer = new Producer
      
       (config);         for (long nEvents = 0; nEvents < events; nEvents++) {                long runtime = new Date().getTime();                 String ip = “192.168.2.” + rnd.nextInt(255);                String msg = runtime + “,www.example.com,” + ip;                KeyedMessage
       
         data = new KeyedMessage
        
         ("page_visits", ip, msg); //指定topic，key，value               producer.send(data);        }        producer.close();    }}

对于自定义partitioner也很简单，

对于partition，两个参数，key和partitions的数目

所要完成的逻辑就是，如果根据key在partitions中挑选一个合适的partition

import kafka.producer.Partitioner;import kafka.utils.VerifiableProperties; public class SimplePartitioner implements Partitioner {    public SimplePartitioner (VerifiableProperties props) {     }     public int partition(String key, int a_numPartitions) {        int partition = 0;        int offset = key.lastIndexOf('.');        if (offset > 0) {           partition = Integer.parseInt( key.substring(offset+1)) % a_numPartitions;        }       return partition;  } }

2014-06-23