Kafka学习笔记之扫盲

摘要：相关概念协议高级消息队列协议是一个标准开放的应用层的消息中间件协议。可以用命令与不同，不是线程安全的。手动提交执行相关逻辑提交注意点将写成单例模式，有助于减少端占用的资源。自身是线程安全的类，只要封装得当就能最恰当的发挥好的作用。

本文使用的Kafka版本0.11

先思考些问题:

我想分析一下用户行为（pageviews），以便我能设计出更好的广告位

我想对用户的搜索关键词进行统计，分析出当前的流行趋势。这个很有意思，在经济学上有个长裙理论，就是说，如果长裙的销量高了，说明经济不景气了，因为姑娘们没钱买各种丝袜了。

有些数据，我觉得存数据库浪费，直接存硬盘又怕到时候操作效率低。

这个时候，我们就可以用到分布式消息系统了。虽然上面的描述更偏向于一个日志系统，但确实kafka在实际应用中被大量的用于日志系统。
这些场景都有一个共同点：数据是由上游模块产生，上游模块，使用上游模块的数据计算、统计、分析，这个时候就可以使用消息系统，尤其是分布式消息系统！

Kafka是一个分布式消息系统，由linkedin使用scala编写. Kafka的动态扩容是通过Zookeeper来实现的。
Zookeeper是一种在分布式系统中被广泛用来作为：分布式状态管理、分布式协调管理、分布式配置管理、和分布式锁服务的集群。kafka增加和减少服务器都会在Zookeeper节点上触发相应的事件。

