WindowskafkaPHP技巧_浅谈Kafka28在Windows下的搭建与运用

文章目录 [+]

比拟：

1. 横向，比较其他中间件。

关于kafka与其他中间件的比较，网上很多的博主，不管是从运行事理还是中间件架构都有很详细的先容。
由于我平时用Rabbit居多，在没有看别人先容前。
Rabbi比Kafka于PHP开拓更友好。
由于kafka除了PHP的composer依赖包常年不更新外，kafka在windows下的PHP扩展须要自己编译。
从这一点上看Rabbit就更适宜业务性的行列步队，更别说他还有事务等对消费的高保障。
kafka在数据增量方面更具上风，以是多数在大数据和推举系统中都有利用。

WindowskafkaPHP技巧_浅谈Kafka28在Windows下的搭建与运用

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 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依赖

<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.xudongimport 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.xudongimport 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()}") } } }}