| 服务器 | hostname | 软件 | | --- | --- |--- | | 192.168.235.127| demo01 | java,zk,kafka| | 192.168.235.128| demo02 | java,zk,kafka| | 192.168.235.129| demo03 | java,zk,kafka | # zk ## 配置 /etc/hosts * 192.168.235.127 ``` [root@demo01 ~]# vi /etc/hosts 127.0.0.1 localhost localhost.localdomain localhost4 localhost4.localdomain4 ::1 localhost localhost.localdomain localhost6 localhost6.localdomain6 192.168.235.127 demo01 192.168.235.128 demo02 192.168.235.129 demo03 192.168.235.127 zookeeper01 192.168.235.128 zookeeper02 192.168.235.129 zookeeper03 ``` * 192.168.235.128 ``` [root@demo02 ~]# vi /etc/hosts 127.0.0.1 localhost localhost.localdomain localhost4 localhost4.localdomain4 ::1 localhost localhost.localdomain localhost6 localhost6.localdomain6 192.168.235.127 demo01 192.168.235.128 demo02 192.168.235.129 demo03 192.168.235.127 zookeeper01 192.168.235.128 zookeeper02 192.168.235.129 zookeeper03 ``` * 192.168.235.129 ``` [root@demo03 ~]# vi /etc/hosts 127.0.0.1 localhost localhost.localdomain localhost4 localhost4.localdomain4 ::1 localhost localhost.localdomain localhost6 localhost6.localdomain6 192.168.235.127 demo01 192.168.235.128 demo02 192.168.235.129 demo03 192.168.235.127 zookeeper01 192.168.235.128 zookeeper02 192.168.235.129 zookeeper03 ``` ## 创建软件目录 192.168.235.127，192.168.235.128，192.168.235.128 分别创建目录，解决zk软件包 ``` mkdir -p /root/zookeeper tar -zxvf zookeeper-3.4.8.tar.gz -C /root/zookeeper ``` ## 配置myid * 192.168.235.127 ``` mkdir -p /root/kafka/data/zookeeper [root@demo01 zookeeper]# vi myid 1 ``` * 192.168.235.128 ``` mkdir -p /root/kafka/data/zookeeper [root@demo02 zookeeper]# vi myid 2 ``` * 192.168.235.129 ``` mkdir -p /root/kafka/data/zookeeper [root@demo03 zookeeper]# vi myid 3 ``` ## 配置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=/root/kafka/data/zookeeper # 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=zookeeper01:2888:3888 server.2=zookeeper02:2888:3888 server.3=zookeeper03:2888:3888 ``` ## 启动 192.168.235.127 ``` [root@demo01 zookeeper]# cd /root/zookeeper && bin/zkServer.sh start ZooKeeper JMX enabled by default Using config: /root/zookeeper/bin/../conf/zoo.cfg Starting zookeeper ... STARTED ``` 192.168.235.128 ``` [root@demo02 zookeeper]# cd /root/zookeeper && bin/zkServer.sh start ZooKeeper JMX enabled by default Using config: /root/zookeeper/bin/../conf/zoo.cfg Starting zookeeper ... STARTED ``` 192.168.235.129 ``` [root@demo03 zookeeper]# cd /root/zookeeper && bin/zkServer.sh start ZooKeeper JMX enabled by default Using config: /root/zookeeper/bin/../conf/zoo.cfg Starting zookeeper ... STARTED [root@demo03 zookeeper]# ``` ## 查看 ``` [root@demo01 zookeeper]# cd /root/zookeeper && bin/zkServer.sh status ZooKeeper JMX enabled by default Using config: /root/zookeeper/bin/../conf/zoo.cfg Mode: follower [root@demo02 zookeeper]# cd /root/zookeeper && bin/zkServer.sh status ZooKeeper JMX enabled by default Using config: /root/zookeeper/bin/../conf/zoo.cfg Mode: follower [root@demo03 zookeeper]# cd /root/zookeeper && bin/zkServer.sh status ZooKeeper JMX enabled by default Using config: /root/zookeeper/bin/../conf/zoo.cfg Mode: leader ``` ## 停止 ``` [root@demo03 zookeeper]# cd /root/zookeeper && bin/zkServer.sh stop ``` # kafka ## 下载软件 ``` mkdir /root/kafka tar – zxvf kafka_2.11-0.11.0.0.tgz -C /root/kafka mkdir -p /root/kafka/data/kafka-logs ```  ## 配置 /root/kafka/config/server.properties * 192.168.235.127 ``` [root@demo01 config]# vi /root/kafka/config/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=1 # Switch to enable topic deletion or not, default value is false #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://zookeeper01:9092 # The maximum amount of time a message can sit in a log before we force a flush #log.flush.interval.ms=1000 # A comma seperated list of directories under which to store log files log.dirs=/root/kafka/data/kafka-logs ############################# 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 as long as the remaining # segments don't 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=zookeeper01:2181,zookeeper02:2181,zookeeper03: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 ``` * 192.168.235.128 ``` # 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=2 # Switch to enable topic deletion or not, default value is false #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://zookeeper02:9092 # The maximum amount of time a message can sit in a log before we force a flush #log.flush.interval.ms=1000 # A comma seperated list of directories under which to store log files log.dirs=/root/kafka/data/kafka-logs ############################# 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 as long as the remaining # segments don't 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=zookeeper01:2181,zookeeper02:2181,zookeeper03: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 ``` * 192.168.235.129 ``` # 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=3 # Switch to enable topic deletion or not, default value is false #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://zookeeper03:9092 # The maximum amount of time a message can sit in a log before we force a flush #log.flush.interval.ms=1000 # A comma seperated list of directories under which to store log files log.dirs=/root/kafka/data/kafka-logs ############################# 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 as long as the remaining # segments don't 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=zookeeper01:2181,zookeeper02:2181,zookeeper03: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 ``` ## 配置 meta.properties * 192.168.235.127 ``` [root@demo01 kafka-logs]# vi /root/kafka/data/kafka-logs/meta.properties version=0 broker.id=1 ``` * 192.168.235.128 ``` [root@demo02 zookeeper]# vi /root/kafka/data/kafka-logs/meta.properties version=0 broker.id=2 ``` * 192.168.235.129 ``` [root@demo03 zookeeper]# vi /root/kafka/data/kafka-logs/meta.properties version=0 broker.id=3 ``` ## 启动 ``` [root@demo01 ~]# cd /root/kafka/ && bin/kafka-server-start.sh -daemon config/server.properties & [1] 1964 [root@demo01 ~]# jps 2254 Jps 1759 QuorumPeerMain 2191 Kafka [root@demo01 ~]# jps 2264 Jps 1759 QuorumPeerMain 2191 Kafka ``` ## 测试 ``` [root@demo01 config]# cd /root/kafka/ && bin/kafka-topics.sh --create --zookeeper zookeeper01:2181,zookeeper02:2181,zookeeper03:2181 --replication-factor 3 --partitions 3 --topic owen Created topic "owen". [root@demo01 kafka]# bin/kafka-topics.sh --describe --zookeeper zookeeper01:2181,zookeeper02:2181,zookeeper03:2181 --topic owen Topic:owen PartitionCount:3 ReplicationFactor:3 Configs: Topic: owen Partition: 0 Leader: 3 Replicas: 3,2,1 Isr: 3 Topic: owen Partition: 1 Leader: 1 Replicas: 1,3,2 Isr: 1,3,2 Topic: owen Partition: 2 Leader: 2 Replicas: 2,1,3 Isr: 2,1,3 ``` ## 消费消息 ``` [root@demo02 bin]# kafka-console-consumer.sh --bootstrap-server zookeeper03:9092,zookeeper03:9092,zookeeper03:9092 --from-beginning --topic example ``` ## 监控 KafkaOffsetMonitor-assembly-0.2.0.jar 上传至/root/kafka/libs ``` [root@demo01 libs]# java -cp ../libs/KafkaOffsetMonitor-assembly-0.2.0.jar com.quantifind.kafka.offsetapp.OffsetGetterWeb --zk zookeeper01:2181,zookeeper02:2181,zookeeper03:2181 --port 8089 --refresh 10.seconds --retain 2.days serving resources from: jar:file:/root/kafka/libs/KafkaOffsetMonitor-assembly-0.2.0.jar!/offsetapp SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder". SLF4J: Defaulting to no-operation (NOP) logger implementation SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details. 2020-05-28 13:20:39.506:INFO:oejs.Server:jetty-7.x.y-SNAPSHOT log4j:WARN No appenders could be found for logger (org.I0Itec.zkclient.ZkConnection). log4j:WARN Please initialize the log4j system properly. log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info. ```