kafka java示例

時間 2019-12-07

標籤 kafka java 示例欄目 Kafka 简体版

原文原文鏈接

kafka java示例

分類： hadoop 2014-02-13 14:34 9616人閱讀評論(1) 收藏舉報

我使用的kafka版本是：0.7.2

jdk版本是：1.6.0_20 html

http://kafka.apache.org/07/quickstart.html官方給的示例並非很完整，如下代碼是通過我補充的而且編譯後能運行的。 java

Producer Code apache

[java] view plain copy

import java.util.*;
import kafka.message.Message;
import kafka.producer.ProducerConfig;
import kafka.javaapi.producer.Producer;
import kafka.javaapi.producer.ProducerData;
public class ProducerSample {
public static void main(String[] args) {
ProducerSample ps = new ProducerSample();
Properties props = new Properties();
props.put("zk.connect", "127.0.0.1:2181");
props.put("serializer.class", "kafka.serializer.StringEncoder");
ProducerConfig config = new ProducerConfig(props);
Producer<String, String> producer = new Producer<String, String>(config);
ProducerData<String, String> data = new ProducerData<String, String>("test-topic", "test-message2");
producer.send(data);
producer.close();
}
}

Consumer Code

[java] view plain copy

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.Message;
import kafka.message.MessageAndMetadata;
public class ConsumerSample {
public static void main(String[] args) {
// specify some consumer properties
Properties props = new Properties();
props.put("zk.connect", "localhost:2181");
props.put("zk.connectiontimeout.ms", "1000000");
props.put("groupid", "test_group");
// 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
HashMap<String, Integer> map = new HashMap<String, Integer>();
map.put("test-topic", 4);
Map<String, List<KafkaStream<Message>>> topicMessageStreams =
consumerConnector.createMessageStreams(map);
List<KafkaStream<Message>> streams = topicMessageStreams.get("test-topic");
// create list of 4 threads to consume from each of the partitions
ExecutorService executor = Executors.newFixedThreadPool(4);
// consume the messages in the threads
for (final KafkaStream<Message> stream : streams) {
executor.submit(new Runnable() {
public void run() {
for (MessageAndMetadata msgAndMetadata : stream) {
// process message (msgAndMetadata.message())
System.out.println("topic: " + msgAndMetadata.topic());
Message message = (Message) msgAndMetadata.message();
ByteBuffer buffer = message.payload();
<span style="white-space:pre"> </span>byte[] bytes = new byte[message.payloadSize()];
buffer.get(bytes);
String tmp = new String(bytes);
System.out.println("message content: " + tmp);
}
}
});
}
}
}

分別啓動zookeeper,kafka server以後，依次運行Producer,Consumer的代碼

運行ProducerSample：編程

運行ConsumerSample: api

因爲本人不熟悉java的多線程，將官方給的Consumer Code作點小改動，以下所示：多線程

[java] view plain copy

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.Message;
import kafka.message.MessageAndMetadata;
public class ConsumerSample2 {
public static void main(String[] args) {
// specify some consumer properties
Properties props = new Properties();
props.put("zk.connect", "localhost:2181");
props.put("zk.connectiontimeout.ms", "1000000");
props.put("groupid", "test_group");
// Create the connection to the cluster
ConsumerConfig consumerConfig = new ConsumerConfig(props);
ConsumerConnector consumerConnector = Consumer.createJavaConsumerConnector(consumerConfig);
HashMap<String, Integer> map = new HashMap<String, Integer>();
map.put("test-topic", 1);
Map<String, List<KafkaStream<Message>>> topicMessageStreams =
consumerConnector.createMessageStreams(map);
List<KafkaStream<Message>> streams = topicMessageStreams.get("test-topic");
<strong>for (final KafkaStream<Message> stream : streams) {
for (MessageAndMetadata msgAndMetadata : stream) {
// process message (msgAndMetadata.message())
System.out.println("topic: " + msgAndMetadata.topic());
Message message = (Message) msgAndMetadata.message();
ByteBuffer buffer = message.payload();
byte[] bytes = new byte[message.payloadSize()];
buffer.get(bytes);
String tmp = new String(bytes);
System.out.println("message content: " + tmp);
}
}</strong>
}
}

我在Producer端又發送了一條「test-message2」的消息，Consumer收到了兩條消息，以下所示：

kafka做爲分佈式日誌收集或系統監控服務，咱們有必要在合適的場合使用它。kafka的部署包括zookeeper環境/kafka環境，同時還須要進行一些配置操做.接下來介紹如何使用kafka. socket