1、 AMQP协议(Advanced Message Queuing Protocol,高级消息队列协议）
AMQP是一个标准开放的应用层的消息中间件（Message Oriented Middleware）协议。AMQP定义了通过网络发送的字节流的数据格式。因此兼容性非常好，任何实现AMQP协议的程序都可以和与AMQP协议兼容的其他程序交互，可以很容易做到跨语言，跨平台。

2、 一些基本的概念

消费者：（Consumer）：从消息队列中请求消息的客户端应用程序

生产者：（Producer） ：向broker发布消息的应用程序

AMQP服务端（broker）：用来接收生产者发送的消息并将这些消息路由给服务器中的队列，便于fafka将生产者发送的消息，动态的添加到磁盘并给每一条消息一个偏移量，所以对于kafka一个broker就是一个应用程序的实例

主题（Topic）：一个主题类似新闻中的体育、娱乐、教育等分类概念，在实际工程中通常一个业务一个主题。

分区（Partition）：一个Topic中的消息数据按照多个分区组织，分区是kafka消息队列组织的最小单位，一个分区可以看作是一个FIFO（ First Input First Output的缩写，先入先出队列）的队列。

生产者生产(push)消息、kafka集群、消费者获取(pull)消息这样一种架构，kafka集群中的消息，是通过Topic（主题）来进行组织的. 生产者可以选择自己喜欢的序列化方法对消息内容编码。
kafka分区是提高kafka性能的关键所在，当你发现你的集群性能不高时，常用手段就是增加Topic的分区，分区里面的消息是按照从新到老的顺序进行组织，消费者从队列头订阅消息，生产者从队列尾添加消息。

简化图如下：

我们看上面的图，我们把broker的数量减少，只有一台。现在假设我们按照上图进行部署：

Server-1 broker其实就是kafka的server，因为producer和consumer都要去连它。Broker主要还是做存储用。

Server-2是zookeeper的server端，在这里你可以先想象，它维持了一张表，记录了各个节点的IP、端口等信息（以后还会讲到，它里面还存了kafka的相关信息）。

Server-3、4、5他们的共同之处就是都配置了zkClient，这之间的连接都是需要zookeeper来进行分发的。

Server-1和Server-2的关系，他们可以放在一台机器上，也可以分开放，zookeeper也可以配集群。目的是防止某一台挂了。

kafka和JMS（Java Message Service）实现(activeMQ)不同的是:即使消息被消费,消息仍然不会被立即删除.日志文件将会根据broker中的配置要求,保留一定的时间之后删除;比如log文件保留2天,那么两天后,文件会被清除,无论其中的消息是否被消费.但kafka并没有提供JMS中的"事务性""消息传输担保(消息确认机制)""消息分组"等企业级特性;kafka只能使用作为"常规"的消息系统,在一定程度上,无法确保消息的发送与接收绝对可靠(比如,消息重发,消息发送丢失等)

对于consumer而言,它需要保存消费消息的offset,对于offset的保存和使用,有consumer来控制;当consumer正常消费消息时,offset将会"线性"的向前驱动,即消息将依次顺序被消费.事实上consumer可以使用任意顺序消费消息,它只需要将offset重置为任意值..(offset将会保存在zookeeper中)

kafka集群几乎不需要维护任何consumer和producer状态信息,这些信息有zookeeper保存;因此producer和consumer的实现非常轻量级,它们可以随意离开,而不会对集群造成额外的影响.

partitions的目的有多个.最根本原因是kafka基于文件存储.通过分区,可以将日志内容分散到多个上,来避免文件尺寸达到单机磁盘的上限;可以将一个topic切分多任意多个partitions.此外越多的partitions意味着可以容纳更多的consumer,有效提升并发消费的能力.

每个consumer属于一个consumer group;反过来说,每个group中可以有多个consumer.发送到Topic的消息,只会被订阅此Topic的每个group中的一个consumer消费（而不是该group下的所有consumer，一定要注意这点）.

如果所有的consumer都具有相同的group,这种情况和queue模式很像;消息将会在consumers之间负载均衡.

如果所有的consumer都具有不同的group,那这就是"发布-订阅";消息将会广播给所有的消费者.

在kafka中,一个partition中的消息只会被group中的一个consumer消费;每个group中consumer消息消费互相独立;我们可以认为一个group是一个"订阅"者,一个Topic中的每个partions,只会被一个"订阅者"中的一个consumer消费,不过一个consumer可以消费多个partitions中的消息

注意：kafka使用文件存储消息,这就直接决定kafka在性能上严重依赖文件系统的本身特性。

在分布式方面：

broker的部署是一种no central master的概念，并且每个节点都是同等的，节点的增加和减少都不需要改变任何配置。

producer和consumer通过zookeeper去发现topic，并且通过zookeeper来协调生产和消费的过程。

producer、consumer和broker均采用TCP连接，通信基于NIO实现。Producer和consumer能自动检测broker的增加和减少。

使用场景：

常规消息系统。

kafka可以作为"网站活性跟踪"的最佳工具;可以将网页/用户操作等信息发送到kafka中.并实时监控,或者离线统计分析等

kafka的特性决定它非常适合作为"日志收集中心";application可以将操作日志"批量""异步"的发送到kafka集群中,而不是保存在本地或者DB中;kafka可以批量提交消息/压缩消息等,这对producer端而言,几乎感觉不到性能的开支.此时consumer端可以使hadoop等其他系统化的存储和分析系统.

简单说下整个系统运行的顺序：

1.启动zookeeper的server

2.启动kafka的server

3.Producer如果生产了数据，会先通过zookeeper找到broker，然后将数据存放进broker

4.Consumer如果要消费数据，会先通过zookeeper找对应的broker，然后消费。

本地单击测试环境启动顺序：

1.启动zookeeper server ：bin/zookeeper-server-start.sh ../config/zookeeper.properties &

2.启动kafka server: bin/kafka-server-start.sh ../config/server.properties &

3.Kafka为我们提供了一个console来做连通性测试，
先创建一个topic:bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test ,你可以运行bin/kafka-topics.sh --list --zookeeper localhost:2181来检查是否创建成功和topic列表

运行producer(默认broker端口9092)：bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test 这是相当于开启了一个producer的命令行。

4.接下来运行consumer，新启一个terminal：bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

5.执行完consumer的命令后，你可以在producer的terminal中输入信息，马上在consumer的terminal中就会出现你输的信息。有点儿像一个通信客户端。

见http://kafka.apache.org/docum...
必要配置项：

broker.id

log.dirs

zookeeper.connect

APIDOC:http://kafka.apache.org/0110/...
官方github例子： https://github.com/apache/kaf...

    org.apache.kafka
    kafka-clients
    0.11.0.0



    org.apache.kafka
    kafka-streams
    0.11.0.0
 

首先贴一下官方例子：

Producer:

public class MyKafkaProducer {

    public static void main(String[] args) {
        /**
         * 这个例子中，每次调用都会创建一个Producer实例，但此处只是为了演示方便，实际使用中，请将Producer作为单例使用，它是线程安全的。

         * 从Kafka 0.11 开始，KafkaProducer支持两种额外的模式：幂等(idempotent)与事务(transactional)。幂等使得之前的at least once变成exactly once传送
         * 幂等Producer的重试不再会导致重复消息。事务允许应用程序以原子方式将消息发送到多个分区（和主题！）

         * 开启idempotence幂等:props.put("enable.idempotence", true);设置之后retries属性自动被设为Integer.MAX_VALUE;;acks属性自动设为all;;max.inflight.requests.per.connection属性自动设为1.其余一样。

         * 开启事务性： props.put("transactional.id", "my-transactional-id");一旦这个属性被设置，那么幂等也会自动开启。然后使用事务API操作即可
         */
    }
    private static void send(){
        Properties props = new Properties();
         props.put("bootstrap.servers", "localhost:9092");
         props.put("enable.idempotence", true);//开启idempotence幂等 extract-once
//         props.put("acks", "all");//acks配置控制请求被认为完成的条件
//         props.put("retries", 0);重试次数
//         props.put("batch.size", 16384);
//         props.put("linger.ms", 1);
//         props.put("buffer.memory", 33554432);
         props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
         props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

         Producer producer = new KafkaProducer<>(props);
         for (int i = 0; i < 100; i++)
             producer.send(new ProducerRecord("my-topic", Integer.toString(i), Integer.toString(i)));

         producer.close();
    }
    private static void sendInTx(){
         Properties props = new Properties();
         props.put("bootstrap.servers", "localhost:9092");
         props.put("transactional.id", "my-transactional-id");//要启用事务，必须配置一个唯一的事务id

         /**
          * http://kafka.apache.org/0110/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html
          * KafkaProducer类是线程安全的，可以在多线程之间共享。
          */
         Producer producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());

         producer.initTransactions();

         try {
             producer.beginTransaction();
             for (int i = 0; i < 100; i++){
                 // send()是异步的，会立即返回，内部是缓存到producer的buffer中，以便于生产者可以批量提交， 你也可以传递一个回调send(ProducerRecord record, Callback callback)
                 producer.send(new ProducerRecord<>("my-topic", Integer.toString(i), Integer.toString(i)));
             }
             producer.commitTransaction();
         } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
             //无法恢复的异常，我们只能关闭producer 
             producer.close();
         } catch (KafkaException e) {
             // 可恢复的异常，终止事务然后重试即可。
             producer.abortTransaction();
         }
         producer.close();
    }
}

