首页 > 其他分享 >kafka单机安装和简单使用

kafka单机安装和简单使用

时间:2023-05-10 11:45:20浏览次数:35  
标签:单机 ip zookeeper kafka topic apache org 安装

1 kafka下载

  linux环境安装kafka,需要预先准备相关资源

  这里使用的是kafka_2.12-2.5.1版本,下载路径为:http://archive.apache.org/dist/kafka/2.5.1/kafka_2.12-2.5.1.tgz

  也可以通过命令wget http://archive.apache.org/dist/kafka/2.5.1/kafka_2.12-2.5.1.tgz进行资源获取;

 

2 下载zookeeper

  这里是apache-zookeeper-3.5.9-bin.tar.gz为例,官网:https://zookeeper.apache.org/

 

3 安装zookerper

1)解压

tar -zxvf apache-zookeeper-3.5.9-bin.tar.gz

 

2)重命名,看起来舒服点

mv apache-zookeeper-3.5.9-bin zookeeper-3.5.9

 

3)进入配置文件目录conf

  复制一份配置文件

cp zoo_sample.cfg zoo.cfg

 

4)修改配置

  修改zoo.cfg中的内容为,主要修改dataDir路径以及端口号

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.
dataDir=/usr/local/zookeeper-3.6.1/data/
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1

## Metrics Providers
#
# https://prometheus.io Metrics Exporter
#metricsProvider.className=org.apache.zookeeper.metrics.prometheus.PrometheusMetricsProvider
#metricsProvider.httpPort=7000
#metricsProvider.exportJvmInfo=true

 

5)启动zookeeper
  进入bin目录

  执行命令

./zkServer.sh start

 

6)查看状态

./zkServer.sh status

 

6)停止命令

./zkServer.sh stop

 

4 安装kafka

1)解压

tar -zxvf kafka_2.12-2.5.1.tgz

 

2)进入配置文件目录config

  复制一份配置文件

  cp server.properties server-copy.properties

 

3)修改配置文件

  配置以下两个属性(这里是在同一个虚拟机上安装的zookeeper和kafka) 

  listeners:这里的ip填写kafka所在服务器的ip
 zookeeper.connect:这里的ip填写zookeeper所在服务器的ip
listeners=PLAINTEXT://192.168.28.110:9092
zookeeper.connect=192.168.28.110:2181

 

4)启动

  进入bin目录

./kafka-server-start.sh -daemon ../config/server.properties

 

5)停止命令

./kafka-server-stop.sh

 

6)查看日志

   进入logs目录

tail -f server.log

 

5 springboot测试使用kafka

1)添加依赖

<!-- 引入kafka依赖 -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.8.6</version>
        </dependency>

 

2)application.yml添加配置

#kafka
spring:
  kafka:
    bootstrap-servers: 192.168.28.110:9092 # kafka集群信息,多个用逗号间隔
    # 生产者
    producer:
      # 重试次数,设置大于0的值,则客户端会将发送失败的记录重新发送
      retries: 3
      batch-size: 16384 #批量处理大小,16K
      buffer-memory: 33554432 #缓冲存储大,32M
      acks: 1
      # 指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    # 消费者
    consumer:
      # 消费者组
      group-id: TestGroup
      # 是否自动提交
      enable-auto-commit: false
      # 消费偏移配置
      # none:如果没有为消费者找到先前的offset的值,即没有自动维护偏移量,也没有手动维护偏移量,则抛出异常
      # earliest:在各分区下有提交的offset时:从offset处开始消费;在各分区下无提交的offset时:从头开始消费
      # latest:在各分区下有提交的offset时:从offset处开始消费;在各分区下无提交的offset时:从最新的数据开始消费
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    # 监听
    listener:
      # record:当每一条记录被消费者监听器(ListenerConsumer)处理之后提交
      # batch:当每一批poll()的数据被ListenerConsumer处理之后提交
      # time:当每一批poll()的数据被ListenerConsumer处理之后,距离上次提交时间大于TIME时提交
      # count:当每一批poll()的数据被ListenerConsumer处理之后,被处理record数量大于等于COUNT时提交
      # count_time:TIME或COUNT中有一个条件满足时提交
      # manual:当每一批poll()的数据被ListenerConsumer处理之后, 手动调用Acknowledgment.acknowledge()后提交
      # manual_immediate:手动调用Acknowledgment.acknowledge()后立即提交,一般推荐使用这种
      ack-mode: manual_immediate

 

3)添加一个消费者

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * 消费者
 * kafka监听器
 */
@Component
public class KafkaConsumer {

    /**
     * kafka的监听器1,topic为"topic_test",消费者组为"group_topic_test"
     * @param record
     * @param item
     */
    @KafkaListener(topics = "topic_test", groupId = "group_topic_test")
    public void topicListener1(ConsumerRecord<String, String> record, Acknowledgment item) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        //手动提交
        item.acknowledge();
    }


    /**
     * 配置多个消费组
     * kafka的监听器2,topic为"topic_test2",消费者组为"group_topic_test"
     * @param record
     * @param item
     */
    @KafkaListener(topics = "topic_test2",groupId = "group_topic_test2")
    public void topicListener2(ConsumerRecord<String, String> record, Acknowledgment item) {
        String value = record.value();
     
        System.out.println(value);
        System.out.println(record);
        item.acknowledge();
    }

}

 

