欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

kafka入门示例(Java)

程序员文章站 2022-05-11 19:28:00
...

上篇说到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哥的帮助!!!