Java版Kafka使用及配置解釋
一.Java示例
kafka是吞吐量巨大的一個消息系統，它是用scala寫的，和普通的消息的生產消費還有所不同，寫了個demo程序供大家參考。kafka的安裝請參考官方文檔。
引入Maven庫
首先我們需要新建一個maven項目，然后在pom中引用kafka jar包，引用依賴如下：
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
<version>0.8.0</version>
</dependency>
生產者
我們用的版本是0.8， 下面我們看下生產消息的代碼：
package com.sunteng.clickidc.test;
import java.util.Properties;
import com.sun.tools.javah.Util;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
/**
* Created by silly on 16-8-17.
* Kafka生產者測試
* http://kafka.apache.org/documentation.html#introduction
* http://blog.csdn.net/hmsiwtv/article/details/46960053
*/
public class KafkaProducetest {
private final Producer<String, String> producer;
public final static String TOPIC = "clicki_info_topic";
private KafkaProducetest() {
Properties props = new Properties();
//此處配置的是kafka的端口
props.put("metadata.broker.list", "192.168.11.73:9092");
//配置value的序列化類
props.put("serializer.class", "kafka.serializer.StringEncoder");
//配置key的序列化類
props.put("key.serializer.class", "kafka.serializer.StringEncoder");
//0表示不確認主服務器是否收到消息,馬上返回,低延遲但最弱的持久性,數據可能會丟失
//1表示確認主服務器收到消息后才返回,持久性稍強,可是如果主服務器死掉,從服務器數據尚未同步,數據可能會丟失
//-1表示確認所有服務器都收到數據,完美!
props.put("request.required.acks", "-1");
//異步生產,批量存入緩存后再發到服務器去
props.put("producer.type", "async");
//填充配置,初始化生產者
producer = new Producer<String, String>(new ProducerConfig(props));
}
void produce() {
int messageNo = 1000;
final int COUNT = 2000;
while (messageNo < COUNT) {
String key = String.valueOf(messageNo);
String data = "hello kafka message " + key;
String data1="{\"c\":0,\"i\":16114765323924126,\"n\":\"http://www.abbo.cn/clicki.html\",\"s\":0,\"sid\":0,\"t\":\"info_url\",\"tid\":0,\"unix\":0,\"viewId\":0}";
// 發送消息
// producer.send(new KeyedMessage<String, String>(TOPIC,data1));
// 消息類型key:value
producer.send(new KeyedMessage<String, String>(TOPIC, key, data));
System.out.println(data);
messageNo++;
}
producer.close();//必須關閉
}
public static void main(String[] args) {
new KafkaProducetest().produce();
}
}
消費者
下面是消費端的代碼實現：
package com.sunteng.clickidc.test;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;
/**
* Kafka消費者測試
* Created by silly on 16-8-17.
*/
public class KafkaConsumertest {
private final ConsumerConnector consumer;
private KafkaConsumertest() {
Properties props = new Properties();
//zookeeper 配置
props.put("zookeeper.connect", "192.168.11.73:2181");
//group 代表一個消費組,加入組里面,消息只能被該組的一個消費者消費
//如果所有消費者在一個組內,就是傳統的隊列模式,排隊拿消息
//如果所有的消費者都不在同一個組內,就是發布-訂閱模式,消息廣播給所有組
//如果介於兩者之間,那么廣播的消息在組內也是要排隊的
props.put("group.id", "jd-group");
//zk連接超時
props.put("zookeeper.session.timeout.ms", "4000");//ZooKeeper的最大超時時間，就是心跳的間隔，若是沒有反映，那么認為已經死了，不易過大
props.put("zookeeper.sync.time.ms", "200");//zk follower落后於zk leader的最長時間
props.put("auto.commit.interval.ms", "1000");//往zookeeper上寫offset的頻率
/*
* 此配置參數表示當此groupId下的消費者,在ZK中沒有offset值時(比如新的groupId,或者是zk數據被清空),consumer應該從哪個offset開始消費.
* largest表示接受接收最大的offset(即最新消息),smallest表示最小offset,即從topic的開始位置消費所有消息.
* */
props.put("auto.offset.reset", "smallest"); //消費最老消息,最新為largest
//序列化類
props.put("serializer.class", "kafka.serializer.StringEncoder");
ConsumerConfig config = new ConsumerConfig(props);
consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);
}
void consume() {
// 描述讀取哪個topic，需要幾個線程讀
Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
topicCountMap.put(KafkaProducetest.TOPIC, new Integer(1));
/* 默認消費時的數據是byte[]形式的,可以傳入String編碼器*/
StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());
Map<String, List<KafkaStream<String, String>>> consumerMap =
consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
//消費數據時每個Topic有多個線程在讀,所以取List第一個流
KafkaStream<String, String> stream = consumerMap.get(KafkaProducetest.TOPIC).get(0);
ConsumerIterator<String, String> it = stream.iterator();
while (it.hasNext())
System.out.println(it.next().topic()+":"+it.next().partition()+":"+it.next().offset()+":"+it.next().key()+":"+it.next().message());
}
public static void main(String[] args) {
new KafkaConsumertest().consume();
}
}
注意消費端需要配置成zk的地址，而生產端配置的是kafka的ip和端口。
Kafka為broker,producer和consumer提供了很多的配置參數。了解並理解這些配置參數對於我們使用kafka是非常重要的。本文列出了一些重要的配置參數。
官方的文檔 Configuration 比較老了，程中根據 0.8.2 的代碼也做了修正。
二.Config配置
下表列出了Boker的重要的配置參數， 更多的配置請參考 kafka.server.KafkaConfig
| name | 默認值 | 描述 |
|---|---|---|
| brokerid | none | 每一個boker都有一個唯一的id作為它們的名字。 這就允許boker切換到別的主機/端口上， consumer依然知道 |
| enable.zookeeper | true | 允許注冊到zookeeper |
| log.flush.interval.messages | Long.MaxValue | 在數據被寫入到硬盤和消費者可用前最大累積的消息的數量 |
| log.flush.interval.ms | Long.MaxValue | 在數據被寫入到硬盤前的最大時間 |
| log.flush.scheduler.interval.ms | Long.MaxValue | 檢查數據是否要寫入到硬盤的時間間隔。 |
| log.retention.hours | 168 | 控制一個log保留多長個小時 |
| log.retention.bytes | -1 | 控制log文件最大尺寸 |
| log.cleaner.enable | false | 是否log cleaning |
| log.cleanup.policy | delete | delete還是compat. 其它控制參數還包括log.cleaner.threads，log.cleaner.io.max.bytes.per.second，log.cleaner.dedupe.buffer.size，log.cleaner.io.buffer.size，log.cleaner.io.buffer.load.factor，log.cleaner.backoff.ms，log.cleaner.min.cleanable.ratio，log.cleaner.delete.retention.ms |
| log.dir | /tmp/kafka-logs | 指定log文件的根目錄 |
| log.segment.bytes | 1024*1024 | 單一的log segment文件大小 |
| log.roll.hours | 24 * 7 | 開始一個新的log文件片段的最大時間 |
| message.max.bytes | 1000000 + MessageSet.LogOverhead | 一個socket 請求的最大字節數 |
| num.network.threads | 3 | 處理網絡請求的線程數 |
| num.io.threads | 8 | 處理IO的線程數 |
| background.threads | 10 | 后台線程序 |
| num.partitions | 1 | 默認分區數 |
| socket.send.buffer.bytes | 102400 | socket SO_SNDBUFF參數 |
| socket.receive.buffer.bytes | 102400 | socket SO_RCVBUFF參數 |
| zookeeper.connect | localhost:2182/kafka | 指定zookeeper連接字符串， 格式如hostname:port/chroot。chroot是一個namespace |
| zookeeper.connection.timeout.ms | 6000 | 指定客戶端連接zookeeper的最大超時時間 |
| zookeeper.session.timeout.ms | 6000 | 連接zk的session超時時間 |
| zookeeper.sync.time.ms | 2000 | zk follower落后於zk leader的最長時間 |
三.Consumer配置
下表列出了high-level consumer的重要的配置參數。更多的配置請參考 kafka.consumer.ConsumerConfig
| name | 默認值 | 描述 |
|---|---|---|
| groupid | groupid | 一個字符串用來指示一組consumer所在的組 |
| socket.timeout.ms | 30000 | socket超時時間 |
| socket.buffersize | 64*1024 | socket receive buffer |
| fetch.size | 300 * 1024 | 控制在一個請求中獲取的消息的字節數。 這個參數在0.8.x中由fetch.message.max.bytes,fetch.min.bytes取代 |
| backoff.increment.ms | 1000 | 這個參數避免在沒有新數據的情況下重復頻繁的拉數據。 如果拉到空數據，則多推后這個時間 |
| queued.max.message.chunks | 2 | high level consumer內部緩存拉回來的消息到一個隊列中。 這個值控制這個隊列的大小 |
| autocommit.enable | true | 如果true,consumer定期地往zookeeper寫入每個分區的offset |
| auto.commit.interval.ms | 10000 | 往zookeeper上寫offset的頻率 |
| auto.offset.reset | smallnest | 如果offset出了返回，則 smallest : 自動設置reset到最小的offset. largest : 自動設置offset到最大的offset. anything else : 否則拋出異常. |
| consumer.timeout.ms | -1 | 默認-1,consumer在沒有新消息時無限期的block。如果設置一個正值， 一個超時異常會拋出 |
| rebalance.retries.max | 4 | rebalance時的最大嘗試次數 |
四.Producer配置
下表列出了producer的重要的參數。更多的配置請參考 kafka.producer.ProducerConfig
| name | 默認值 | 描述 |
|---|---|---|
| serializer.class | kafka.serializer.DefaultEncoder | 必須實現kafka.serializer.Encoder 接口，將T類型的對象encode成kafka message |
| key.serializer.class | serializer.class | key對象的serializer類 |
| partitioner.class | kafka.producer.DefaultPartitioner | 必須實現kafka.producer.Partitioner ，根據Key提供一個分區策略 |
| producer.type | sync | 指定消息發送是同步還是異步。異步asyc成批發送用kafka.producer.AyncProducer， 同步sync用kafka.producer.SyncProducer |
| metadata.broker.list | boker list | 使用這個參數傳入boker和分區的靜態信息，如host1:port1,host2:port2, 這個可以是全部boker的一部分 |
| compression.codec | NoCompressionCodec | 消息壓縮，默認不壓縮 |
| compressed.topics | null | 在設置了壓縮的情況下，可以指定特定的topic壓縮，為指定則全部壓縮 |
| message.send.max.retries | 3 | 消息發送最大嘗試次數 |
| retry.backoff.ms | 300 | 每次嘗試增加的額外的間隔時間 |
| topic.metadata.refresh.interval.ms | 600000 | 定期的獲取元數據的時間。當分區丟失，leader不可用時producer也會主動獲取元數據，如果為0，則每次發送完消息就獲取元數據，不推薦。如果為負值，則只有在失敗的情況下獲取元數據。 |
| queue.buffering.max.ms | 5000 | 在producer queue的緩存的數據最大時間，僅僅for asyc |
| queue.buffering.max.message | 10000 | producer 緩存的消息的最大數量，僅僅for asyc |
| queue.enqueue.timeout.ms | -1 | 0當queue滿時丟掉，負值是queue滿時block,正值是queue滿時block相應的時間，僅僅for asyc |
| batch.num.messages | 200 | 一批消息的數量，僅僅for asyc |