4)生产者

@RestController
@RequestMapping("/kafka")
public class KafkaController {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @RequestMapping("/send")
    public void send() {
        kafkaTemplate.send("topic_test",  "key", "测试kafka消息");
    }
}

 

5)启动项目,访问接口

 

6 在阿里云上安装zookeeper和kafka

1)安装zookeeper一样

 

2)安装kafka

   在配置上有所不同

  listeners:是监听端口,需要使用内网ip(注意,不要使用localhost或者127.0.0.1)
  advertised.listeners:给外部的,使用外网ip
listeners=PLAINTEXT://阿里云服务器内网ip:9092
advertised.listeners=PLAINTEXT://阿里云服务器外网ip:9092
zookeeper.connect=阿里云服务器外网ip:2181

  阿里云服务器的内网ip可以在以下位置看到

  

  

   也可以通过命令ifconfig看到

  

 

标签:单机,ip,zookeeper,kafka,topic,apache,org,安装
From: https://www.cnblogs.com/jthr/p/17387512.html

相关文章

  • Rocky Linux:编译安装nginx 1.24.0(Rocky Linux 9.1)
    一,官网:http://nginx.org/如图:二,下载并解压下载:[root@imgnginx]#wgethttp://nginx.org/download/nginx-1.24.0.tar.gz解压:[root@imgnginx]#tar-zxvfnginx-1.24.0.tar.gz 说明:刘宏缔的架构森林是一个专注架构的博客,地址:https://www.cnblogs.com/architectfor......
  • Google Chrome浏览器离线安装包下载方式
    GoogleChrome是应用很广泛的浏览器,默认是在线安装模式。如果网络速度很慢,或者没有网络的时候,就需要离线安装包了。32位Chrome离线包下载:http://www.google.cn/chrome/browser/desktop/index.html?standalone=164位Chrome离线包下载:http://www.google.cn/chrome/brow......
  • JMeter安装配置
    一、下载并配置JDK(每个学JAVA的应该都会,不过多赘述)注:需JDK8以上版本二、下载安装JMeter1、官网连接:ApacheJMeter-DownloadApacheJMeter2、选择所需版本进行下载3、将下载成功的压缩包解压到指定目录当中,即安装成功。4、配置环境变量变量名输入:JMETER_HOME变量值输入:JM......
  • Windows安装Wireshark实现127.0.0.1抓包
    Windows安装Wireshark实现127.0.0.1抓包 简介Wireshark(前称Ethereal)是一个网络封包分析软件。官网下载Npcap支持Win7及以上系统,基于WinPcap库,可以抓取通过127.0.0.1本地环回地址的包。官网WinPcap是Wireshark默认抓包工具,据Wireshark和Npcap的官网说不再更新......
  • kafka消费者组状态--rebalance(参考其他作者文档)
    零、kafka消费者组状态Stable:消费者组中所有消费者都已经加入并且正在消费消息。Rebalancing:消费者组正在重新平衡,即有消费者加入或退出消费者组时发生的状态。在重新平衡期间,Kafka会暂停所有消费者的消费,并重新分配分区。Empty:消费者组中没有消费者或者所有消费者都已经退出......
  • Linux安装gitbook
    Linux安装gitbook下载nodejs包wgethttps://npm.taobao.org/mirrors/node/v12.16.1/node-v12.16.1-linux-x64.tar.gz解压tar-zxvfnode-v12.16.1-linux-x64.tar.gz建立软连接[root@VM-16-13-centos~]#lsnode-v12.16.1-linux-x64node-v12.16.1-linux-x64.tar.......
  • 安装docker和docker-compose的shell脚本(Centos7版本)
    在执行脚本之前,我们需要先做两件事:避免防火墙与docker产生冲突,应先关闭防火墙。shell#去掉防火墙的开机自启动systemctldisablefirewalld.service#关闭防火墙systemctlstopfirewalld.service国内拉取dockerhub中的镜像速度一般都很慢,现在有一种方法可以提高......
  • Centos 安装 supervisord服务
    安装supervisord执行如下命令安装supervisordyuminstallepel-release-yyuminstallsupervisor-y设置开机启动supervisordsystemctlenablesupervisord启动supervisordsystemctlstartsupervisord配置supervisord默认配置文件位置:/etc/supervisord.conf默......
  • Building wheel for opencv-python (pyproject.toml) ,安装命令增加 --verbose 参数
    Mac安装paddlehub出现Buildingwheelsforcollectedpackages:opencv-python,ffmpy,jieba,seqeval,futureBuildingwheelforopencv-python(pyproject.toml)...解决方法pipinstallopencv-python-ihttps://pypi.tuna.tsinghua.edu.cn/simple--verbose......
  • CasaOs安装和卸载
    QuickSetupCasaOS一步安装:wget-qO-https://get.casaos.io|sudobashorcurl-fsSLhttps://get.casaos.io|sudobashUninstallCasaOSv0.3.3ornewercasaos-uninstallBeforev0.3.3curl-fsSLhttps://get.icewhale.io/casaos-uninstall.sh|sudobash这......