咱們使用3個zookeeper實例構建zk集羣，使用2個kafka broker構建kafka集羣. async

其中kafka爲0.8V，zookeeper爲3.4.5V maven

一.Zookeeper集羣構建 分佈式

咱們有3個zk實例，分別爲zk-0,zk-1,zk-2;若是你僅僅是測試使用，可使用1個zk實例.

1) zk-0

調整配置文件：

Php代碼

clientPort=2181
server.0=127.0.0.1:2888:3888
server.1=127.0.0.1:2889:3889
server.2=127.0.0.1:2890:3890
##只須要修改上述配置，其餘配置保留默認值

啓動zookeeper

Java代碼

./zkServer.sh start

2) zk-1

調整配置文件(其餘配置和zk-0一隻)：

Php代碼

clientPort=2182
##只須要修改上述配置，其餘配置保留默認值

啓動zookeeper

Java代碼

./zkServer.sh start

3) zk-2

調整配置文件(其餘配置和zk-0一隻)：

Php代碼

clientPort=2183
##只須要修改上述配置，其餘配置保留默認值

啓動zookeeper

Java代碼

./zkServer.sh start

二. Kafka集羣構建

由於Broker配置文件涉及到zookeeper的相關約定，所以咱們先展現broker配置文件.咱們使用2個kafka broker來構建這個集羣環境，分別爲kafka-0,kafka-1.

1) kafka-0

在config目錄下修改配置文件爲：

Java代碼

broker.id=0
port=9092
num.network.threads=2
num.io.threads=2
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
log.dir=./logs
num.partitions=2
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168
#log.retention.bytes=1073741824
log.segment.bytes=536870912
##replication機制,讓每一個topic的partitions在kafka-cluster中備份2個
##用來提升cluster的容錯能力..
default.replication.factor=1
log.cleanup.interval.mins=10
zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183
zookeeper.connection.timeout.ms=1000000

由於kafka用scala語言編寫，所以運行kafka須要首先準備scala相關環境。

Java代碼

> cd kafka-0
> ./sbt update
> ./sbt package
> ./sbt assembly-package-dependency

其中最後一條指令執行有可能出現異常，暫且無論。啓動kafka broker：

Java代碼

> JMS_PORT=9997 bin/kafka-server-start.sh config/server.properties &

由於zookeeper環境已經正常運行了，咱們無需經過kafka來掛載啓動zookeeper.若是你的一臺機器上部署了多個kafka broker，你須要聲明JMS_PORT.

2) kafka-1

Java代碼

broker.id=1
port=9093
##其餘配置和kafka-0保持一致

而後和kafka-0同樣執行打包命令，而後啓動此broker.

Java代碼

> JMS_PORT=9998 bin/kafka-server-start.sh config/server.properties &

仍然能夠經過以下指令查看topic的"partition"/"replicas"的分佈和存活狀況.

Java代碼

> bin/kafka-list-topic.sh --zookeeper localhost:2181
topic: my-replicated-topic partition: 0 leader: 2 replicas: 1,2,0 isr: 2
topic: test partition: 0 leader: 0 replicas: 0 isr: 0

到目前爲止環境已經OK了,那咱們就開始展現編程實例吧。[配置參數詳解]

三.項目準備

項目基於maven構建，不得不說kafka java客戶端實在是太糟糕了；構建環境會遇到不少麻煩。建議參考以下pom.xml;其中各個依賴包必須版本協調一致。若是kafka client的版本和kafka server的版本不一致,將會有不少異常,好比"broker id not exists"等;由於kafka從0.7升級到0.8以後(正名爲2.8.0),client與server通信的protocol已經改變.

Java代碼

<dependencies>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.14</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.8.2</artifactId>
<version>0.8.0</version>
<exclusions>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.8.2</version>
</dependency>
<dependency>
<groupId>com.yammer.metrics</groupId>
<artifactId>metrics-core</artifactId>
<version>2.2.0</version>
</dependency>
<dependency>
<groupId>com.101tec</groupId>
<artifactId>zkclient</artifactId>
<version>0.3</version>
</dependency>
</dependencies>

四.Producer端代碼

1) producer.properties文件：此文件放在/resources目錄下

Java代碼

