关于kafka:Windows下Kafka环境搭建

前言：

周末闲暇工夫无心找到了一套个性化举荐的源码，整体我的项目使用了 SSH，HDFS，Flume，Hive，Kafka，Spark，Scala 等。运行时，原本通过 spark 计算业务埋点数据时，却发现本地没有 Kafka。因为我始终也没应用过 Kafka，所以也作为新人，浅谈以下 Kafka 的环境装置与别离在 PHP，Scala 中的应用。

比照：

1. 横向，相比其余中间件。

对于 kafka 与其余消息中间件的比拟，网上很多的博主，不论是从运行原理还是中间件架构都有很具体的介绍。因为我平时用 Rabbit 居多，在没有看他人介绍前。Rabbi 比 Kafka 于 PHP 开发更敌对。因为 kafka 除了 PHP 的 composer 依赖包长年不更新外，kafka 在 windows 下的 PHP 扩大须要本人编译。从这一点上看 Rabbit 就更适宜业务性的音讯队列，更别说他还有事务等对音讯生产的高保障。kafka 在数据增量方面更具劣势，所以少数在大数据和举荐零碎中都有使用。

2. 纵向，相比其余版本。

如题目所见，这里次要是 2.8+ 与之前版本的比照。因为在 2.8 以前，kafka 装置前须要装置 zookeeper。这里只是一小个区别，其余的新个性具体参考 kafka 官网文档，因为我看到网上对于 kafka 的装置文章，他人都会装置 zookeeper，所以这里就特地阐明以下，以及前面启动时与其他人博客的不同。

装置：

1. 下载

下载地址能够在浏览器搜寻 kafka 官网自行下载，见上图。

2. 配置

下载完后目录构造如下，进入 config, 次要对 zookeeper.properties 和 server.properties 进行散布节点，服务端口，日志寄存目录等等的设置，后期也是什么不必管放弃默认配置进行启动。

3. 启动

也不晓得是不是从 2.8 开始，bin 目录下多了一个 windows。所以在 windows 下启动进入到改目录，运行如下命令执行 bat 文件。留神启动的时候先 zookeeper 后 kafka，进行的时候先 kafka 后 zookeeper。

(1). zookeeper 启动 zookeeper-server-start.bat ….\config\zookeeper.properties &

(2).kafka 启动 kafka-server-start.bat ….\config\server.properties &

(3). 其余命令

 查看所有 topics
kafka-topics.bat --list --zookeeper localhost:2181

 新增 topics 
kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

Kafka 存储机制：

topic 中 partition 存储散布 partiton 中文件存储形式 partiton 中 segment 文件存储构造在 partition 中通过 offset 查找 message 

图形化工具：

后期能够借助图形化工具疾速具象的查看 kafka 的音讯数据，也能便于了解其基本操作流程。以下举荐一块桌面端工具——offsetexplorer，能够在网上搜寻下载，当然 web 控制台也不错，比方 kafka manager。

1. kafka 连贯

2. Cluster name 查看

这个值如果没有设置的状况是生成的，能够在启动日志中查看，根目录 /logs/server.log

3. Topics 查看

通过运行一下新增 topics 或新增音讯后就能够在 Offset Explorer 查看了，更多的应用办法也能够在网上找到。

PHP 操作：

1. 下载依赖

composer require nmred/kafka-php

2.  生产者

Producer.php

<?php
 
require './vendor/autoload.php';
 
date_default_timezone_set('PRC');
 
/* 创立一个配置实例 */
$config = \Kafka\ProducerConfig::getInstance();
 
/* Topic 的元信息刷新的距离 */
$config->setMetadataRefreshIntervalMs(10000);
 
/* 设置 broker 的地址 */
$config->setMetadataBrokerList('127.0.0.1:9092');
 
/* 设置 broker 的代理版本 */
$config->setBrokerVersion('1.0.0');
 
/* 只须要 leader 确认音讯 */
$config->setRequiredAck(1);
 
/* 抉择异步 */
$config->setIsAsyn(false);
 
/* 每 500 毫秒发送音讯 */
$config->setProduceInterval(500);
 
