Kafka单机环境开发示例

zhoujun.g

　　在Kafka下载页面下载0.8版本，解压缩。
　　1.修改config目录下的server.properties 里面的host.name为机器的ip。假如部署kafka和开发运行kafka例子为同一台机器，不用修改，用默认的localhost也行。
　　2.修改config目录下的zookeeper.properties 里面的dataDir属性为你需要的目录。
　　3.假如你要配置集群，在kafka解压缩目录下新建zoo_data目录(第一次的时候需要新建),在zoo_data目录新建myid文件，设置内容为1。同时修改zookeeper.properties，具体可参考：solrcloud在tomcat下安装(三)
　　4.启动kafka。

//启动zookeeper server (用&是为了能退出命令行)： bin/zookeeper-server-start.sh config/zookeeper.properties  & //启动kafka server:   bin/kafka-server-start.sh config/server.properties  &

　　5.新建一个生产者例子

import java.util.Properties;   import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.producer.ProducerConfig;   public class KafkaTest { public static void main(String[] args) {   Properties props = new Properties();   props.put("zk.connect", "10.103.22.47:2181");   props.put("serializer.class", "kafka.serializer.StringEncoder");   props.put("metadata.broker.list", "10.103.22.47:9092"); props.put("request.required.acks", "1"); //props.put("partitioner.class", "com.xq.SimplePartitioner"); ProducerConfig config = new ProducerConfig(props);   Producer<String, String> producer = new Producer<String, String>(config);   String ip = "192.168.2.3"; String msg ="this is a messageuuu!"; KeyedMessage<String, String> data = new KeyedMessage<String, String>("test", ip,msg);   producer.send(data); producer.close();   }     }

　　新建一个消费者例子

import java.nio.ByteBuffer; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors;   import kafka.consumer.Consumer; import kafka.consumer.ConsumerConfig; import kafka.consumer.KafkaStream; import kafka.javaapi.consumer.ConsumerConnector; import kafka.message.Message; import kafka.message.MessageAndMetadata;     public class ConsumerSample {   public static void main(String[] args) {   // specify some consumer properties   Properties props = new Properties();   props.put("zookeeper.connect", "10.103.22.47:2181");   props.put("zookeeper.connectiontimeout.ms", "1000000");   props.put("group.id", "test_group");     // Create the connection to the cluster   ConsumerConfig consumerConfig = new ConsumerConfig(props);   ConsumerConnector connector = Consumer.createJavaConsumerConnector(consumerConfig);       Map<String,Integer> topics = new HashMap<String,Integer>();   topics.put("test", 2);   Map<String, List<KafkaStream<byte[], byte[]>>> topicMessageStreams = connector.createMessageStreams(topics);   List<KafkaStream<byte[], byte[]>> streams = topicMessageStreams.get("test"); ExecutorService threadPool = Executors.newFixedThreadPool(2);   for (final KafkaStream<byte[], byte[]> stream : streams) {   threadPool.submit(new Runnable() {   public void run() {   for (MessageAndMetadata msgAndMetadata : stream) {   // process message (msgAndMetadata.message())   System.out.println("topic: " + msgAndMetadata.topic());   Message message = (Message) msgAndMetadata.message();   ByteBuffer buffer = message.payload();   byte[] bytes = new byte[message.payloadSize()];   buffer.get(bytes);   String tmp = new String(bytes);   System.out.println("message content: " + tmp);   }   }   });   }    } }

　　先启动消费者例子，然后再启动生产者例子，这样会立即看到效果。