关于kafka:Windows下Kafka环境搭建

105次阅读

共计 5146 个字符，预计需要花费 13 分钟才能阅读完成。

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

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

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

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

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

也不晓得是不是从 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

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

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

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

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

composer require nmred/kafka-php

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, 
        ],
    ]);
}

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-archetype-simple 的地位并填写。

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

 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()}
 
}

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

正文完

Kafka

发表至： Kafka

2023-05-27

0

关于kafka:聊聊-KafkaKafka-如何保证一致性

关于kafka:Apache-Kafka教程Kafka新手入门

关于kafka:二Kafka之集群架构原理

Kafka学习笔记

关于python:用Pillow处理图像Python中的实践指南

关于kafka:Windows下Kafka环境搭建

前言：

比照：

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

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

装置：

1. 下载

2. 配置

3. 启动

Kafka 存储机制：

图形化工具：

1. kafka 连贯

2. Cluster name 查看

3. Topics 查看

PHP 操作：

1. 下载依赖

2. 生产者

3. 消费者

Scala 操作：

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

(1). 创立

(2). 增加模板

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

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

2. 配置

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

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

4. 创立消费者 package com.xudong

5. 创立消费者 package com.xudong

Just My Socks（注册教程内含优惠码）

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

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

	<?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,
	],
	]);
	}


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

	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()}

	}

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

	}

关于kafka:Windows下Kafka环境搭建

前言：

比照：

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

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

装置：

1. 下载

2. 配置

3. 启动

Kafka 存储机制：

图形化工具：

1. kafka 连贯

2. Cluster name 查看

3. Topics 查看

PHP 操作：

1. 下载依赖

2. 生产者

3. 消费者

Scala 操作：

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

(1). 创立

(2). 增加模板

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

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

2. 配置

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

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

4. 创立消费者 package com.xudong

5. 创立消费者 package com.xudong

Just My Socks（注册教程 内含优惠码）

Just My Socks（注册教程内含优惠码）