kafka速记

列所有topic

./kafka-topics.sh -zookeeper 135.251.103.162:2181 -describe

列特定topic

./kafka-topics.sh -zookeeper 135.251.103.162:2181 -describe -topic notif.ctrl

重新刷各topic的partition, replica配置

./kafka-reassign-partitions.sh -zookeeper 135.251.103.162:2181 -reassignment-json-file ./my-topic-conf/topic-conf.json -execute
其中./my-topic-conf/topic-conf.json的内容如下：

{
    "partitions":
    [
        {
            "topic": "alarm.raw",
            "partition": 0,
            "replicas": [0,1,2] 
        },
                {
            "topic": "alarm.raw",
            "partition": 1,
            "replicas": [0,1,2] 
        },
                {
            "topic": "alarm.raw",
            "partition": 2,
            "replicas": [0,1,2] 
        },
        {
            "topic": "alarm.rc.conn",
            "partition": 0,
            "replicas": [0,1,2] 
        },
        {
            "topic": "enms.notification",
            "partition": 0,
            "replicas": [0,1,2] 
        },
        {
            "topic": "event",
            "partition": 0,
            "replicas": [0,1,2] 
        },
        {
            "topic": "log",
            "partition": 0,
            "replicas": [0,1,2] 
        },
        {
            "topic": "notif.adp",
            "partition": 0,
            "replicas": [0,1,2] 
        },
        {
            "topic": "notif.ctrl",
            "partition": 0,
            "replicas": [0,1,2] 
        },
        {
            "topic": "notif.pmc",
            "partition": 0,
            "replicas": [0,1,2] 
        },
        {
            "topic": "takeover.map.och",
            "partition": 0,
            "replicas": [0,1,2] 
        },
                {
            "topic": "takeover.map.och",
            "partition": 1,
            "replicas": [0,1,2] 
        },
                {
            "topic": "takeover.map.och",
            "partition": 2,
            "replicas": [0,1,2] 
        },
        {
            "topic": "takeover.reduce",
            "partition": 0,
            "replicas": [0,1,2] 
        }
    ]
}

给__consumer_offsets增加replica

./kafka-reassign-partitions.sh -zookeeper 135.251.103.162:2181 -reassignment-json-file ./increase-consumer-offsets-replication-factor.json -execute
increase-consumer-offsets-replication-factor.json内容：

{"partitions":[
{"topic":"__consumer_offsets","partition":0,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":1,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":2,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":3,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":4,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":5,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":6,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":7,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":8,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":9,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":10,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":11,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":12,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":13,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":14,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":15,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":16,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":17,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":18,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":19,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":20,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":21,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":22,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":23,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":24,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":25,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":26,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":27,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":28,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":29,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":30,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":31,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":32,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":33,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":34,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":35,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":36,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":37,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":38,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":39,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":40,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":41,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":42,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":43,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":44,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":45,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":46,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":47,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":48,"replicas":[0,1,2]},
{"topic":"__consumer_offsets","partition":49,"replicas":[0,1,2]}]
}

kafka manager 启动

bin/kafka-manager -Dconfig.file=/path/to/application.conf -Dhttp.port=8080

启动Kafka

bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties

创建Topic

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

修改Topic的partition数量

./kafka-topics.sh -zookeeper 135.251.103.162:2181 --alter --topic alarm.raw --partitions 3

发送消息

bin/kafka-console-producer.sh --broker-list "135.251.103.162:9092,135.251.103.163:9092,135.251.103.164:9092" --topic test

启动consumer

bin/kafka-console-consumer.sh --bootstrap-server "135.251.103.162:9092,135.251.103.163:9092,135.251.103.164:9092" --topic test --from-beginning