kafka入门示例(Java)
上篇说到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哥的帮助!!!
下一篇: 历史上真实的虞姬是什么人?她的结局如何?