当前位置：首页 > article >正文

deepin 安装 kafka

article 2024/12/27 17:00:35

deepin 安装 kafka

kafka 有两种部署模式，分别是zookeeper模式和kraft模式
zookeeper模式需要提前部署zookeeper
kraft模式不需要部署zookeeper
本案例将分别使用zookeepper模式和kraft模式部署kafka

zookeeper模式

创建软件目录

mkdir  /home/lhz/opt
cd  /home/lhz/opt

下载

wget https://downloads.apache.org/kafka/3.9.0/kafka_2.13-3.9.0.tgz

解压

tar -zxvf kafka_2.13-3.9.0.tgz

修改目录名称

mv kafka_2.13-3.9.0 kafka

配置环境变量

vim /home/lhz/.bashrc

export KAFKA_HOME= /home/lhz/opt/kafka
export PATH=$PATH:$KAFKA_HOME/bin

修改配置文件

配置文件存放在 kafka/config目录

vim  /home/lhz/opt/kafka/config/server.properties

主要修改以下三个参数：

broker.id=1 注意不同的节点id号不同

log.dirs=/tmp/kafka-logs 修改为 log.dirs= /home/lhz/opt/kafka/kafka-logs

zookeeper.connect=localhost:2181 修改为

zookeeper.connect=lihaozhe01:2181,lihaozhe02:2181,lihaozhe03:2181/kafka

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

#
# This configuration file is intended for use in ZK-based mode, where Apache ZooKeeper is required.
# See kafka.server.KafkaConfig for additional details and defaults
#

############################# Server Basics #############################

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=0

############################# Socket Server Settings #############################

# The address the socket server listens on. If not configured, the host name will be equal to the value of
# java.net.InetAddress.getCanonicalHostName(), with PLAINTEXT listener name, and port 9092.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
#listeners=PLAINTEXT://:9092

# Listener name, hostname and port the broker will advertise to clients.
# If not set, it uses the value for "listeners".
#advertised.listeners=PLAINTEXT://your.host.name:9092

# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# The number of threads that the server uses for receiving requests from the network and sending responses to the network
num.network.threads=3

# The number of threads that the server uses for processing requests, which may include disk I/O
num.io.threads=8

# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400

# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400

# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600


############################# Log Basics #############################

# A comma separated list of directories under which to store log files
log.dirs=/home/lhz/data/kafka-logs

# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=1

# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1

############################# Internal Topic Settings  #############################
# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

############################# Log Flush Policy #############################

# Messages are immediately written to the filesystem but by default we only fsync() to sync
# the OS cache lazily. The following configurations control the flush of data to disk.
# There are a few important trade-offs here:
#    1. Durability: Unflushed data may be lost if you are not using replication.
#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.
# The settings below allow one to configure the flush policy to flush data after a period of time or
# every N messages (or both). This can be done globally and overridden on a per-topic basis.

# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000

# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000

############################# Log Retention Policy #############################

# The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.

# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168

# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824

# The maximum size of a log segment file. When this size is reached a new log segment will be created.
#log.segment.bytes=1073741824

# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000

############################# Zookeeper #############################

# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=lihaozhe:2181

# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=18000


############################# Group Coordinator Settings #############################

# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
# The default value for this is 3 seconds.
# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
group.initial.rebalance.delay.ms=0

启动停止

kafka-server-start.sh -daemon /home/lhz/opt/kafka/config/server.properties

kafka-server-stop.sh

启动脚本

vim  /home/lhz/opt/kafka/bin/kafka-zookeeper.service.sh

#!/bin/bash

case $1 in
"start"){
	echo  ------------- kafka 启动 ------------
	kafka-server-start.sh -daemon /home/lhz/opt/kafka/config/server.properties
}
;;
"stop"){
	echo  ------------- kafka 停止 ------------
	kafka-server-stop.sh
}
esac

chmod 655  /home/lhz/opt/kafka/bin/kafka-zookeeper.service.sh

kraft模式

创建软件目录

mkdir  /home/lhz/opt
cd  /home/lhz/opt

下载

wget https://downloads.apache.org/kafka/3.9.0/kafka_2.13-3.9.0.tgz

解压

tar -zxvf kafka_2.13-3.9.0.tgz

修改目录名称

mv kafka_2.13-3.9.0 kafka

配置环境变量

vim /home/lhz/.bashrc

export KAFKA_HOME= /home/lhz/opt/kafka
export PATH=$PATH:$KAFKA_HOME/bin

修改配置文件

配置文件存放在 kafka/config/kraft目录

vim  /home/lhz/opt/kafka/config/kraft/server.properties

主要修改以下五个参数：

process.roles=broker,controller
node.id=1 注意不同的节点id号不同
controller.quorum.voters=controller.quorum.voters=1@localhost:9093 修改为 controller.quorum.voters=controller.quorum.voters=1@lihaozhe01:9093,2@lihaozhe02:9093,3@lihaozhe03:9093
advertised.listeners=PLAINTEXT://localhost:9092 修改为 advertised.listeners=PLAINTEXT://lihaozhe01:9092
log.dirs=/tmp/kraft-combined-logs 修改为 log.dirs= /home/lhz/opt/kafka/kraft-combined-logs

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

#
# This configuration file is intended for use in KRaft mode, where
# Apache ZooKeeper is not present.
#

############################# Server Basics #############################

# The role of this server. Setting this puts us in KRaft mode
process.roles=broker,controller

# The node id associated with this instance's roles
node.id=1

# The connect string for the controller quorum
controller.quorum.voters=1@lihaozhe:9093