发送完之后，我们可以用bin目录下的kafka-console-consumer来看发送的结果（当然现在用的topic是test）。可以用命令：

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

Consumer:

/**
 *与producer不同，Kafka consumer不是线程安全的。
 */
public class MyKafkaConsumer {
    /**
     * 通过配置enable.auto.commit,auto.commit.interval.ms来定期自动提交消费的偏移量
     */
    private  void recieveByAutoCommitOffset(){
        Properties props = new Properties();
         props.put("bootstrap.servers", "localhost:9092");
         props.put("group.id", "test");
         props.put("enable.auto.commit", "true");
         props.put("auto.commit.interval.ms", "1000");
         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         KafkaConsumer consumer = new KafkaConsumer<>(props);
         consumer.subscribe(Arrays.asList("foo", "bar"));
         while (true) {
             ConsumerRecords records = consumer.poll(100);
             for (ConsumerRecord record : records)
                 System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
         }
//         consumer.wakeup();
    }
    /**
     * 手动提交消费的偏移量,这样用户可以控制记录何时被视为已消费，从而提交其偏移量。 当消息的消耗与一些处理逻辑相结合时，这是有用的，因为在完成处理之前不应将消息视为已消费。
     */
    private void recieveByManualCommitOffset(){
        Properties props = new Properties();
         props.put("bootstrap.servers", "localhost:9092");
         props.put("group.id", "test");
         props.put("enable.auto.commit", "false");//手动提交offset
         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         KafkaConsumer consumer = new KafkaConsumer<>(props);
         consumer.subscribe(Arrays.asList("foo", "bar"));
         final int minBatchSize = 200;
         List> buffer = new ArrayList<>();
         while (true) {
             ConsumerRecords records = consumer.poll(100);
             for (ConsumerRecord record : records) {
                 buffer.add(record);
             }
             if (buffer.size() >= minBatchSize) {
//                 insertIntoDb(buffer); 执行相关逻辑
                 consumer.commitSync();//提交offset
                 buffer.clear();
             }
         }
    }
}

