kafka入门示例(Java)

renran421

　　上篇说到kafka在window环境下的搭建之后，这篇我们就开始尝试写一个简单的producer和consumer来测试了
　　依次启动bin/windows下的zookeeper-start.bat和kafak-start.bat(这2个.bat是我自己为了方便启动而写的)。下面就开始测试了：kafak jar包版本：kafak_2.9.2-0.8.1.jar
　　Produce端：

import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
public class KafkaProducerTest {
String topic = "test";
public static void main(String[] args) {
Properties props = new Properties();
//props.put("zookeeper.connect", "10.16.0.200:2181");
props.put("serializer.class", "kafka.serializer.StringEncoder");
props.put("producer.type", "async");//默认是sync
props.put("compression.codec", "1");
props.put("metadata.broker.list", "127.0.0.1:9092");
ProducerConfig config = new ProducerConfig(props);
Producer<String, Object> producer = new Producer<String, Object>(config);
KeyedMessage<String, Object> message =
new KeyedMessage<String, Object>("test", "hello world");
producer.send(message);
}
　　其中ProducerConfig是Producer端的属性配置类，更多属性可参见kakfa的jar包反编译后的kafka.producer.ProducerConfig类，该类的属性定义里面有很多Producer端的必须或可选属性。上述代码中的属性不过
　　是必须要配置的属性而已。
　　注意kafak的jar包里会有2个Producer类，我们必须引用的是kafka.javaapi.producer包下面的才行。
　　KeyedMessage即表明消息。它的构造函数有以下几个：

public KeyedMessage(String topic, K key, Object partKey, V message) { Product.class.$init$(this);
if (topic == null)
throw new IllegalArgumentException("Topic cannot be null.");  }
public KeyedMessage(String topic, V message) {
this(topic, null, null, message);
}
public KeyedMessage(String topic, K key, V message) { this(topic, key, key, message); }
　　第一个参数是消息的topic，第二个参数是消息的内容
　　metadata.broker.list为Producer端配置的用于指定元数据节点的属性，节点与节点之间用,隔开。关于集群式节点的配置这里不再祥述。
　　Consumer端：

public class KafkaConsumerTest {  
public static void main(String[] args) {  
// specify some consumer properties  
Properties props = new Properties();  
props.put("zookeeper.connect", "127.0.0.1:2181");  
props.put("zookeeper.connectiontimeout.ms", "1000000");  
props.put("group.id", "test_group");
props.put("zookeeper.session.timeout.ms", "40000");
props.put("zookeeper.sync.time.ms", "200");
props.put("auto.commit.interval.ms", "1000");
// Create the connection to the cluster  
ConsumerConfig consumerConfig = new ConsumerConfig(props);  
ConsumerConnector consumerConnector = Consumer.createJavaConsumerConnector(consumerConfig);  
// create 4 partitions of the stream for topic “test-topic”, to allow 4 threads to consume  
Map<String, Integer> topicCountMap = new HashMap<String, Integer>();  
topicCountMap.put("test", new Integer(4));  
//key--topic  
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap =
consumerConnector.createMessageStreams(topicCountMap);
KafkaStream<byte[], byte[]> stream = consumerMap.get("test").get(0);
ConsumerIterator<byte[], byte[]> it = stream.iterator();
StringBuffer sb = new StringBuffer();
while(it.hasNext()) {
try {
String msg = new String(it.next().message(), "utf-8").trim();
System.out.println("receive:" + msg);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
}
　　while(it.hasNext())循环里可以进一步改进成这样：

while (it.hasNext()) {
String msg= "";
byte[] packs= it.next().message();
InputStream is = new ByteArrayInputStream(bPack);
//如果阻塞了，也就是该流不可以被读取，那么ready()就返回false.
BufferedReader ioBuffRead = new BufferedReader(new InputStreamReader(is, charset));
while(ioBuffRead.ready()){
msg= ioBuffRead.readLine();
//接下来的代码略。。
}
}
　　同样注意kafak的jar包里会有2个Consumer类，我们必须引用的是kafka.javaapi.consumer包下面的才行；而关于Consumer端具体的属性配置的也可在ConsumerConfig类的属性定义下面找到。
　　Producer端的代码比较简单。我们需要好好理解的是Consumer端的代码——
　　1)ConsumerConnector类是消费端根据消费配置的连接类
　　2)topicCountMap为·topic的map，key为topic，value为该topic消息流的分区数
　　3)consumerMap为消费端的map，key为topic，value为该topic对应的消息队列，表现为一个List集合，头元素即为该队列的头部元素，就是为什么consumerMap.get("test").get(0);
　　的原因，该集合的大小是动态的，因为队列中有元素不停地进出；
　　4）ConsumerIterator为该topic消息流的迭代器，用于迭代从而取出里面的消息。
　　运行之后，会发现在Consumer端打印：hello world。你也可以在控制台运行命令查看该topic的消息。
　　至此，kafka的一个入门级生产-消费测试就完毕了。
　　在成功启动windows下的kafka或Linux版本的kafka之后，运行该简单的demo示例，你很可能会出现如下异常：
　　java.nio.channels.UnresolvedAddressException
　　at sun.nio.ch.Net.checkAddress(Net.java:29)
　　at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:512)
　　at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57)
　　at kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
　　at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:156)
　　at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68)
　　解决这个异常也是比较诡异，多亏F哥的帮助。这是由于你缺少必要的地址的dns解析，修改C盘下的hosts文件，加上：
　　服务器ip地址 服务器域名
　　即可。其中很有可能你访问windows本地的kafka也会出现该异常，则加上127.0.0.1 localhost即可
　　注：1个topic对应1个offset(消费内容的位置)； 1个group.id可对应多个topic
　　最后再次衷心的感谢F哥的帮助！！！