############################# Socket Server Settings #############################

# The address the socket server listens on.
# Combined nodes (i.e. those with `process.roles=broker,controller`) must list the controller listener here at a minimum.
# If the broker listener is not defined, the default listener will use a host name that is equal to the value of java.net.InetAddress.getCanonicalHostName(),
# with PLAINTEXT listener name, and port 9092.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://:9092,CONTROLLER://:9093

# Name of listener used for communication between brokers.
inter.broker.listener.name=PLAINTEXT

# Listener name, hostname and port the broker or the controller will advertise to clients.
# If not set, it uses the value for "listeners".
advertised.listeners=PLAINTEXT://lihaozhe:9092,CONTROLLER://lihaozhe:9093

# A comma-separated list of the names of the listeners used by the controller.
# If no explicit mapping set in `listener.security.protocol.map`, default will be using PLAINTEXT protocol
# This is required if running in KRaft mode.
controller.listener.names=CONTROLLER

# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# The number of threads that the server uses for receiving requests from the network and sending responses to the network
num.network.threads=3

# The number of threads that the server uses for processing requests, which may include disk I/O
num.io.threads=8

# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400

# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400

# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600


############################# Log Basics #############################

# A comma separated list of directories under which to store log files
log.dirs=/home/lhz/data/kraft-combined-logs

# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=1

# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1

############################# Internal Topic Settings  #############################
# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

############################# Log Flush Policy #############################

# Messages are immediately written to the filesystem but by default we only fsync() to sync
# the OS cache lazily. The following configurations control the flush of data to disk.
# There are a few important trade-offs here:
#    1. Durability: Unflushed data may be lost if you are not using replication.
#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.
# The settings below allow one to configure the flush policy to flush data after a period of time or
# every N messages (or both). This can be done globally and overridden on a per-topic basis.

# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000

# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000

############################# Log Retention Policy #############################

# The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.

# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168

# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824

# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824

# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000

分发配置到其它节点

scp -r  /home/lhz/opt/kafka root@lihaozhe02: /home/lhz/opt
scp -r  /home/lhz/opt/kafka root@lihaozhe03: /home/lhz/opt

scp /home/lhz/.bashrc root@lihaozhe02:/etc
scp /home/lhz/.bashrc root@lihaozhe03:/etc

在所有节点刷新环境变量

source /home/lhz/.bashrc

初始化集群数据目录

生成存储目录唯一 ID

kafka-storage.sh random-uuid

生成结果：

CHQOTE5dRxaKL1QvP2raKg

用该 ID 格式化 kafka 存储目录

注意：在每个节点都要执行一次

kafka-storage.sh format -t CHQOTE5dRxaKL1QvP2raKg \
-c  /home/lhz/opt/kafka/config/kraft/server.properties

执行结果：

Formatting metadata directory  /home/lhz/opt/kafka/kraft-combined-logs with metadata.version 3.9-IV0

启动停止

在每个节点分别启动

kafka-server-start.sh -daemon  /home/lhz/opt/kafka/config/kraft/server.properties

kafka-server-stop.sh

启动脚本

vim  /home/lhz/opt/kafka/bin/kafka-kraft-service

#!/bin/bash

case $1 in
"start"){
	echo  ------------- kafka 启动 ------------
	kafka-server-start.sh -daemon /home/lhz/opt/kafka/config/kraft/server.properties
}
;;
"stop"){
	echo  ------------- kafka 停止 ------------
	kafka-server-stop.sh
}
esac

chmode 655  /home/lhz/opt/kafka/bin/kafka-kraft-service.sh

查看全文

http://www.kler.cn/a/453810.html

【089】基于51单片机环境监测系统【Proteus仿真+Keil程序+报告+原理图】

Kalilinux下MySQL的安装

python+requests接口自动化测试框架实例详解

ubuntu 网络管理--NetworkManager

蓝牙BLE开发——解决iOS设备获取MAC方式

音视频入门基础：MPEG2-TS专题（22）——FFmpeg源码中，获取TS流的音频信息的实现

传统网络架构与SDN架构对比

Qt笔记：网络编程UDP

智慧交通-Android车牌识别接口-车牌识别系统

【提审】Android包提审报权限问题

excel技巧：excel文件怎么加密防止泄密？加密Excel文件的四种方法

微机接口课设——基于Proteus和8086的打地鼠设计（8255、8253、8259）Proteus中Unknown 1-byte opcode / Unknown 2-byte opcode错误

【Compose multiplatform教程09】【组件】Image组件

Effective C++ 条款 04：确定对象被使用前已先被初始化

flask后端开发（9）：ORM模型外键+迁移ORM模型

Java重要面试名词整理（七）：分库分表

redis使用注意哪些事项

深入理解Nginx工作原理及优化技巧

子网掩码计算route命令

Spark常用的转化操作和动作操作详解

Linux系统编程——理解系统内核中的信号捕获

深度学习-76-大模型量化之压缩映射方法和量化校准方法简介

mybatis SqlSessionFactory

Java配置文件的使用-相同信息在不同环境的赋值访问

解决在windows中mysql安装服务后启动服务失败的问题

求两大数和

deepin 安装 kafka

zookeeper模式

创建软件目录

下载

解压

修改目录名称

配置环境变量

修改配置文件

启动停止

启动脚本

kraft模式

创建软件目录

下载

解压

修改目录名称

配置环境变量

修改配置文件

分发配置到其它节点

初始化集群数据目录

生成存储目录唯一 ID

用该 ID 格式化 kafka 存储目录

启动停止

启动脚本

相关文章：