首页 > 系统相关 >浅谈Kafka2.8+在Windows下的搭建与使用

浅谈Kafka2.8+在Windows下的搭建与使用

时间:2023-05-13 14:00:54浏览次数:48  
标签:浅谈 Windows zookeeper Kafka2.8 prop put org kafka config

前言:

        周末空闲时间无意找到了一套个性化推荐的源码,整体项目运用了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()}")
      }
    }
  }

}

 

标签:浅谈,Windows,zookeeper,Kafka2.8,prop,put,org,kafka,config
From: https://blog.51cto.com/u_11161174/6273696

相关文章

  • 打开windows批处理大门
    大家好,我是xiezhr。1前言打开历史文章一看,上一篇文章是2021年3月20号更新的,又拖更了。一个原因是,最近工作上真的挺忙的,有比较着急需要加班加点赶的需求。好在清明前算是把比较着急的改好了。本来安排清明也是要加班的,但是真的加不动了。(连着加班真的挺影响效率的,适当休息是非......
  • VCL 中的 Windows API 函数: BeginDeferWindowPos
    BeginDeferWindowPos和DeferWindowPos、EndDeferWindowPos是一组一起使用的函数,可对一组窗口的位置、大小、Z序等进行调整,在ExtCtrls单元有用到.下面先用常规方法实现对Panel1中的一组Button进行调整,然后再用上面三个函数重新实现.用BeginDeferWindowPos、Defer......
  • VCL 中的 Windows API 函数: AlphaBlend
    AlphaBlend是指定图像混合透明的函数,在Graphics、GraphUtil、RibbonStyleActnCtrls单元用到.下面的测试是把一张图片显示在窗体,并可以调整透明度.、 unitUnit1;interfaceuses Windows,Messages,SysUtils,Variants,Classes,Graphics,Controls,Forms, ......
  • windows提权-绕过UAC提权
    windows提权-绕过UAC提权目录windows提权-绕过UAC提权一、UAC概述二、MSF和CS过UAC三、基于白名单AutoElevate绕过UAC四、基于白名单DLL劫持绕过UAC提权五、CVE-2019-1388UAC提权六、Windows令牌概述和令牌窃取攻击一、UAC概述UAC(UserAccountControl),中文翻译为用户帐户控制,......
  • 【Windows】【Chrome】获取 Chrome 离线安装包
    背景有的时候我们需要下载最新版的chrome来试试新特性通用方法https://blog.csdn.net/RadiantJeral/article/details/114436104......
  • Windows——Windows11右键经典栏与传统栏转换
    图1.Windows11右键默认为经典菜单栏图2.传统右键菜单栏/显示更多选项由经典右键菜单栏变为传统右键菜单栏【图1变图2】运行reg.exeadd"HKCU\Software\Classes\CLSID\{86ca1aa0-34aa-4e8b-a509-50c905bae2a2}\InprocServer32"/f/ve将【显示更多选项】作为默......
  • Windows查看端口情况
    一、查看Windows目前所有使用端口netstat-ano二、查看指定的端口netstat-ano|findstr3389三、使用tasklist命令查看指定PID对应的进程名tasklist|findstr283664  ......
  • windows task
    Win+Randtypingtaskschd.mscschtasks.exe/Create Addanewscheduledtask/tn Nameoftask/sc schedulefrequency(MINUTE,HOURLY,DAILYandsoon)/d Onwhichdayordayofmonththetaskshallbescheduled.Youcanuse*forschedulingoneveryday......
  • 修改Windows远程桌面的端口
    一、图形界面下修改启动注册表编辑器。(在“搜索”框中键入regedit。)导航到以下注册表子项:HKEY_LOCAL_MACHINE\System\CurrentControlSet\Control\TerminalServer\WinStations\RDP-Tcp查找端口号单击“编辑”>“修改”,然后单击“十进制”。键入新端口号,然后单击“确定”......
  • 浅谈一下ThinkPHP5.1实现事务嵌套的特性
    前言:       在我们平时做的一个项目中,线上环境突然发现数据库被锁住。导致很多有关数据插入和修改的接口全都瘫痪,项目基于ThinkPHP5.1。报错的时候,我们发现了一条sql错误日志,如下。   根据错误信息提示,是说有一个事务回滚时没有找到savepoint的暂存点。所以问题应该......