惠州百度seo哪家好,长沙seo外包平台,杭州优化外包,长沙logo设计公司哪家好些Kafka 部署
Kafka 部署说明
kafka 版本选择 kafka 基于scala语言实现,所以使用kafka需要指定scala的相应的版本.kafka 为多个版本的Scala构建。这仅在使用 Scala 时才重要#xff0c;并且希望为使用的相同 Scala 版本构建一个版本。否则#xff0c;任何版本都可以
kafka下…Kafka 部署
Kafka 部署说明
kafka 版本选择 kafka 基于scala语言实现,所以使用kafka需要指定scala的相应的版本.kafka 为多个版本的Scala构建。这仅在使用 Scala 时才重要并且希望为使用的相同 Scala 版本构建一个版本。否则任何版本都可以
kafka下载链接:Apache KafkaApache Kafka: A Distributed Streaming Platform.http://kafka.apache.org/downloads
kafka版本格式 kafka_scala 版本_kafka 版本 #示例:kafka_2.13-2.7.0.tgz scala 语言官网: https://www.scala-lang.org/ scale 与 java关系: https://baike.baidu.com/item/Scala/2462287?fraladdin Kafka 支持单机和集群部署生产通常为集群模式 官方文档 Apache Kafka 单机部署
Download the latest Kafka release and extract it $ apt update apt -y install openjdk-8-jdk $ tar -xzf kafka_2.13-3.4.0.tgz $ cd kafka_2.13-3.4.0 NOTE: Your local environment must have Java 8 installed. Apache Kafka can be started using ZooKeeper or KRaft. To get started with either configuration follow one the sections below but not both. Kafka with ZooKeeper Run the following commands in order to start all services in the correct order:
# Start the ZooKeeper service
$ bin/zookeeper-server-start.sh config/zookeeper.properties
Open another terminal session and run:
# Start the Kafka broker service
$ bin/kafka-server-start.sh config/server.properties 集群部署
环境准备 ZooKeeper
当前版本 Kafka 依赖 Zookeeper 服务,但以后将不再依赖 http://kafka.apache.org/quickstart # Note: Soon, ZooKeeper will no longer be required by Apache Kafka. 环境说明
#在三个Ubuntu18.04节点提前部署zookeeper和kafka三个节点复用
node1:10.0.0.101
node2:10.0.0.102
node3:10.0.0.103
#注意:生产中zookeeper和kafka一般是分开独立部署的,kafka安装前需要安装java环境 确保三个节点的zookeeper启动
[rootnode1 ~]#zkServer.sh status
/usr/bin/java
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: follower
[rootnode2 ~]#zkServer.sh status
/usr/bin/java
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: leader
[rootnode3 ~]#zkServer.sh status
/usr/bin/java
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: follower
各节点部署 Kafka
Kafka 节点配置
配置文件说明
#配置文件 ./conf/server.properties内容说明
############################# Server Basics###############################
# broker的id值为整数且必须唯一在一个集群中不能重复
broker.id1
############################# Socket ServerSettings ######################
# kafka监听端口默认9092
listenersPLAINTEXT://10.0.0.101:9092
# 处理网络请求的线程数量默认为3个
num.network.threads3
# 执行磁盘IO操作的线程数量默认为8个
num.io.threads8
# socket服务发送数据的缓冲区大小默认100KB
socket.send.buffer.bytes102400
# socket服务接受数据的缓冲区大小默认100KB
socket.receive.buffer.bytes102400
# socket服务所能接受的一个请求的最大大小默认为100M
socket.request.max.bytes104857600
############################# Log Basics###################################
# kafka存储消息数据的目录
log.dirs../data
# 每个topic默认的partition
num.partitions1
# 设置副本数量为3,当Leader的Replication故障会进行故障自动转移。
default.replication.factor3
# 在启动时恢复数据和关闭时刷新数据时每个数据目录的线程数量
num.recovery.threads.per.data.dir1
############################# Log FlushPolicy #############################
# 消息刷新到磁盘中的消息条数阈值
log.flush.interval.messages10000
# 消息刷新到磁盘中的最大时间间隔,1s
log.flush.interval.ms1000
############################# Log RetentionPolicy #########################
# 日志保留小时数超时会自动删除默认为7天
log.retention.hours168
# 日志保留大小超出大小会自动删除默认为1G
#log.retention.bytes1073741824
# 日志分片策略单个日志文件的大小最大为1G超出后则创建一个新的日志文件
log.segment.bytes1073741824
# 每隔多长时间检测数据是否达到删除条件,300s
log.retention.check.interval.ms300000
############################# Zookeeper ####################################
# Zookeeper连接信息如果是zookeeper集群则以逗号隔开
zookeeper.connect10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181
# 连接zookeeper的超时时间,6s
zookeeper.connection.timeout.ms6000
# 是否允许删除topic默认为falsetopic只会标记为marked for deletion
delete.topic.enabletrue
范例
#在所有节点上执行安装java
[rootnode1 ~]#apt install openjdk-8-jdk -y
#在所有节点上执行下载官方下载
[rootnode1 ~]#wget https://downloads.apache.org/kafka/3.3.1/kafka_2.13-3.3.1.tgz
[rootnode1 ~]#wget https://archive.apache.org/dist/kafka/2.7.0/kafka_2.13-2.7.0.tgz
#解压缩
[rootnode1 ~]#tar xf kafka_2.13-2.7.0.tgz -C /usr/local/
[rootnode1 ~]#ln -s /usr/local/kafka_2.13-2.7.0/ /usr/local/kafka
#国内镜像下载
[rootnode1 ~]#wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/3.3.1/kafka_2.13-3.3.1.tgz
[rootnode1 ~]#wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/3.0.0/kafka_2.13-3.0.0.tgz
[rootnode1 ~]#wget https://mirror.bit.edu.cn/apache/kafka/2.7.0/kafka_2.13-2.7.0.tgz
#配置PATH变量
[rootnode1 ~]#echo PATH/usr/local/kafka/bin:$PATH /etc/profile.d/kafka.sh
[rootnode1 ~]#. /etc/profile.d/kafka.sh
#修改配置文件
[rootnode1 ~]#vim /usr/local/kafka/config/server.properties
broker.id1 #每个broker在集群中每个节点的正整数唯一标识此值保存在log.dirs下的meta.properties文件
listenersPLAINTEXT://10.0.0.101:9092 #指定当前主机的IP做为监听地址,注意:不支持0.0.0.0
log.dirs/usr/local/kafka/data #kakfa用于保存数据的目录所有的消息都会存储在该目录当中
num.partitions1 #设置创建新的topic时默认分区数量,建议和kafka的节点数量一致
default.replication.factor3 #指定默认的副本数为3可以实现故障的自动转移
log.retention.hours168 #设置kafka中消息保留时间默认为168小时即7天
zookeeper.connect10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181 #指定连接的zk的地址,zk中存储了broker的元数据信息
zookeeper.connection.timeout.ms6000 #设置连接zookeeper的超时时间单位为ms,默认6秒钟
#准备数据目录
[rootnode1 ~]#mkdir /usr/local/kafka/data
[rootnode1 ~]#scp /usr/local/kafka/config/server.properties 10.0.0.102:/usr/local/kafka/config
[rootnode1 ~]#scp /usr/local/kafka/config/server.properties 10.0.0.103:/usr/local/kafka/config
#修改第2个节点配置
[rootnode2 ~]#vim /usr/local/kafka/config/server.properties
broker.id2 #每个broker 在集群中的唯一标识正整数。
listenersPLAINTEXT://10.0.0.102:9092 #指定当前主机的IP做为监听地址,注意:不支持0.0.0.0
#修改第3个节点配置
[rootnode3 ~]#vim /usr/local/kafka/config/server.properties
broker.id3 #每个broker 在集群中的唯一标识正整数。
listenersPLAINTEXT://10.0.0.103:9092 #指定当前主机的IP做为监听地址,注意:不支持0.0.0.0
#可以调整内存
[rootnode1 ~]#vim /usr/local/kafka/bin/kafka-server-start.sh
......
if[ x$KAFKA_HEAP_OPTSx] ; then
export KAFKA_HEAP_OPTS -Xmx1G -Xms1G
fi...... 启动服务
在所有kafka节点执行下面操作 [rootnode1 ~]#kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties 确保服务启动状态
[rootnode1 ~]#ss -ntl|grep 9092LISTEN 0 50[::ffff:10.0.0.101]:9092 *:*
[rootnode1 ~]#tail /usr/local/kafka/logs/server.log
[2021-02-16 12:10:01,276] INFO [ExpirationReaper-1-AlterAcls]: Starting
(kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2021-02-16 12:10:01,311] INFO [/config/changes-event-process-thread]: Starting
(kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
[2021-02-16 12:10:01,332] INFO [SocketServer brokerId1] Starting socket server acceptors and processors
(kafka.network.SocketServer)
[2021-02-16 12:10:01,339] INFO [SocketServer brokerId1] Started data-plane acceptor and processor(s) for endpoint : ListenerName(PLAINTEXT) (kafka.network.SocketServer)
[2021-02-16 12:10:01,340] INFO [SocketServer brokerId1] Started socket server acceptors and processors
(kafka.network.SocketServer)
[2021-02-16 12:10:01,344] INFO Kafka version: 2.7.0 (org.apache.kafka.common.utils.AppInfoParser)
[2021-02-16 12:10:01,344] INFO Kafka commitId: 448719dc99a19793 (org.apache.kafka.common.utils.AppInfoParser)
[2021-02-16 12:10:01,344] INFO Kafka startTimeMs: 1613448601340 (org.apache.kafka.common.utils.AppInfoParser)
[2021-02-16 12:10:01,346] INFO [KafkaServer id1] started (kafka.server.KafkaServer)
[2021-02-16 12:10:01,391] INFO [broker-1-to-controller-send-thread]: Recorded new controller, from now on will
use broker 1 (kafka.server.BrokerToControllerRequestThread)
#如果使用id,还需要修改/usr/local/kafka/data/meta.properties
#打开zooinspector可以看到三个id Broker 依赖于 Zookeeper每个Broker 的id 和 Topic、Partition这些元数据信息都会写入Zookeeper 的 ZNode 节点中 consumer 依赖于ZookeeperConsumer 在消费消息时每消费完一条消息会将产生的offset 保存到 Zookeeper 中下次消费在当前offset往后继续消费.kafka0.9 之前Consumer 的offset 存储在 Zookeeper 中kafka0.9 以后offset存储在本地。 Partition 依赖于 ZookeeperPartition 完成Replication 备份后选举出一个Leader这个是依托于 Zookeeper 的选举机制实现的
准备Kafka的service文件 [rootnode1 ~]#cat /lib/systemd/system/kafka.service
[unit]
DescriptionApache kafka
Afternetwork.target
[service]
Typesimple
#EnvironmentJAVA_HOME/data/server/java
PIDFile/usr/local/kafka/kafka.pid
Execstart/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server. properties
Execstop/bin/kill -TERM ${MAINPID}
Restartalways
RestartSec20
[Install]
wantedBymulti-user.target
[rootnode1 ~]#systemctl daemon-load
[rootnode1 ~]#systemctl restart kafka.service
一键部署kafka集群
#!/bin/bashKAFKA_VERSION3.9.0
#KAFKA_VERSION3.4.0
#KAFKA_VERSION3.3.2
#KAFKA_VERSION3.2.0
SCALA_VERSION2.13
#KAFKA_VERSION-3.0.0
KAFKA_URLhttps://mirrors.tuna.tsinghua.edu.cn/apache/kafka/${KAFKA_VERSION}/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz
#KAFKA_URLhttps://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.8.1/kafka_2.13-2.8.1.tgz
#KAFKA_URLhttps://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.7.1/kafka_2.13-2.7.1.tgz
ZK_VERSOIN3.8.1
#ZK_VERSOIN3.7.1
#ZK_VERSOIN3.6.3
ZK_URLhttps://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/stable/apache-zookeeper-${ZK_VERSOIN}-bin.tar.gzZK_INSTALL_DIR/usr/local/zookeeper
KAFKA_INSTALL_DIR/usr/local/kafkaNODE110.0.0.187
NODE210.0.0.188
NODE310.0.0.189HOSThostname -I|awk {print $1}
. /etc/os-releasecolor () {RES_COL60MOVE_TO_COLecho -en \\033[${RES_COL}GSETCOLOR_SUCCESSecho -en \\033[1;32mSETCOLOR_FAILUREecho -en \\033[1;31mSETCOLOR_WARNINGecho -en \\033[1;33mSETCOLOR_NORMALecho -en \E[0mecho -n $1 $MOVE_TO_COLecho -n [if [ $2 success -o $2 0 ] ;then${SETCOLOR_SUCCESS}echo -n $ OK elif [ $2 failure -o $2 1 ] ;then ${SETCOLOR_FAILURE}echo -n $FAILEDelse${SETCOLOR_WARNING}echo -n $WARNINGfi${SETCOLOR_NORMAL}echo -n ]echo
}install_jdk() {if [ $ID centos -o $ID rocky ];thenyum -y install java-1.8.0-openjdk-devel || { color 安装JDK失败! 1; exit 1; }elseapt updateapt install openjdk-8-jdk -y || { color 安装JDK失败! 1; exit 1; } fijava -version
}zk_myid () {read -p 请输入node编号(默认为 1): MYIDif [ -z $MYID ] ;thenMYID1elif [[ ! $MYID ~ ^[0-9]$ ]];thencolor 请输入正确的node编号! 1exitelsetruefi
}install_zookeeper() {wget -P /usr/local/src/ $ZK_URL || { color 下载失败! 1 ;exit ; }tar xf /usr/local/src/${ZK_URL##*/} -C dirname ${ZK_INSTALL_DIR}ln -s /usr/local/apache-zookeeper-*-bin/ ${ZK_INSTALL_DIR}echo PATH${ZK_INSTALL_DIR}/bin:$PATH /etc/profile.d/zookeeper.sh. /etc/profile.d/zookeeper.shmkdir -p ${ZK_INSTALL_DIR}/data echo $MYID ${ZK_INSTALL_DIR}/data/myidcat ${ZK_INSTALL_DIR}/conf/zoo.cfg EOF
tickTime2000
initLimit10
syncLimit5
dataDir${ZK_INSTALL_DIR}/data
clientPort2181
maxClientCnxns128
autopurge.snapRetainCount3
autopurge.purgeInterval24
server.1${NODE1}:2888:3888
server.2${NODE2}:2888:3888
server.3${NODE3}:2888:3888
EOFcat /lib/systemd/system/zookeeper.service EOF
[Unit]
Descriptionzookeeper.service
Afternetwork.target[Service]
Typeforking
#Environment${ZK_INSTALL_DIR}
ExecStart${ZK_INSTALL_DIR}/bin/zkServer.sh start
ExecStop${ZK_INSTALL_DIR}/bin/zkServer.sh stop
ExecReload${ZK_INSTALL_DIR}/bin/zkServer.sh restart[Install]
WantedBymulti-user.target
EOFsystemctl daemon-reloadsystemctl enable --now zookeeper.servicesystemctl is-active zookeeper.serviceif [ $? -eq 0 ] ;then color zookeeper 安装成功! 0 else color zookeeper 安装失败! 1exit 1fi
}install_kafka(){if [ ! -f kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz ];thenwget -P /usr/local/src $KAFKA_URL || { color 下载失败! 1 ;exit ; }elsecp kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz /usr/local/src/fitar xf /usr/local/src/${KAFKA_URL##*/} -C /usr/local/ln -s ${KAFKA_INSTALL_DIR}_*/ ${KAFKA_INSTALL_DIR}echo PATH${KAFKA_INSTALL_DIR}/bin:$PATH /etc/profile.d/kafka.sh. /etc/profile.d/kafka.shcat ${KAFKA_INSTALL_DIR}/config/server.properties EOF
broker.id$MYID
listenersPLAINTEXT://${HOST}:9092
log.dirs${KAFKA_INSTALL_DIR}/data
num.partitions1
log.retention.hours168
zookeeper.connect${NODE1}:2181,${NODE2}:2181,${NODE3}:2181
zookeeper.connection.timeout.ms6000
EOFmkdir ${KAFKA_INSTALL_DIR}/datacat /lib/systemd/system/kafka.service EOF
[Unit]
DescriptionApache kafka
Afternetwork.target[Service]
Typesimple
#EnvironmentJAVA_HOME/data/server/java
#PIDFile${KAFKA_INSTALL_DIR}/kafka.pid
ExecStart${KAFKA_INSTALL_DIR}/bin/kafka-server-start.sh ${KAFKA_INSTALL_DIR}/config/server.properties
ExecStop/bin/kill -TERM \${MAINPID}
Restartalways
RestartSec20[Install]
WantedBymulti-user.targetEOFsystemctl daemon-reloadsystemctl enable --now kafka.service#kafka-server-start.sh -daemon ${KAFKA_INSTALL_DIR}/config/server.properties systemctl is-active kafka.serviceif [ $? -eq 0 ] ;then color kafka 安装成功! 0 else color kafka 安装失败! 1exit 1fi
}zk_myid#install_jdk#install_zookeeperinstall_kafka