#partitioner.class=
##broker列表能夠爲kafka server的子集,由於producer須要從broker中獲取metadata
##儘管每一個broker均可以提供metadata,此處仍是建議,將全部broker都列舉出來
metadata.broker.list=127.0.0.1:9092,127.0.0.1:9093
##,127.0.0.1:9093
##同步,建議爲async
producer.type=sync
compression.codec=0
serializer.class=kafka.serializer.StringEncoder
##在producer.type=async時有效
#batch.num.messages=100

2) LogProducer.java代碼樣例

Java代碼

package com.test.kafka;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
public class LogProducer {
private Producer<String,String> inner;
public LogProducer() throws Exception{
Properties properties = new Properties();
properties.load(ClassLoader.getSystemResourceAsStream("producer.properties"));
ProducerConfig config = new ProducerConfig(properties);
inner = new Producer<String, String>(config);
}
public void send(String topicName,String message) {
if(topicName == null || message == null){
return;
}
KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,message);//若是具備多個partitions,請使用new KeyedMessage(String topicName,K key,V value).
inner.send(km);
}
public void send(String topicName,Collection<String> messages) {
if(topicName == null || messages == null){
return;
}
if(messages.isEmpty()){
return;
}
List<KeyedMessage<String, String>> kms = new ArrayList<KeyedMessage<String, String>>();
for(String entry : messages){
KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,entry);
kms.add(km);
}
inner.send(kms);
}
public void close(){
inner.close();
}
/**
* @param args
*/
public static void main(String[] args) {
LogProducer producer = null;
try{
producer = new LogProducer();
int i=0;
while(true){
producer.send("test-topic", "this is a sample" + i);
i++;
Thread.sleep(2000);
}
}catch(Exception e){
e.printStackTrace();
}finally{
if(producer != null){
producer.close();
}
}
}
}

五.Consumer端

1) consumer.properties:文件位於/resources目錄下

Java代碼

zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183
##,127.0.0.1:2182,127.0.0.1:2183
# timeout in ms for connecting to zookeeper
zookeeper.connectiontimeout.ms=1000000
#consumer group id
group.id=test-group
#consumer timeout
#consumer.timeout.ms=5000
auto.commit.enable=true
auto.commit.interval.ms=60000

2) LogConsumer.java代碼樣例

Java代碼

package com.test.kafka;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
public class LogConsumer {
private ConsumerConfig config;
private String topic;
private int partitionsNum;
private MessageExecutor executor;
private ConsumerConnector connector;
private ExecutorService threadPool;
public LogConsumer(String topic,int partitionsNum,MessageExecutor executor) throws Exception{
Properties properties = new Properties();
properties.load(ClassLoader.getSystemResourceAsStream("consumer.properties"));
config = new ConsumerConfig(properties);
this.topic = topic;
this.partitionsNum = partitionsNum;
this.executor = executor;
}
public void start() throws Exception{
connector = Consumer.createJavaConsumerConnector(config);
Map<String,Integer> topics = new HashMap<String,Integer>();
topics.put(topic, partitionsNum);
Map<String, List<KafkaStream<byte[], byte[]>>> streams = connector.createMessageStreams(topics);
List<KafkaStream<byte[], byte[]>> partitions = streams.get(topic);
threadPool = Executors.newFixedThreadPool(partitionsNum);
for(KafkaStream<byte[], byte[]> partition : partitions){
threadPool.execute(new MessageRunner(partition));
}
}
public void close(){
try{
threadPool.shutdownNow();
}catch(Exception e){
//
}finally{
connector.shutdown();
}
}
class MessageRunner implements Runnable{
private KafkaStream<byte[], byte[]> partition;
MessageRunner(KafkaStream<byte[], byte[]> partition) {
this.partition = partition;
}
public void run(){
ConsumerIterator<byte[], byte[]> it = partition.iterator();
while(it.hasNext()){
//connector.commitOffsets();手動提交offset,當autocommit.enable=false時使用
MessageAndMetadata<byte[],byte[]> item = it.next();
System.out.println("partiton:" + item.partition());
System.out.println("offset:" + item.offset());
executor.execute(new String(item.message()));//UTF-8,注意異常
}
}
}
interface MessageExecutor {
public void execute(String message);
}
/**
* @param args
*/
public static void main(String[] args) {
LogConsumer consumer = null;
try{
MessageExecutor executor = new MessageExecutor() {
public void execute(String message) {
System.out.println(message);
}
};
consumer = new LogConsumer("test-topic", 2, executor);
consumer.start();
}catch(Exception e){
e.printStackTrace();
}finally{
// if(consumer != null){
// consumer.close();
// }
}
}
}