kafka

kafka

介绍

简介：Kafka是一个分布式的，可划分的，冗余备份的持久性的日志服务。它主要用于处理活跃的流式数据。

消息系统术语：消息归纳topic，发布消息producers，消费消息consumer，集群服务broker。

关系图

图片链接

总结：两个作用，1.降低系统组网复杂度。2.降低编程复杂度。

Kafka独特设计在什么地方？

1.同时为发布和订阅提供高吞吐量。据了解，Kafka每秒可以生产约25万消息（50 MB），每秒处理55万消息（110 MB）。

2.可进行持久化操作。将消息持久化到磁盘，因此可用于批量消费，例如ETL，以及实时应用程序。通过将数据持久化到硬盘以及replication防止数据丢失。

3.分布式系统，易于向外扩展。所有的producer、broker和consumer都会有多个，均为分布式的。无需停机即可扩展机器。

4.消息被处理的状态是在consumer端维护，而不是由server端维护。当失败时能自动平衡。

5.支持online和offline的场景。

kafka架构

架构

producer，broker，consumer关系图

Kafka的整体架构非常简单，是显式分布式架构，producer、broker（kafka）和consumer都可以有多个。

Producer，consumer实现Kafka注册的接口，数据从producer发送到broker，broker承担一个中间缓存和分发的作用。

broker分发注册到系统中的consumer。broker的作用类似于缓存，即活跃的数据和离线处理系统之间的缓存。

部件

Partition

Topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。

Topic

特指Kafka处理的消息源（feeds of messages）的不同分类。

Message

消息，是通信的基本单位，每个producer可以向一个topic（主题）发布一些消息。

Producers

消息和数据生产者，向Kafka的一个topic发布消息的过程叫做producers。

Consumers

消息和数据消费者，订阅topics并处理其发布的消息的过程叫做consumers。

Broker

缓存代理，Kafa集群中的一台或多台服务器统称为broker。

消息发送图

流程图

Kafka如何搭建及创建topic、发送消息、消费消息？

Topics 和Logs

topic细节图

一个topic是对一组消息的归纳。对每个topic，Kafka 对它的日志进行了分区

log implementation

实际上每个consumer唯一需要维护的数据是消息在日志中的位置，也就是offset.这个offset有consumer来维护：一般情况下随着consumer不断的读取消息，这offset的值不断增加，但其实consumer可以以任意的顺序读取消息，比如它可以将offset设置成为一个旧的值来重读之前的消息。

通常简单的由负载均衡机制随机选择分区，但也可以通过特定的分区函数选择分区。使用的更多的是第二种。

队列模式和发布-订阅模式

队列模式中，consumers可以同时从服务端读取消息，每个消息只被其中一个consumer读到。

发布-订阅模式中消息被广播到所有的consumer中。

Consumers可以加入一个consumer 组，共同竞争一个topic，topic中的消息将被分发到组中的一个成员中。

如果所有的consumer都在一个组中，这就成为了传统的队列模式，在各consumer中实现负载均衡。

如果所有的consumer都不在不同的组中，这就成为了发布-订阅模式，所有的消息都被分发到所有的consumer中。

集群关系图

优点

相比传统的消息系统，Kafka可以很好的保证有序性。

Kafka可以在多个consumer组并发的情况下提供较好的负载均衡。

环境搭建

下载解压

https://www.apache.org/dyn/closer.cgi?path=/kafka//kafka_.tgz

tar -xzf kafka_2.9.2-0.8.1.1.tgz

cd kafka_2.9.2-0.8.1.1

启动服务

Kafka用到了Zookeeper，所有首先启动Zookper。

bin/zookeeper-server-start.sh config/zookeeper.properties

启动kafka

bin/kafka-server-start.sh config/server.properties

测试

创建topic

创建一个叫做“test”的topic，它只有一个分区，一个副本。

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

通过list命令查看创建的topic

bin/kafka-topics.sh --list --zookeeper localhost:2181

发送消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

启动consumer

Kafka也有一个命令行consumer可以读取消息并输出到标准输出：

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

kafka开发环境搭建

maven项目

\u00A0 \u00A0 \u00A0 \u00A0 org.apache.kafka\u00A0 \u00A0 \u00A0 \u00A0 kafka_2.10\u00A0 \u00A0 \u00A0 \u00A0 0.8.0

参数配置

package com.sohu.kafkademon;public interface KafkaProperties{\u00A0 \u00A0 final static String zkConnect = \"10.22.10.139:2181\";\u00A0 \u00A0 final static String groupId = \"group1\";\u00A0 \u00A0 final static String topic = \"topic1\";\u00A0 \u00A0 final static String kafkaServerURL = \"10.22.10.139\";\u00A0 \u00A0 final static int kafkaServerPort = 9092;\u00A0 \u00A0 final static int kafkaProducerBufferSize = 64 * 1024;\u00A0 \u00A0 final static int connectionTimeOut = 20000;\u00A0 \u00A0 final static int reconnectInterval = 10000;\u00A0 \u00A0 final static String topic2 = \"topic2\";\u00A0 \u00A0 final static String topic3 = \"topic3\";\u00A0 \u00A0 final static String clientId = \"SimpleConsumerDemoClient\";}

producer，consumer等

学习总结来源

http://www.aboutyun.com/thread-12882-1-1.html

http://blog.jobbole.com/75328/

http://www.cnblogs.com/likehua/p/3999538.html