實驗環境：

kafka_2.11-0.9.0.1.tgz

zookeeper-3.4.6.tar.gz

樣例代碼：

git clone https://github.com/downgoon/hello-world

git checkout kafka9

git checkout kafka-c1-message

kafka 0.9 java producer and consumer demo

客戶端

kafka 客戶端是官方的jar包： org.apache.kafka:kafka-clients:0.9.0.1，maven配置如下：

org.apache.kafka

kafka-clients

0.9.0.1

Producer

import java.util.Properties;

import java.util.concurrent.Future;

import org.apache.kafka.clients.producer.Callback;

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

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

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

import org.apache.kafka.clients.producer.RecordMetadata;

public class ProducerDemo {

public static void main(String[] args) throws Exception {

String topic = "test";

// producer properties setting

Properties props = new Properties();

props.put("bootstrap.servers",

"10.213.42.230:10062,10.213.42.234:10062,10.213.44.188:10062,10.213.44.189:10062");

props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

props.put("acks", "all");

props.put("retries", 1);

// create an instance of producer

Producer producer = new KafkaProducer(props);

// asynchronously & future

// Send the given record asynchronously and return a future which will

// eventually contain the response information

for (int i = 0; i < 100; i++) {

Future future = producer.send(new ProducerRecord(topic, "Hello"));

RecordMetadata recMeta = future.get();

System.out.println(

"record meta " + i + ": offset=" + recMeta.offset() + ", partition=" + recMeta.partition());

}

// asynchronously & callback

producer.send(new ProducerRecord(topic, "World"), new Callback() {

@Override

public void onCompletion(RecordMetadata metadata, Exception e) {

if (e != null) {

e.printStackTrace();

} else {

System.out.println("metadata: offset=" + metadata.offset() + ", partition=" + metadata.partition());

}

}

});

// Flush any accumulated records from the producer.

// Blocks until all sends are complete.

producer.flush();

producer.close();

}

}

幾點說明

broker 集群地址

Producer 需要知道 kafka broker 集群中的一個或多個broker 地址(并不需要全部，當然寫得越多，越有利于高可用)。但是現在的Producer已經不需要知道ZK地址了，因為kafka一直在致力于把實現隱藏起來。

異步發送

為了提高發送端的效率，都流行異步。但是異步如何保證消息不丟呢 ？

Consumer

import java.util.Arrays;

import java.util.Properties;

import org.apache.kafka.clients.consumer.Consumer;

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

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

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

public class ConsumerDemo {

public static void main(String[] args) throws Exception {

String topic = "test";

Properties props = new Properties();

props.put("bootstrap.servers", "10.213.42.230:10062,10.213.42.234:10062,10.213.44.188:10062,10.213.44.189:10062");

props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

props.setProperty("group.id", "jconsumer-passport"); // 定義group

props.setProperty("enable.auto.commit", "true");

props.setProperty("auto.offset.reset", "earliest");

Consumer consumer = new KafkaConsumer(props);

// consumer.subscribe(Lists.newArrayList(topic));

consumer.subscribe(Arrays.asList(new String[] {topic})); // 可以同時消費多個 Topic

for (int i = 0; i < 2; i++) {

ConsumerRecords records = consumer.poll(1000); // 拉取

System.out.println("polled: "+ records.count()); // 一次可拉取成千上萬條

for (ConsumerRecord record : records) {

System.out.println("\t\trecord: " +record);

//consumer.seekToBeginning(new TopicPartition(record.topic(), record.partition()));

}

}

consumer.close();

}

}

幾點說明

broker 集群地址

從0.9開始，Consumer API被kafka官方重寫，Consumer也不需要依賴ZK了，以前用ZK是因為：Rebalance和Offset管理，現在是kafka用一個叫"__consumer_offsets"的Topic來管理。

groupid 與 topic 是多對多的關系

一個groupid 可以消費多個topic的內容，一個topic也可以被多個groupid消費。 我們可以把groupid理解為對數據的一個應用場景。

groupid與consumerid的關系是多對1的關系。

監控畫面

參考資料

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的kafka 0.9 java开发_kafka 0.9 java producer and consumer demo的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。