Kafka 学习笔记（二） ：初探 Kafka

本文由 GodPan 发表在 ScalaCool 团队博客。

看完上一篇，相信你们对消息系统以及Kafka的总体构成都有了初步了解，学习一个东西最好的办法，就是去使用它，今天就让咱们一块儿窥探一下Kafka，并完成本身的处女做。

消息在Kafka中的历程

虽然咱们掌握东西要一步一步来，可是咱们在大体了解了一个东西后，会有利于咱们对它的理解和学习，因此咱们能够先来看一下一条消息从发出到最后被消息者接收到底经历了什么？

上图简要的说明了消息在Kafka中的整个流转过程（假设已经部署好了整个Kafka系统，并建立了相应的Topic，分区等细节后续再单独讲）：

• 1.消息生产者将消息发布到具体的Topic，根据必定算法或者随机被分发到具体的分区中；
• 2.根据实际需求，是否须要实现处理消息逻辑；
• 3.若须要，则实现具体逻辑后将结果发布到输出Topic；
• 4.消费者根据需求订阅相关Topic，并消费消息；

总的来讲，怎么流程仍是比较清晰和简单的，下面就跟我一块儿来练习Kafka的基本操做，最后实现一个单词计数的小demo。

基础操做

如下代码及相应测试在如下环境测试经过：Mac OS + JDK1.8，Linux系统应该也能跑通，Windows有兴趣的同窗能够去官网下载相应版本进行相应的测试练习。

下载Kafka

Mac系统同窗可使用brew安装：

brew install kafka
复制代码

Linux系统同窗能够从官网下载源码解压，也能够直接执行如下命令:

cd 
mkdir test-kafka && cd test-kafka
curl -o kafka_2.11-1.0.1.tgz http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/1.0.1/kafka_2.11-1.0.1.tgz
tar -xzf kafka_2.11-1.0.1.tgz
cd kafka_2.11-1.0.1

复制代码

启动

Kafka使用Zookeeper来维护集群信息，因此这里咱们先要启动Zookeeper，Kafka与Zookeeper的相关联系跟结合后续再深刻了解，毕竟不能一口吃成一个胖子。

bin/zookeeper-server-start.sh config/zookeeper.properties
复制代码

接着咱们启动一个Kafka Server节点：

bin/kafka-server-start.sh config/server.properties
复制代码

这时候Kafka系统已经算是启动起来了。

建立Topic

在一切就绪以后，咱们要开始作极其重要的一步，那就是建立Topic，Topic是整个系统流转的核心，另外Topic自己也包含着不少复杂的参数，好比复制因子个数，分区个数等，这里为了从简，咱们将对应的参数都设为1，方便你们测试：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic kakfa-test
复制代码

其中参数的具体含义：

属性	功能
--create	表明建立Topic
--zookeeper	zookeeper集群信息
--replication-factor	复制因子
--partitions	分区信息
--topic	Topic名称

这时候咱们已经建立好了一个叫kakfa-test的Topic了。

向Topic发送消息

在有了Topic后咱们就能够向其发送消息：

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kakfa-test
复制代码

而后咱们向控制台输入一些消息：

this is my first test kafka
so good
复制代码

这时候消息已经被发布在kakfa-test这个主题上了。

从Topic获取消息

如今Topic上已经有消息了，如今能够从中获取消息被消费：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka-test --from-beginning
复制代码

这时候咱们能够在控制台看到：

this is my first test kafka
so good
复制代码

至此咱们就测试了最简单的Kafka Demo，但愿你们能本身动手去试试，虽然很简单，可是这能让你对整个Kafka流程能更熟悉。

WordCount

下面咱们来利用上面的一些基本操做来实现一个简单WordCount程序，它具有如下功能：

• 1.支持词组持续输入，即生产者不断生成消息；
• 2.程序自动从输入Topic中获取原始数据，而后通过处理，将处理结果发布在计数Topic中；
• 3.消费者能够从计数Topic获取相应的WordCount的结果；

1.启动kafka

与上文的启动同样，按照其操做便可。

2.建立输入Topic

bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic kafka-word-count-input --partitions 1 --replication-factor 1
复制代码

3.向Topic输入消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kafka-word-count-input
复制代码

4.流处理逻辑

这部份内容是整个例子的核心，这部分代码有Java 8+和Scala版本，我的认为流处理用函数式语法表达的更加简洁清晰，推荐你们用函数式的思惟去尝试写如下，发现本身不再想写Java匿名内部类这种语法了。

咱们先来看一个Java 8的版本:

public class WordCount {
    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-word-count");
        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());

        final StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> source = builder.<String, String>stream("kafka-word-count-input");
        Pattern pattern = Pattern.compile("\\W+");
        source
           .flatMapValues(value -> Arrays.asList(pattern.split(value.toLowerCase(Locale.getDefault()))))
           .groupBy((key, value) -> value)
           .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("counts-store")).mapValues(value->Long.toString(value))
           .toStream()
           .to("kafka-word-count-output");
        final KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}
复制代码

是否是很惊讶，用java也能写出如此简洁的代码，因此说若是有适用场景，推荐你们尝试的用函数式的思惟去写写java代码。

咱们再来看看Scala版本的：

object WordCount {
  def main(args: Array[String]) {
    val props: Properties = {
      val p = new Properties()
      p.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-word-count")
      p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
      p.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String.getClass)
      p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String.getClass)
      p
    }

    val builder: StreamsBuilder = new StreamsBuilder()
    val source: KStream[String, String] = builder.stream("kafka-word-count-input")
    source
      .flatMapValues(textLine => textLine.toLowerCase.split("\\W+").toIterable.asJava)
      .groupBy((_, word) => word)
      .count(Materialized.as[String, Long, KeyValueStore[Bytes, Array[Byte]]]("counts-store")).toStream.to("kafka-word-count-output")
    val streams: KafkaStreams = new KafkaStreams(builder.build(), props)
    streams.start()
  }
}
复制代码

能够发现使用Java 8函数式风格编写的代码已经跟Scala很类似了。

5.启动处理逻辑

不少同窗电脑上并无装sbt，因此这里演示的利用Maven构建的Java版本，具体执行步骤请参考戳这里kafka-word-count上的说明。

6.启动消费者进程

最后咱们启动消费者进程，并在生产者中输入一些单词，好比：

最后咱们能够在消费者进程中看到如下输出：

bin/kafka-console-consumer.sh --topic kafka-word-count-output --from-beginning --bootstrap-server localhost:9092  --property print.key=true
复制代码

总结

本篇文章主要是讲解了Kafka的基本运行过程和一些基础操做，但这是咱们学习一个东西必不可少的一步，只有把基础扎实好，才能更深刻的去了解它，理解它为何这么设计，我在这个过程当中也遇到不少麻烦，因此仍是但愿你们可以本身动手去实践一下，最终能收获更多。