Home | 简体中文 | 繁体中文 | 杂文 | 打赏(Donations) | 云栖社区 | OSChina 博客 | Facebook | Linkedin | 知乎专栏 | Github | Search | About

26.4. 入门例子

26.4.1. 订阅例子

		
package cn.netkiller.kafka.test;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class KafkaConsumerExample {

	private static KafkaConsumer<String, String> consumer;

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		Properties props = new Properties();
		props.put("bootstrap.servers", "kafka.netkiller.cn:9092");
		props.put("group.id", "test");
		props.put("enable.auto.commit", "true");
		props.put("auto.commit.interval.ms", "1000");
		props.put("session.timeout.ms", "30000");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		consumer = new KafkaConsumer<String, String>(props);
		consumer.subscribe(Arrays.asList("test"));
		while (true) {
			ConsumerRecords<String, String> records = consumer.poll(100);
			for (ConsumerRecord<String, String> record : records)
				System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
		}
	}

}
		
			

测试方法

		
root@netkiller ~ % /srv/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>Helloworld
		
			

下面详细讲解上面的程序。首先我们通过Properties文件来配置消费属性。

		
		Properties props = new Properties();
		props.put("bootstrap.servers", "kafka.netkiller.cn:9092");
		props.put("group.id", "test");
		props.put("enable.auto.commit", "true");
		props.put("auto.commit.interval.ms", "1000");
		props.put("session.timeout.ms", "30000");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		
			

然后订阅TOPIC,从那个TOPIC 读取消息,Kafka 可以同时订阅多个 TOPIC。下面的例子中,同时订阅了foo和bar两个topic:

		
consumer.subscribe(Arrays.asList("foo", "bar"));
		
			

取出消息(消费消息),通过循环调用poll方法,从队列中取出消息。

		
		while (true) {
			ConsumerRecords<String, String> records = consumer.poll(100);
			for (ConsumerRecord<String, String> record : records)
				System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
		}		
		
			

poll方法里面的参数是等待消息的时间(Long类型),如果队列里面有消息,会立马返回,如果没有,会等待指定的时间然后返回。

		

		
			

26.4.2. 发布例子

		
package cn.netkiller.kafka.test;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "kafka.netkiller.cn:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        for(int i = 0; i < 100; i++)
            producer.send(new ProducerRecord<>("test", Integer.toString(i), Integer.toString(i)));

        producer.close();
    }
}