/* 创立一个生产者实例 */
$producer = new \Kafka\Producer();
 
for($i = 0; $i < 100; $i++) {
    $producer->send([
        [
            'topic' => 'test', 
            'value' => 'test'.$i, 
        ],
    ]);
}

3. 消费者

Consumer.php

<?php
 
require './vendor/autoload.php';
 
date_default_timezone_set('PRC');
 
$config = \Kafka\ConsumerConfig::getInstance();
$config->setMetadataRefreshIntervalMs(10000);
$config->setMetadataBrokerList('127.0.0.1:9092');
$config->setGroupId('test');
$config->setBrokerVersion('1.0.0');
$config->setTopics(['test']);
 
$consumer = new \Kafka\Consumer();
$consumer->start(function($topic, $part, $message) {var_dump($message);
});

Scala 操作：

1.  创立基于 Maven 的 Scala 我的项目

(1). 创立

(2). 增加模板

（没有模板的前提）网上搜寻 Scala-archetype-simple 的地位并填写。

(3). 实现创立期待 IDE 主动构建

(4). 给我的项目增加 Scala SDK 依赖

2. 配置

(1). 批改 pom.xml 的 scala 版本为本地装置 scala 对应的号。

(2). Cannot resolve plugin org.scala-tools:maven-scala-plugin: unknown 解决办法增加一下依赖后再 Maven 刷新

<dependency>
      <groupId>org.scala-tools</groupId>
      <artifactId>maven-scala-plugin</artifactId>
      <version>2.11</version>
    </dependency>
    <dependency>
      <groupId>org.apache.maven.plugins</groupId>
      <artifactId>maven-eclipse-plugin</artifactId>
      <version>2.5.1</version>
    </dependency> 3. 增加 kafka 依赖 <!--kafka-->
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka_2.11</artifactId>
  <version>1.1.0</version>
</dependency>

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>1.1.0</version>
</dependency>

4. 创立消费者 package com.xudong

import java.util.Properties
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}

object KafkaProducerDemo {def main(args: Array[String]): Unit = {
    val prop = new Properties
    // 指定申请的 kafka 集群列表
    prop.put("bootstrap.servers", "127.0.0.1:9092")
    prop.put("acks", "all")
    // 申请失败重试次数
    //prop.put("retries", "3")
    // 指定 key 的序列化形式, key 是用于存放数据对应的 offset
    prop.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    // 指定 value 的序列化形式
    prop.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    // 配置超时工夫
    prop.put("request.timeout.ms", "60000")

    val producer = new KafkaProducer[String, String](prop)

    // 发送给 kafka
    for (i <- 1 to 25) {val msg = s"${i}: this is a linys ${i} kafka data"
      println("send -->" + msg)
      val rmd: RecordMetadata = producer.send(new ProducerRecord[String, String]("ceshi", msg)).get()
      println(rmd.toString)
      Thread.sleep(500)
    }

    producer.close()}

}

5. 创立消费者 package com.xudong

import java.util.{Collections, Properties}
import org.apache.kafka.clients.consumer.{ConsumerRecords, KafkaConsumer}

object KafkaConsumerDemo {def main(args: Array[String]): Unit = {
    val prop = new Properties
    prop.put("bootstrap.servers", "127.0.0.1:9092")
    prop.put("group.id", "group01")
    prop.put("auto.offset.reset", "earliest")
    prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    prop.put("enable.auto.commit", "true")
    prop.put("session.timeout.ms", "30000")
    val kafkaConsumer = new KafkaConsumer[String, String](prop)
    kafkaConsumer.subscribe(Collections.singletonList("ceshi"))
    // 开始生产数据
    while (true) {val msgs: ConsumerRecords[String, String] = kafkaConsumer.poll(2000)
      // println(msgs.count())
      val it = msgs.iterator()
      while (it.hasNext) {val msg = it.next()
        println(s"partition: ${msg.partition()}, offset: ${msg.offset()}, key: ${msg.key()}, value: ${msg.value()}")
      }
    }
  }

}