Kafka 学习笔记
内容大部分引用自Info – Apache Kafka：下一代分布式消息系统
原文作者Abhishek Sharma的项目Kafka-Message-Server
介绍
Kafka是使用scala语言开发，类似于RabbitMQ的分布式消息系统。
Kafka是分布式的，它通过可以多个broker组成一个集群。
Kafka依赖于Zookeeper。
概念
Topic(话题) 特定类型的消息流。消息是字节的有效负载(Payload)，话题是消息的分类或种子(Feed)名。
Producer(生产者) 能够发布消息到话题的任何对象。
Broker(代理) 或称Kafka集群。用于保存消息的服务器。
Consumer(消费者) 可以订阅一个或多个话题，并从Broker拉取数据，从而消费这些已发布的消息。
Kafka生产者, 消费者 & 代理
架构
生产者使用自己的序列化方法对消息内容进行编码。然后向broker发起消息。为了提高效率，一个发布请求中可以包含一组消息。
消费者订阅话题，并为话题创建一个或多个消息流。发布到该话题的消息被均衡的分发到这些流中。
每个消息流为不断产生的消息提供了迭代接口。
消费者迭代流中每一条消息，并处理消息的有效负载。
迭代器不会停止。如果当前没有消息，迭代器将阻塞直至有新的消息发布到该话题。
Kafka支持点到点分发模型(Proint-to-point delivery model)，即多个消费者共同消费队列中某个消息的单个副本；也支持发布-订阅模型(Publish-subscribe model)，即多个消费者接收自己的消息副本。
kafka-proc-topic-part-con.png
存储
kafka的存储，话题的每个分区对应一个逻辑日志。物理上，一个日志为相同大小的一段分组文件。
每次生产者发布消息到一个分区，代理就将消息追加到最后一个段文件中。
当发布的消息数量达到设定值或经过一段时间后，段文件真正写入磁盘中。
写入完成后，消息公开给消费者。
与传统的消息不同，kafka系统中存储的消息没有明确的id，而是通过日志的逻辑偏移量来公开。相比其他方式，这种处理更为高效。
消费者始终从特殊分区顺序的获取消息。
代理
不同于其他消息系统，kafka代理是无状态的，即消费者必须维护已消费的状态消息，而代理完全不管。
这种设计的创新在于:
代理以一个基于时间的SLA应用于保留策略。当消息在代理中超过一定时间后，将会被自动删除。
消费者可以故意倒回到老的偏移量再次消费数据。虽然这违法了队列的常见约定，但常见于许多业务中。
与zookeeper的关系
kafka使用ZooKeeper用于管理、协调代理。每个Kafka代理通过Zookeeper协调其他Kafka代理。
当Kafka系统中新增了代理或某个代理失效时，Zookeeper服务将通知生产者和消费者。
生产者与消费者据此开始与其他代理协调工作。
安装
wget http://mirrors.cnnic.cn/apache/kafka/0.8.2.1/kafka_2.11-0.8.2.1.tgz
tar -xzvf kafka_2.11-0.8.2.1.tgz
配置
config/server.properties
# 如果配置多个kafka节点，id需设置为不同的值
broker.id=1
# !!务必将host.name配置为ip地址。
# 在java代码里连接kafka时，服务端会把host.name的值传给zookeeper
# 如果使用默认配置的localhost，会出现连接失败的异常
host.name=192.168.1.1
port=9092
log.dir=./logs
# 如果有多个zookeeper服务，用,号隔开即可。
# zookeeper使用默认配置的2181端口
zookeeper.connect=192.168.1.1:2181
zookeeper.connection.timeout.ms=6000
启动
bin/
# 先启动zookeeper服务
./zookeeper-server-start.sh ../config/zookeeper.properties &
# 再启动kafka服务
./kafka-server-start.sh ../config/server.properties
测试
bin/
创建一个Topic
./kafka-topics.sh --create --zookeeper 192.168.1.1:2181 --replication-factor 2 --partitions 1 --topic test
查看Topic信息
./kafka-topics.sh --describe --zookeeper 192.168.1.1:2181 --topic test
启动生产者(producer)生产该Topic的消息
./kafka-console-producer.sh --broker-list 192.168.1.1:9092 --topic test
启动消费者(consumer)消费该Topic的消息
./kafka-console-consumer.sh --zookeeper 192.168.1.1:2181 --from-beginning -topic test
Java代码
Producer 测试
import java.util.Date;
import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;
public class KafkaProducer {
public static void testProducer() {
Properties props = new Properties();
props.put("metadata.broker.list", "192.168.1.1:9092");
props.put("serializer.class", StringEncoder.class.getName());
//props.put("partitioner.class", );
props.put("request.required.arks", "1");
ProducerConfig config = new ProducerConfig(props);
Producer<String, String> producer = new Producer<String, String>(config);
String msg = new Date() + " - hello world : 测试 " ;
KeyedMessage<String, String> data = new KeyedMessage<String, String>("test", msg);
producer.send(data);
producer.close();
System.out.println("--> producer sended： " + msg);
}
public static void main(String[] args) {
testProducer();
}
}
Consumer 测试
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.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
public class KafkaConsumer {
private final ConsumerConnector consumer;
private final String topic;
private ExecutorService executor;
public KafkaConsumer(String a_zookeeper, String a_groupId, String a_topic) {
this.consumer = kafka.consumer.Consumer
.createJavaConsumerConnector(createConsumerConfig(a_zookeeper,
a_groupId));
this.topic = a_topic;
}
private static ConsumerConfig createConsumerConfig(String a_zookeeper,
String a_groupId) {
Properties props = new Properties();
props.put("zookeeper.connect", a_zookeeper);
props.put("group.id", a_groupId);
props.put("zookeeper.session.timeout.ms", "1000");
props.put("zookeeper.sync.time.ms", "1000");
props.put("auto.commit.interval.ms", "1000");
props.put("auto.offset.reset", "smallest");
return new ConsumerConfig(props);
}
public void shutdown() {
if (consumer != null)
consumer.shutdown();
if (executor != null)
executor.shutdown();
}
public void run(int a_numThreads) {
Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
topicCountMap.put(topic, new Integer(a_numThreads));
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer
.createMessageStreams(topicCountMap);
List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
System.out.println("streams.size = " + streams.size());
// now launch all the threads
//
executor = Executors.newFixedThreadPool(a_numThreads);
// now create an object to consume the messages
//
int threadNumber = 0;
for (final KafkaStream stream : streams) {
executor.submit(new ConsumerTest(stream, threadNumber));
threadNumber++;
}
}
public static void main(String[] args) {
String zooKeeper = "192.168.212.100:2181";
String groupId = "group1";
String topic = "test";
int threads = 3;
KafkaConsumer example = new KafkaConsumer(zooKeeper, groupId, topic);
example.run(threads);
}
public class ConsumerTest implements Runnable {
private KafkaStream m_stream;
private int m_threadNumber;
public ConsumerTest(KafkaStream a_stream, int a_threadNumber) {
m_threadNumber = a_threadNumber;
m_stream = a_stream;
}
public void run() {
System.out.println("calling ConsumerTest.run()");
ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
while (it.hasNext()) {
System.out.println("--> consumer Thread " + m_threadNumber + ": "
+ new String(it.next().message()));
}
System.out.println("Shutting down Thread: " + m_threadNumber);
}
}
}
