002.Kafka的Java访问

首先创建Producer，

其次创建Consumer，

最后自定义Partition

 

1）Kafka的生产者KafkaProducer

package com.sinovatech.framework.kafka;

 

import java.util.Properties;

 

import kafka.javaapi.producer.Producer;

import kafka.producer.KeyedMessage;

import kafka.producer.ProducerConfig;

 

public class KafkaProducer {

 private static String brokers = "192.168.30.104:9092,192.168.30.105:9092,192.168.30.106:9092";

 private static String serializer = "kafka.serializer.StringEncoder";

 private static String partition = "com.sinovatech.framework.kafka.KafkaPartitioner";

 private static final String topic = "mall";

 

 private static Properties props = new Properties();

 

 public static void main(String[] args) {

  // serializer.class为消息的序列化类

  props.put("serializer.class", serializer);

  // 配置metadata.broker.list, 为了高可用, 最好配两个broker实例

  props.put("metadata.broker.list", brokers);

  // ACK机制, 消息发送需要kafka服务端确认

  props.put("request.required.acks", "1");

  props.put("partitioner.class", partition);

 

  ProducerConfig config = new ProducerConfig(props);

  Producer<String, String> producer = new Producer<String, String>(config);

 

  for (int i = 0; i < 500; i++) {

   String message = "hello=" + i;

   String key = i + "";

   producer.send(new KeyedMessage<String, String>(topic, key, message));

  }

 }

}

 

1）Kafka的消费者KafkaConsumer

package com.sinovatech.framework.kafka;

 

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Properties;

 

import kafka.consumer.Consumer;

import kafka.consumer.ConsumerConfig;

import kafka.consumer.ConsumerIterator;

import kafka.consumer.KafkaStream;

import kafka.javaapi.consumer.ConsumerConnector;

 

public class KafkaConsumer extends Thread {

 private static String zookeepers = "192.168.30.104:2181,192.168.30.105:2181,192.168.30.106:2181";

 private static Properties props = new Properties();

 private static String group = "mallGroup";

 private static final String topic = "mall";

 private final ConsumerConnector consumer;

 

 public static void main(String[] args) {

  KafkaConsumer consumerThread = new KafkaConsumer();

  consumerThread.start();

 }

 

 public KafkaConsumer() {

  consumer = Consumer.createJavaConsumerConnector(createConsumerConfig());

 }

 

 private static ConsumerConfig createConsumerConfig() {

  // 设置zookeeper的链接地址

  props.put("zookeeper.connect", zookeepers);

  // 设置group id

  props.put("group.id", group);

  // kafka的group 消费记录是保存在zookeeper上的, 但这个信息在zookeeper上不是实时更新的, 需要有个间隔时间更新

  props.put("auto.commit.interval.ms", "1000");

  props.put("zookeeper.session.timeout.ms", "10000");

  return new ConsumerConfig(props);

 }

 

 public void run() {

  // 设置Topic=>Thread Num映射关系, 构建具体的流

  Map<String, Integer> topickMap = new HashMap<String, Integer>();

  topickMap.put(topic, 1);

 

  Map<String, List<KafkaStream<byte[], byte[]>>> streamMap = consumer

    .createMessageStreams(topickMap);

  KafkaStream<byte[], byte[]> stream = streamMap.get(topic).get(0);

  ConsumerIterator<byte[], byte[]> it = stream.iterator();

 

  System.out.println("*********Results********");

  while (it.hasNext()) {

   System.err.println("get data:" + new String(it.next().message()));

   try {

    Thread.sleep(500);

   } catch (InterruptedException e) {

    e.printStackTrace();

   }

  }

 }

 

}

 

3）Kafka自定义Partition

         发 送一条消息时，可以指定这条消息的key，producer根据这个key和partition机制来判断将这条消息发送到哪个parition。 paritition机制可以通过指定producer的paritition.class这一参数来指定，该class必须实现kafka.producer.Partitioner接口。本例中如果key可以被解析为整数则将对应的整数与partition总数取余，该消息会被发送到该数对应的partition。（每个parition都会有个序号）

 

package com.sinovatech.framework.kafka;

 

import kafka.producer.Partitioner;

 

public class KafkaPartitioner implements Partitioner {

// 解决执行报错的问题：Caused by: java.lang.NoSuchMethodException: //com.sinovatech.framework.kafka.KafkaPartitioner.<init>(kafka.utils.VerifiableProperties)

public KafkaPartitioner(VerifiableProperties props) {
 }

 public int partition(Object key, int numPartitions) {

  try {

   return Math.abs(Integer.parseInt((String) key) % numPartitions);

  } catch (Exception e) {

   return Math.abs(key.hashCode() % numPartitions);

  }

 }

 

}