1.1 kafka搭建
Kafka部署
實驗環境
|
Centos |
Memory |
cpu |
|
7.6 |
2G |
2G |
|
kafka_2.11-1.1.1 zookeeper-3.4.14.tar |
||
aliyun雲伺服器
39.99.224.205 kafka01
39.99.132.23 kafka02
39.101.130.158 kafka03
#新增hosts檔案
[root@kafka01 ~]# cat <<END >>/etc/hosts
> 39.99.224.205 kafka01
> 39.99.132.23 kafka02
> 39.101.130.158 kafka03
> END
[root@kafka02~]# cat <<END >>/etc/hosts
> 39.99.224.205 kafka01
> 39.99.132.23 kafka02
> 39.101.130.158 kafka03
> END
[root@kafka03~]# cat <<END >>/etc/hosts
> 39.99.224.205 kafka01
> 39.99.132.23 kafka02
> 39.101.130.158 kafka03
> END
#新增金鑰
[root@kafka01 ~]# ssh-keygen -t rsa
Generating public/private rsa key pair.
Enter file in which to save the key (/root/.ssh/id_rsa):
Enter passphrase (empty for no passphrase):
Enter same passphrase again:
Your identification has been saved in /root/.ssh/id_rsa.
Your public key has been saved in /root/.ssh/id_rsa.pub.
The key fingerprint is:
SHA256:FApgI9Ew+XwjO+KXRjGDLoQ/c4sQF7lbWqZxnRT6Gjo root@kafka01
The key's randomart image is:
+---[RSA 2048]----+
|==+o. ... |
|.++. o.. . |
|.o.o.o... |
|ooB+B.o. |
|o+ &=.. S |
|+.Xo.o |
|ooE*o. |
| ..=. |
| o |
+----[SHA256]-----+
[root@kafka01 ~]# ssh-copy-id -i .ssh/id_rsa.pub kafka02:
[root@kafka01 ~]# ssh-copy-id -i .ssh/id_rsa.pub kafka03:
[root@kafka01 ~]# ssh-copy-id -i .ssh/id_rsa.pub kafka01:
#kafka02生成金鑰
[root@kafka02 ~]# ssh-keygen -t rsa
Generating public/private rsa key pair.
Enter file in which to save the key (/root/.ssh/id_rsa):
Enter passphrase (empty for no passphrase):
Enter same passphrase again:
Your identification has been saved in /root/.ssh/id_rsa.
Your public key has been saved in /root/.ssh/id_rsa.pub.
The key fingerprint is:
SHA256:CjjZVU5cs/3T1hFkhEyULdBxCaxTdy/momOAgg0P2Xs root@kafka02
The key's randomart image is:
+---[RSA 2048]----+
| .o.o.B=O*.|
| +. + Oo+o|
| o . . . + o.o|
| ++.. o .o.+|
| +*o. .S .oo.+|
| ..=.E.. . .o |
| o. . . . |
| + |
| . . |
+----[SHA256]-----+
[root@kafka02 ~]# ssh-copy-id -i .ssh/id_rsa.pub kafka01:
[root@kafka02 ~]# ssh-copy-id -i .ssh/id_rsa.pub kafka02:
[root@kafka02 ~]# ssh-copy-id -i .ssh/id_rsa.pub kafka03:
#kafka03生成金鑰
[root@kafka023~]# ssh-keygen -t rsa
Generating public/private rsa key pair.
Enter file in which to save the key (/root/.ssh/id_rsa):
Enter passphrase (empty for no passphrase):
Enter same passphrase again:
Your identification has been saved in /root/.ssh/id_rsa.
Your public key has been saved in /root/.ssh/id_rsa.pub.
The key fingerprint is:
SHA256:CjjZVU5cs/3T1hFkhEyULdBxCaxTdy/momOAgg0P2Xs root@kafka02
The key's randomart image is:
+---[RSA 2048]----+
| .o.o.B=O*.|
| +. + Oo+o|
| o . . . + o.o|
| ++.. o .o.+|
| +*o. .S .oo.+|
| ..=.E.. . .o |
| o. . . . |
| + |
| . . |
+----[SHA256]-----+
[root@kafka03 ~]# ssh-copy-id -i .ssh/id_rsa.pub kafka01:
[root@kafka03~]# ssh-copy-id -i .ssh/id_rsa.pub kafka02:
[root@kafka03~]# ssh-copy-id -i .ssh/id_rsa.pub kafka03:
#上傳軟體
[root@kafka01 ~]# scp zookeeper-3.4.14.tar.gz kafka_2.11-1.1.1.tgzjdk-8u221-linux-x64.tar.gzkafka02:
[root@kafka01 ~]# scp zookeeper-3.4.14.tar.gz kafka_2.11-1.1.1.tgzjdk-8u221-linux-x64.tar.gzkafka03:
#解壓jdk
[root@kafka01 ~]# tar xf jdk-8u221-linux-x64.tar.gz -C /usr/local/
[root@kafka01 ~]# cd /usr/local/
[root@kafka01 local]# ls
aegis etc include lib libexec share zookeeper
bin games jdk1.8.0_221 lib64 sbin src
[root@kafka01 local]# mv jdk1.8.0_221 java
[root@kafka01 local]# vim /etc/profile.d/java.sh
[root@kafka01 local]# cat /etc/profile.d/java.sh
export JAVA_HOME=/usr/local/java
export PATH=$PATH:$JAVA_HOME/bin
[root@kafka01 ~]# . /etc/profile.d/java.sh
[root@kafka01 ~]# java -version
java version "1.8.0_221"
Java(TM) SE Runtime Environment (build 1.8.0_221-b11)
Java HotSpot(TM) 64-Bit Server VM (build 25.221-b11, mixed mode
Kafka02 kafka03同上
#解壓zookeeper
[root@kafka01 ~]# tar xf zookeeper-3.4.14.tar.gz -C /usr/local/
[root@kafka01 ~]# cd /usr/local/
[root@kafka01 local]# ls
aegis bin etc games include lib lib64 libexec sbin share src zookeeper-3.4.14
[root@kafka01 local]# mv zookeeper-3.4.14 zookeeper
[root@kafka01 local]# cd zookeeper/conf/
[root@kafka01 conf]# ls
configuration.xsl log4j.properties zoo_sample.cfg
[root@kafka01 conf]# cp zoo_sample.cfg zoo.cfg
[root@kafka01 conf]# cat zoo.cfg
|
# 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/date # 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 server.1=kafka01:2888:3888 server.2=kafka02:2888:3888 server.3=kafka03:2888:3888 quorumListenOnAllIPs=true |
[root@kafka01zookeeper]# mkdir date
[root@kafka01 zookeeper]# echo "1" >/date/myid
#kafka02步驟同上
[root@kafka02~]# tar xf zookeeper-3.4.14.tar.gz -C /usr/local/
[root@kafka02~]# cd /usr/local/
[root@kafka01 local]# mv zookeeper-3.4.14 zookeeper
[root@kafka02local]# cd zookeeper/conf/
[root@kafka02conf]# ls
configuration.xsl log4j.properties zoo_sample.cfg
[root@kafka02conf]# cp zoo_sample.cfg zoo.cfg
[root@kafka02 conf]# cat zoo.cfg
|
# 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/date # 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 server.1=kafka01:2888:3888 server.2=kafka02:2888:3888 server.3=kafka03:2888:3888 quorumListenOnAllIPs=true |
[root@kafka02zookeeper]# mkdir date
[root@kafka02zookeeper]# echo "2" >/date/myid
#kafka03步驟同上
前段略
[root@kafka03zookeeper]# mkdir date
[root@kafka03zookeeper]# echo "3" >/date/myid
#啟動zookeeper
#kafka01
[root@kafka01 zookeeper]# ./bin/zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
#kafka02
[root@kafka02zookeeper]# ./bin/zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
#kafka03
[root@kafka03zookeeper]# ./bin/zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
#檢視狀態
[root@kafka01 zookeeper]# ./bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Mode: follower
[root@kafka02 zookeeper]# ./bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Mode: follower
[root@kafka03 zookeeper]# ./bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Mode: leader
#新增全域性變數
[root@kafka01 zookeeper]# echo "PATH=$PATH:/usr/local/zookeeper/bin/" >>/etc/profile
[root@kafka02zookeeper]# echo "PATH=$PATH:/usr/local/zookeeper/bin/" >>/etc/profile
[root@kafka03zookeeper]# echo "PATH=$PATH:/usr/local/zookeeper/bin/" >>/etc/profile
部署kafka
[root@kafka01 ~]# tar xf kafka_2.11-1.1.1.tgz -C /usr/local/
[root@kafka01 ~]# cd /usr/local/
[root@kafka01 local]# ls
aegis etc include kafka_2.11-1.1.1 lib64 sbin src
bin games java lib libexec share zookeeper
[root@kafka01 local]# mv kafka_2.11-1.1.1 kafka
[root@kafka01 local]# cd kafka/
[root@kafka01 kafka]# ls
bin config libs LICENSE NOTICE site-docs
[root@kafka01 kafka]# cd config/
[root@kafka01 config]# ls
connect-console-sink.properties consumer.properties
connect-console-source.properties log4j.properties
connect-distributed.properties producer.properties
connect-file-sink.properties server.properties
connect-file-source.properties tools-log4j.properties
connect-log4j.properties zookeeper.properties
connect-standalone.properties
[root@kafka01 config]# cp server.properties{,.bak}
[root@kafka01 config]# vim server.properties
|
# 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. # 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 delete.topic.enable=true ############################# Socket Server Settings ############################# # The address the socket server listens on. It will get the value returned from # java.net.InetAddress.getCanonicalHostName() if not configured. # FORMAT: # listeners = listener_name://host_name:port # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 listeners=PLAINTEXT://:9092 # Hostname and port the broker will advertise to producers and consumers. If not set, # it uses the value for "listeners" if configured. Otherwise, it will use the value # returned from java.net.InetAddress.getCanonicalHostName(). #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=/u01/data/kafka # 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 for 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=kafka01:2181,kafka02:2181,kafka03:2181 # Timeout in ms for connecting to zookeeper zookeeper.connection.timeout.ms=6000 ############################# 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 |
Kafka02 kafka03 解壓步驟同上
[root@kafka01 config]# scp server.properties kafka02:/usr/local/kafka/config/server.properties 100% 6899 4.7MB/s 00:00
[root@kafka01 config]# scp server.properties kafka03:/usr/local/kafka/config/
server.properties 100% 6899 5.2MB/s 00:00
#修改Kafka02和kafka03scp過去的檔案
[root@kafka02 config]# vim server.properties
|
broker.id=2 |
[root@kafka03 config]# vim server.properties
|
broker.id=3 |
#新增環境變數
[root@kafka01 ~]# echo "PATH=$PATH:/usr/local/bin/" >>/etc/profile
[root@kafka01 ~]# . /etc/profile
Kafka02 kafka03 步驟同上
#啟動kafka
[root@kafka01 ~]# kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
[root@kafka02~]# kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
[root@kafka03~]# kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
[root@kafka01 ~]# jps
11923 QuorumPeerMain
14325 Jps
14251 Kafka
[root@kafka02 ~]# jps
14594 Jps
11814 QuorumPeerMain
14524 Kafka
[root@kafka03 ~]# jps
12177 QuorumPeerMain
13106 Kafka
13461 Jps
註釋
1）Producer ：訊息生產者，就是向 kafka broker 發訊息的客戶端；
2）Consumer ：訊息消費者，向 kafka broker 取訊息的客戶端；
3）Consumer Group （CG）：消費者組，由多個 consumer 組成。消費者組內每個消費者負
責消費不同分割槽的資料，一個分割槽只能由一個組內消費者消費；消費者組之間互不影響。所
有的消費者都屬於某個消費者組，即消費者組是邏輯上的一個訂閱者。
4）Broker ：一臺 kafka 伺服器就是一個 broker。一個叢集由多個 broker 組成。一個 broker
可以容納多個 topic。
5）Topic ：可以理解為一個佇列，生產者和消費者面向的都是一個 topic；
6）Partition：為了實現擴充套件性，一個非常大的 topic 可以分佈到多個 broker（即伺服器）上，
一個 topic 可以分為多個 partition，每個 partition 是一個有序的佇列；
7）Replica：副本，為保證叢集中的某個節點發生故障時，該節點上的 partition 資料不丟失，
且 kafka 仍然能夠繼續工作，kafka 提供了副本機制，一個 topic 的每個分割槽都有若干個副本， 一個 leader 和若干個 follower。
8）leader：每個分割槽多個副本的“主”，生產者傳送資料的物件，以及消費者消費資料的對
象都是 leader。
9）follower：每個分割槽多個副本中的“從”，實時從 leader 中同步資料，保持和 leader 資料
的同步。leader 發生故障時，某個 follower 會成為新的 follower。
測試
#新增topic
[root@kafka01 ~]# kafka-topics.sh --zookeeper kafka01:2181 --create --replication-factor 3 --partitions 1 --topic dunimo
Created topic "dunimo".
#檢視topic
[root@kafka01 ~]# kafka-topics.sh --zookeeper kafka01:2181 --list
dunimo
[root@kafka01 ~]# kafka-topics.sh --zookeeper kafka02:2181 --create --replication-factor 2 --partitions 2 --topic cdb_gk
[root@kafka01 ~]# kafka-topics.sh --zookeeper kafka01:2181 --list
cdb_gk
Dunimo
#生產者生產訊息
[root@kafka02 ~]# kafka-console-producer.sh --broker-list kafka01:9092 --topic dunimo03
>hello
>yangdongyang
>
消費者消費訊息
[root@kafka01 ~]# kafka-console-consumer.sh --bootstrap-server kafka01:9092 --topic dunimo03
hello
yangdongyang
#檢視所有訊息
[root@kafka03 ~]# kafka-console-consumer.sh --bootstrap-server kafka01:9092 --from-beginning --topic dunimo03
yangdongyang
Hello
#刪除訊息
[root@kafka02 ~]# kafka-topics.sh --zookeeper kafka01:2181 --delete --topic dunim
o01
Topic dunimo01 is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.
[root@kafka02 ~]# kafka-topics.sh --zookeeper kafka01:2181 --list
__consumer_offsets
dunimo02
dunimo03
#檢視一個詳細topic
[root@kafka02 ~]# kafka-topics.sh --zookeeper kafka01:2181 --describe --topic dunimo03
Topic:dunimo03 PartitionCount:2 ReplicationFactor:3 Configs:
Topic: dunimo03 Partition: 0 Leader: 0 Replicas: 0,3,2 Isr: 0,3,2
Topic: dunimo03 Partition: 1 Leader: 2 Replicas: 2,0,3 Isr: 2,0,3
#更改topic dunio03的分割槽
[root@kafka02 ~]# kafka-topics.sh --zookeeper kafka01:2181 --alter --topic dunimo03 --partitions 5
WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
Adding partitions succeeded!