消息系统Kafka介绍
本文链接: http://www.54chen.com/java-ee/linkedin-kafka-usage.html
1、 概述
Kafka是Linkedin于2010年12月份开源的消息系统，它主要用于处理活跃的流式数据。活跃的流式数据在web网站应用中非常常见，这些数据包括网站的pv、用户访问了什么内容，搜索了什么内容等。 这些数据通常以日志的形式记录下来，然后每隔一段时间进行一次统计处理。
传统的日志分析系统提供了一种离线处理日志信息的可扩展方案，但若要进行实时处理，通常会有较大延迟。而现有的消（队列）系统能够很好的处理实时或者近似实时的应用，但未处理的数据通常不会写到磁盘上，这对于Hadoop之类（一小时或者一天只处理一部分数据）的离线应用而言，可能存在问题。Kafka正是为了解决以上问题而设计的，它能够很好地离线和在线应用。
2、 设计目标
（1）数据在磁盘上存取代价为O(1)。一般数据在磁盘上是使用BTree存储的，存取代价为O（lgn）。
（2）高吞吐率。即使在普通的节点上每秒钟也能处理成百上千的message。
（3）显式分布式，即所有的producer、broker和consumer都会有多个，均为分布式的。
（4）支持数据并行加载到Hadoop中。
3、 KafKa部署结构
kafka是显式分布式架构，producer、broker（Kafka）和consumer都可以有多个。Kafka的作用类似于缓存，即活跃的数据和离线处理系统之间的缓存。几个基本概念：
（1）message（消息）是通信的基本单位，每个producer可以向一个topic（主题）发布一些消息。如果consumer订阅了这个主题，那么新发布的消息就会广播给这些consumer。
（2）Kafka是显式分布式的，多个producer、consumer和broker可以运行在一个大的集群上，作为一个逻辑整体对外提供服务。对于consumer，多个consumer可以组成一个group，这个message只能传输给某个group中的某一个consumer.
4、 KafKa关键技术点
（1） zero-copy
在Kafka上，有两个原因可能导致低效：1）太多的网络请求 2）过多的字节拷贝。为了提高效率，Kafka把message分成一组一组的，每次请求会把一组message发给相应的consumer。 此外， 为了减少字节拷贝，采用了sendfile系统调用。为了理解sendfile原理，先说一下传统的利用socket发送文件要进行拷贝：
Sendfile系统调用：
（2） Exactly once message transfer
怎样记录每个consumer处理的信息的状态？在Kafka中仅保存了每个consumer已经处理数据的offset。这样有两个好处：1）保存的数据量少 2）当consumer出错时，重新启动consumer处理数据时，只需从最近的offset开始处理数据即可。
（3）Push/pull
Producer 向Kafka（push）推数据，consumer 从kafka 拉（pull）数据。
（4）负载均衡和容错
Producer和broker之间没有负载均衡机制。
broker和consumer之间利用zookeeper进行负载均衡。所有broker和consumer都会在zookeeper中进行注册，且zookeeper会保存他们的一些元数据信息。如果某个broker和consumer发生了变化，所有其他的broker和consumer都会得到通知。
【参考资料】
【1】Kafka主页：http://sna-projects.com/kafka/design.php
【2】Zero-copy原理：https://www.ibm.com/developerworks/linux/library/j-zerocopy/
【3】Kafka与Hadoop：http://sna-projects.com/sna/media/kafka_hadoop.pdf
kafka是一种高吞吐量的分布式发布订阅消息系统，她有如下特性：
通过O(1)的磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。
高吞吐量：即使是非常普通的硬件kafka也可以支持每秒数十万的消息。
支持通过kafka服务器和消费机集群来分区消息。
支持Hadoop并行数据加载。
设计侧重高吞吐量，用于好友动态，相关性统计，排行统计，访问频率控制，批处理等系统。大部分的消息中间件能够处理实时性要求高的消息/数据，但是对于队列中大量未处理的消息/数据在持久性方面比较弱。
kakfa的consumer使用拉的方式工作。
安装kafka
下载：http://people.apache.org/~nehanarkhede/kafka-0.7.0-incubating/kafka-0.7.0-incubating-src.tar.gz
> tar xzf kafka-.tgz
> cd kafka-
> ./sbt update
> ./sbt package
启动zkserver:
bin/zookeeper-server-start.sh config/zookeeper.properties
启动server:
bin/kafka-server-start.sh config/server.properties
就是这么简单。
使用kafka
- import java.util.Arrays;
- import java.util.List;
- import java.util.Properties;
- import kafka.javaapi.producer.SyncProducer;
- import kafka.javaapi.message.ByteBufferMessageSet;
- import kafka.message.Message;
- import kafka.producer.SyncProducerConfig;
- ...
- 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);
- Send a single message
- // The message is sent to a randomly selected partition registered in ZK
- ProducerData<String, String> data = new ProducerData<String, String>("test-topic", "test-message");
- producer.send(data);
- producer.close();
这样就是一个标准的producer。
consumer的代码
- // 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”, to allow 4 threads to consume
- Map<String, List<KafkaMessageStream<Message>>> topicMessageStreams =
- consumerConnector.createMessageStreams(ImmutableMap.of("test", 4));
- List<KafkaMessageStream<Message>> streams = topicMessageStreams.get("test");
- // 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 KafkaMessageStream<Message> stream: streams) {
- executor.submit(new Runnable() {
- public void run() {
- for(Message message: stream) {
- // process message
- }
- }
- });
- }