Kafka实战系列--Kafka API使用体验

　　org.apache.kafka
　　kafka_2.9.2
　　0.8.1.1

public class SimplePartitioner implements Partitioner {
　　public SimplePartitioner (VerifiableProperties props) {
　　}
　　public int partition(Object key, int numPartitions) {
　　　　return (key.hashCode() & 0x0FFFFFFF) % numPartitions;
　　}
}

Properties props = new Properties();
// 配置metadata.broker.list, 为了高可用, 最好配两个broker实例
props.put("metadata.broker.list", "127.0.0.1:9092");
// serializer.class为消息的序列化类
props.put("serializer.class", "kafka.serializer.StringEncoder");
// 设置Partition类, 对队列进行合理的划分
props.put("partitioner.class", "mmxf.kafka.practise.SimplePartitioner");
// ACK机制, 消息发送需要kafka服务端确认
props.put("request.required.acks", "1");

ProducerConfig config = new ProducerConfig(props);
Producer producer = new Producer(config);

// KeyedMessage
// 　　K对应Partition Key的类型
// 　　V对应消息本身的类型
//　　 topic: "test", key: "key", message: "message"
KeyedMessage message = new KeyedMessage("test", "key", "message");
producer.send(message);

// 关闭producer实例
producer.close();

public static void main(String[] args) {

　　// *) 创建ConsumerConfig
　　Properties props = new Properties();
　　// 设置zookeeper的链接地址
　　props.put("zookeeper.connect", "127.0.0.1:2181");
　　// 设置group id
　　props.put("group.id", "group_id");
　　// kafka的group 消费记录是保存在zookeeper上的, 但这个信息在zookeeper上不是实时更新的, 需要有个间隔时间更新
　　props.put("auto.commit.interval.ms", "1000");

　　ConsumerConfig consumerConfig = new ConsumerConfig(props);
　　ConsumerConnector consumer = (ConsumerConnector) Consumer.createJavaConsumerConnector(consumerConfig);

　　String topic = "test";
　　int threadNum = 1;

　　// *) 设置Topic=>Thread Num映射关系, 构建具体的流
　　Map topicCountMap = new HashMap();
　　topicCountMap.put(topic,threadNum);
　　Map>> consumerMap = consumer.createMessageStreams(topicCountMap);

　　List> streams = consumerMap.get(topic);

　　// *) 启动线程池去消费对应的消息
　　ExecutorService executor = Executors.newCachedThreadPool();
　　for ( final KafkaStream stream : streams ) {
　　　　executor.submit(new Runnable() {
　　　　　　public void run() {
　　　　　　　　ConsumerIterator iter = stream.iterator();
　　　　　　　　while ( iter.hasNext() ) {
　　　　　　　　　　MessageAndMetadata mam = iter.next();
　　　　　　　　　　System.out.println(
　　　　　　　　　　　　String.format("thread_id: %d, key: %s, value: %s",
　　　　　　　　　　　　　　　　Thread.currentThread().getId(),
　　　　　　　　　　　　　　　　new String(mam.key()),
　　　　　　　　　　　　　　　　new String(mam.message())
　　　　　　　　　　　　　　)
　　　　　　　　　　);　　
　　　　　　　　}
　　　　　　}
　　　　});
　　}

　　try {
　　　　Thread.sleep(1000 * 10);
　　} catch (InterruptedException e) {
　　　　e.printStackTrace();
　　}

　　// *) 优雅地退出
　　consumer.shutdown();
　　executor.shutdown();

　　while ( !executor.isTerminated() ) {
　　　　try {
　　　　　　executor.awaitTermination(1, TimeUnit.SECONDS);
　　　　} catch (InterruptedException e) {
　　　　}
　　}

}

thread_id: 18, key: key, value: message