Kafka笔记--参数说明及Demo

参考资料：http://blog.csdn.net/honglei915/article/details/37563647
参数说明：http://ju.outofmemory.cn/entry/119243
参数说明/Demo：http://www.aboutyun.com/thread-9906-1-1.html

Kafka+Spark:
http://shiyanjun.cn/archives/1097.html
http://ju.outofmemory.cn/entry/84636

1. Kafka启动：
　　1. 先启动所有节点的zookeeper : 进入ZOOKEEPER_HOME/bin 执行./zkServer.sh start
　　2. 启动所有节点的kafka：进入 KAFKA_HOME/bin 执行 ./kafka-server-start.sh config/server.properties &

2. 参数说明

2.0 boker参数说明 (配置文件位于config/server.properties)

name	默认值	描述
broker.id	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	110241024*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的最长时间

2.1 producer参数说明(配置文件位于config/producer.properties或者在程序内定义)

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()){
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();
//            }
}

}

}

View Code