前言：
周末闲暇工夫无心找到了一套个性化举荐的源码，整体我的项目使用了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). 其余命令
查看所有topicskafka-topics.bat --list --zookeeper localhost:2181新增topics kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testKafka存储机制：
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.Propertiesimport 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()}") } } }}