1.环境介绍
当前构建的kafka集群IP地址和端口:
172.19.32.68:9292,172.19.32.69:9292,172.19.32.70:9292
采用固定ip的方式访问
当前环境需要组件包含：3个kafka节点、1个KafkaManager、3个zookeeper；
2.集群搭建过程
2.1首先构建zookeeper集群
具体参考zookeeper集群构建(已有相应的具体附件)；
2.1 构建kafka集群
（注意：前提是zookeeper搭建并无误之后，因为kafka依赖于zookeeper）
kafka集群的构建json脚本，如下所示：
（1）kafka01
{
"env": {
"HOSTNAME_COMMAND": "ip route | awk '/172.19.32.0/{print $7}'",
"KAFKA_MESSAGE_MAX_BYTES": "100000000",
"KAFKA_ADVERTISED_LISTENERS": "PLAINTEXT://_{HOSTNAME_COMMAND}:9292",
"KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS": "3",
"KAFKA_LISTENERS": "PLAINTEXT://0.0.0.0:9292",
"TZ": "Asia/Shanghai",
"KAFKA_ZOOKEEPER_CONNECT": "cluster-poc-trial-production- zk.marathon.l4lb.thisdcos.directory:21815/trial-kafkacluster",
"KAFKA_NUM_PARTITIONS": "3",
"KAFKA_DEFAULT_REPLICATION_FACTOR": "3",
"KAFKA_BROKER_ID": "0",
"JMX_PORT": "29998",
"KAFKA_REPLICA_FETCH_MAX_BYTES": "100000000",
"KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR": "3"
},
"id": "/cms-poc/trial-production/components/msg-queue/nodes-kafka/trial-kafkacluster/kafka01",
"backoffFactor": 1.15,
"backoffSeconds": 1,
"constraints": [
["hostname", "IS", "172.19.32.68"]
],
"container": {
"type": "DOCKER",
"volumes": [{
"containerPath": "/kafka",
"hostPath": "/data/poc/trial-production/kafka-cluster/trial-broker-data",
"mode": "RW"
}, {
"containerPath": "/etc/localtime",
"hostPath": "/etc/localtime",
"mode": "RO"
}],
"docker": {
"image": "registry.marathon.l4lb.thisdcos.directory:5000/wurstmeister-kafka:2.12-2.2.0",
"forcePullImage": false,
"privileged": false,
"parameters": []
}
},
"cpus": 1,
"disk": 0,
"instances": 1,
"maxLaunchDelaySeconds": 3600,
"mem": 2048,
"gpus": 0,
"networks": [{
"mode": "host"
}],
"portDefinitions": [{
"labels": {
"VIP_0": "trial-kafkacluster-kafka01:9292"
},
"name": "kafka-endpoint-fct",
"protocol": "tcp",
"port": 9292
}, {
"labels": {
"VIP_1": "/cms-poc/trial-production/components/msg-queue/nodes-kafka/trial- kafkacluster/kafka01:29998"
},
"name": "kafka-monitor-endpoint1",
"protocol": "tcp",
"port": 29998
}],
"requirePorts": true,
"upgradeStrategy": {
"maximumOverCapacity": 1,
"minimumHealthCapacity": 1
},
"killSelection": "YOUNGEST_FIRST",
"unreachableStrategy": {
"inactiveAfterSeconds": 0,
"expungeAfterSeconds": 0
},
"healthChecks": [],
"fetch": []
}
（2）Kafka02
{ "env": { "HOSTNAME_COMMAND": "ip route | awk '/172.19.32.0/{print $7}'", "KAFKA_MESSAGE_MAX_BYTES": "100000000", "KAFKA_ADVERTISED_LISTENERS": "PLAINTEXT://_{HOSTNAME_COMMAND}:9292", "KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS": "3", "KAFKA_LISTENERS": "PLAINTEXT://0.0.0.0:9292", "TZ": "Asia/Shanghai", "KAFKA_ZOOKEEPER_CONNECT": "cluster-poc-trial-production- zk.marathon.l4lb.thisdcos.directory:21815/trial-kafkacluster", "KAFKA_NUM_PARTITIONS": "3", "KAFKA_DEFAULT_REPLICATION_FACTOR": "3", "KAFKA_BROKER_ID": "1", "JMX_PORT": "29998", "KAFKA_REPLICA_FETCH_MAX_BYTES": "100000000", "KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR": "3" }, "id": "/cms-poc/trial-production/components/msg-queue/nodes-kafka/trial-kafkacluster/kafka02", "backoffFactor": 1.15, "backoffSeconds": 1, "constraints": [ ["hostname", "IS", "172.19.32.69"] ], "container": { "type": "DOCKER", "volumes": [{ "containerPath": "/kafka", "hostPath": "/data/poc/trial-production/kafka-cluster/trial-broker-data", "mode": "RW" }, { "containerPath": "/etc/localtime", "hostPath": "/etc/localtime", "mode": "RO" }], "docker": { "image": "registry.marathon.l4lb.thisdcos.directory:5000/wurstmeister-kafka:2.12-2.2.0", "forcePullImage": false, "privileged": false, "parameters": [] } }, "cpus": 1, "disk": 0, "instances": 1, "maxLaunchDelaySeconds": 3600, "mem": 2048, "gpus": 0, "networks": [{ "mode": "host" }], "portDefinitions": [{ "labels": { "VIP_0": " trial-kafkacluster-kafka02:9292" }, "name": "kafka-endpoint-fct", "protocol": "tcp", "port": 9292 }, { "labels": { "VIP_1": "/cms-poc/trial-production/components/msg-queue/nodes-kafka/trial- kafkacluster/kafka02:29998" }, "name": "kafka-monitor-endpoint1", "protocol": "tcp", "port": 29998 }], "requirePorts": true, "upgradeStrategy": { "maximumOverCapacity": 1, "minimumHealthCapacity": 1 }, "killSelection": "YOUNGEST_FIRST", "unreachableStrategy": { "inactiveAfterSeconds": 0, "expungeAfterSeconds": 0 }, "healthChecks": [], "fetch": [] }
（3）kafka03
{ "env": { "HOSTNAME_COMMAND": "ip route | awk '/172.19.32.0/{print $7}'", "KAFKA_MESSAGE_MAX_BYTES": "100000000", "KAFKA_ADVERTISED_LISTENERS": "PLAINTEXT://_{HOSTNAME_COMMAND}:9292", "KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS": "3", "KAFKA_LISTENERS": "PLAINTEXT://0.0.0.0:9292", "TZ": "Asia/Shanghai", "KAFKA_ZOOKEEPER_CONNECT": "cluster-poc-trial-production- zk.marathon.l4lb.thisdcos.directory:21815/trial-kafkacluster", "KAFKA_NUM_PARTITIONS": "3", "KAFKA_DEFAULT_REPLICATION_FACTOR": "3", "KAFKA_BROKER_ID": "2", "JMX_PORT": "29998", "KAFKA_REPLICA_FETCH_MAX_BYTES": "100000000", "KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR": "3" }, "id": "/cms-poc/trial-production/components/msg-queue/nodes-kafka/trial-kafkacluster/kafka- 03", "backoffFactor": 1.15, "backoffSeconds": 1, "constraints": [ ["hostname", "IS", "172.19.32.70"] ], "container": { "type": "DOCKER", "volumes": [{ "containerPath": "/kafka", "hostPath": "/data/poc/trial-production/kafka-cluster/trial-broker-data", "mode": "RW" }, { "containerPath": "/etc/localtime", "hostPath": "/etc/localtime", "mode": "RO" }], "docker": { "image": "registry.marathon.l4lb.thisdcos.directory:5000/wurstmeister-kafka:2.12-2.2.0", "forcePullImage": false, "privileged": false, "parameters": [] } }, "cpus": 1, "disk": 0, "instances": 1, "maxLaunchDelaySeconds": 3600, "mem": 2048, "gpus": 0, "networks": [{ "mode": "host" }], "portDefinitions": [{ "labels": { "VIP_0": " trial-kafkacluster-kafka03:9292" }, "name": "kafka-endpoint-fct", "protocol": "tcp", "port": 9292 }, { "labels": { "VIP_1": "/cms-poc/trial-production/components/msg-queue/nodes-kafka/trial- kafkacluster/kafka03:29998" }, "name": "kafka-monitor-endpoint1", "protocol": "tcp", "port": 29998 }], "requirePorts": true, "upgradeStrategy": { "maximumOverCapacity": 1, "minimumHealthCapacity": 1 }, "killSelection": "YOUNGEST_FIRST", "unreachableStrategy": { "inactiveAfterSeconds": 0, "expungeAfterSeconds": 0 }, "healthChecks": [], "fetch": [] }
2.3搭建kafkaManger
kafkaManger提供管理、查看kafka的topic的信息等功能；代替了shell界面命令行，提供了便捷的可视化操作。
{ "env": { "KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS": "3", "ZK_HOSTS": "cluster-poc-trial-production-zk.marathon.l4lb.thisdcos.directory:21815", "KAFKA_MANAGER_USERNAME": "admin", "KAFKA_MANAGER_PASSWORD": "123456", "KAFKA_DEFAULT_REPLICATION_FACTOR": "3", "KAFKA_MANAGER_AUTH_ENABLED": "true", "KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR": "3" }, "id": "/cms-poc/trial-production/components/msg-queue/nodes-kafka/kafka-manager", "backoffFactor": 1.15, "backoffSeconds": 1, "constraints": [ ["hostname", "IS", "172.19.32.68"] ], "container": { "portMappings": [{ "containerPort": 9000, "hostPort": 9082, "protocol": "tcp", "servicePort": 10102 }], "type": "DOCKER", "volumes": [], "docker": { "image": "registry.marathon.l4lb.thisdcos.directory:5000/hlebalbau-kafka-manager:2.0.0.2", "forcePullImage": false, "privileged": false, "parameters": [] } }, "cpus": 1, "disk": 0, "instances": 1, "maxLaunchDelaySeconds": 3600, "mem": 1024, "gpus": 0, "networks": [{ "mode": "container/bridge" }], "requirePorts": false, "upgradeStrategy": { "maximumOverCapacity": 1, "minimumHealthCapacity": 1 }, "killSelection": "YOUNGEST_FIRST", "unreachableStrategy": { "inactiveAfterSeconds": 0, "expungeAfterSeconds": 0 }, "healthChecks": [], "fetch": [] }
注意事项：此处修改同步修改kafkaManager的链接zookeeper的服务目录地址；
具体来说，环境变量需要修改对应的zookeeper地址；
"ZK_HOSTS": "cluster-poc-trial-production-zk.marathon.l4lb.thisdcos.directory:21815",
3.使用kafkaManager可视化UI管理kafka集群
访问地址: http://172.19.32.68:9082/