Streams:

public class MyKafkaStreams {
    public void test(){
        Map props = new HashMap<>();
         props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-stream-processing-application");
         props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
         props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
         props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
         StreamsConfig config = new StreamsConfig(props);

         KStreamBuilder builder = new KStreamBuilder();
         builder.stream("my-input-topic").mapValues(value -> value.toString()+"!!!").to("my-output-topic");

         KafkaStreams streams = new KafkaStreams(builder, config);
         streams.start();
    }
}

注意点：

将producer写成单例模式，有助于减少zookeeper端占用的资源。Producer自身是线程安全的类，只要封装得当就能最恰当的发挥好producer的作用。(ZkClient去连接zookeeper的server时候都会创建sendThread和eventThread两个线程，其中sendThread主要用于client与server端之间的网络连接，真正的处理线程由eventThread来执行。Zookeeper是一个分布式的协调框架，而分布式应用中经常会出现动态的增加或删除节点的操作，所以为了实时了解分布式整个节点的数量和基本信息，就有必要维护一个长连接的线程与服务端保持连接。另外zookeeper连接时占用的时间也比较长，如果每次生产数据时都连接发起一次连接势必造成了大量时间的耗费。)

kafka是将消息按照topic的形式存储，一个topic会按照partition存在同一个文件夹下，目录在config/server.properties中指定：

# The directory under which to store log files
log.dir=/tmp/kafka-logs

在消息系统中都会有这样一个问题存在，数据消费状态这个信息到底存哪里。是存在consumer端，还是存在broker端。对于这样的争论，一般会出现三种情况：

At most once ：消息一旦发出就立马标记已消费，不会再有第二发生即使失败了，缺点是容易丢失消息。

At least once ：消息至少发送一次,如果消息未能接受成功,可能会重发,直到接收成功.

Exactly once ：每个消息仅发生一次，而且一次就能确保到达。这是理想状态。(kafka0.11支持幂等之后，在开启幂等的情况下，就是这种模式)

at most once: 消费者fetch消息,然后保存offset,然后处理消息;当client保存offset之后,但是在消息处理过程中出现了异常,导致部分消息未能继续处理.那么此后"未处理"的消息将不能被fetch到,这就是"atmost once".

at least once: 消费者fetch消息,然后处理消息,然后保存offset.如果消息处理成功之后,但是在保存offset阶段zookeeper异常导致保存操作未能执行成功,这就导致接下来再次fetch时可能获得上次已经处理过的消息,这就是"at least once"，原因offset没有及时的提交给zookeeper，zookeeper恢复正常还是之前offset状态.

https://github.com/xbynet/log...

http://kafka.apache.org/docum...
http://kafka.apache.org/intro...
https://my.oschina.net/ielts0...
http://blog.csdn.net/my_bai/a...
http://www.infoq.com/cn/artic...
http://www.cnblogs.com/likehu...
http://www.cnblogs.com/likehu...
https://www.iteblog.com/archi...