（三）kafka基本使用

[TOC]

版本： zookeeper: zookeeper-3.4.12.tar.gz kafka: kafka_2.11-2.0.0.tgz

參考：

• ZooKeeper Getting Started Guide
• Kafka Documentation

1. zookeeper

[root@thor zookeeper-3.4.12]# bin/zkCli.sh -server 127.0.0.1:2181
Connecting to 127.0.0.1:2181
2018-09-12 14:24:21,455 [myid:] - INFO  [main:Environment@100] - Client environment:zookeeper.version=3.4.12-e5259e437540f349646870ea94dc2658c4e44b3b, built on 03/27/2018 03:55 GMT
2018-09-12 14:24:21,460 [myid:] - INFO  [main:Environment@100] - Client environment:host.name=Thor
2018-09-12 14:24:21,460 [myid:] - INFO  [main:Environment@100] - Client environment:java.version=1.8.0_162
2018-09-12 14:24:21,462 [myid:] - INFO  [main:Environment@100] - Client environment:java.vendor=Oracle Corporation
2018-09-12 14:24:21,462 [myid:] - INFO  [main:Environment@100] - Client environment:java.home=/usr/local/jdk1.7.0_10/jre
2018-09-12 14:24:21,462 [myid:] - INFO  [main:Environment@100] - Client environment:java.class.path=/mnt/liudong/lean/zookeeper-3.4.12/bin/../build/classes:/mnt/liudong/lean/zookeeper-3.4.12/bin/../build/lib/*.jar:/mnt/liudong/lean/zookeeper-3.4.12/bin/../lib/slf4j-log4j12-1.7.25.jar:/mnt/liudong/lean/zookeeper-3.4.12/bin/../lib/slf4j-api-1.7.25.jar:/mnt/liudong/lean/zookeeper-3.4.12/bin/../lib/netty-3.10.6.Final.jar:/mnt/liudong/lean/zookeeper-3.4.12/bin/../lib/log4j-1.2.17.jar:/mnt/liudong/lean/zookeeper-3.4.12/bin/../lib/jline-0.9.94.jar:/mnt/liudong/lean/zookeeper-3.4.12/bin/../lib/audience-annotations-0.5.0.jar:/mnt/liudong/lean/zookeeper-3.4.12/bin/../zookeeper-3.4.12.jar:/mnt/liudong/lean/zookeeper-3.4.12/bin/../src/java/lib/*.jar:/mnt/liudong/lean/zookeeper-3.4.12/bin/../conf:/usr/local/jdk1.7.0_10/lib:/usr/local/jdk1.7.0_10/jre/lib:/usr/local/jdk1.7.0_10/lib:/usr/local/jdk1.7.0_10/jre/lib:
2018-09-12 14:24:21,462 [myid:] - INFO  [main:Environment@100] - Client environment:java.library.path=../lib:/usr/java/packages/lib/amd64:/usr/lib64:/lib64:/lib:/usr/lib
2018-09-12 14:24:21,462 [myid:] - INFO  [main:Environment@100] - Client environment:java.io.tmpdir=/tmp
2018-09-12 14:24:21,462 [myid:] - INFO  [main:Environment@100] - Client environment:java.compiler=<NA>
2018-09-12 14:24:21,462 [myid:] - INFO  [main:Environment@100] - Client environment:os.name=Linux
2018-09-12 14:24:21,463 [myid:] - INFO  [main:Environment@100] - Client environment:os.arch=amd64
2018-09-12 14:24:21,463 [myid:] - INFO  [main:Environment@100] - Client environment:os.version=3.10.0-123.el7.x86_64
2018-09-12 14:24:21,463 [myid:] - INFO  [main:Environment@100] - Client environment:user.name=root
2018-09-12 14:24:21,463 [myid:] - INFO  [main:Environment@100] - Client environment:user.home=/root
2018-09-12 14:24:21,463 [myid:] - INFO  [main:Environment@100] - Client environment:user.dir=/mnt/liudong/lean/zookeeper-3.4.12
2018-09-12 14:24:21,464 [myid:] - INFO  [main:ZooKeeper@441] - Initiating client connection, connectString=127.0.0.1:2181 sessionTimeout=30000 watcher=org.apache.zookeeper.ZooKeeperMain$MyWatcher@4b85612c
Welcome to ZooKeeper!
2018-09-12 14:24:21,492 [myid:] - INFO  [main-SendThread(127.0.0.1:2181):ClientCnxn$SendThread@1028] - Opening socket connection to server 127.0.0.1/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error)
JLine support is enabled
2018-09-12 14:24:21,559 [myid:] - INFO  [main-SendThread(127.0.0.1:2181):ClientCnxn$SendThread@878] - Socket connection established to 127.0.0.1/127.0.0.1:2181, initiating session
[zk: 127.0.0.1:2181(CONNECTING) 0] 2018-09-12 14:24:21,596 [myid:] - INFO  [main-SendThread(127.0.0.1:2181):ClientCnxn$SendThread@1302] - Session establishment complete on server 127.0.0.1/127.0.0.1:2181, sessionid = 0x100191eca480000, negotiated timeout = 30000

WATCHER::

WatchedEvent state:SyncConnected type:None path:null

[zk: 127.0.0.1:2181(CONNECTED) 0] help
ZooKeeper -server host:port cmd args
        stat path [watch]
        set path data [version]
        ls path [watch]
        delquota [-n|-b] path
        ls2 path [watch]
        setAcl path acl
        setquota -n|-b val path
        history 
        redo cmdno
        printwatches on|off
        delete path [version]
        sync path
        listquota path
        rmr path
        get path [watch]
        create [-s] [-e] path data acl
        addauth scheme auth
        quit 
        getAcl path
        close 
        connect host:port
[zk: 127.0.0.1:2181(CONNECTED) 1] ls .
Command failed: java.lang.IllegalArgumentException: Path must start with / character
[zk: 127.0.0.1:2181(CONNECTED) 2] ls ./
Command failed: java.lang.IllegalArgumentException: Path must start with / character
[zk: 127.0.0.1:2181(CONNECTED) 3] ls
[zk: 127.0.0.1:2181(CONNECTED) 4] ls /
[zookeeper]
[zk: 127.0.0.1:2181(CONNECTED) 5] ls /zookeeper
[quota]
[zk: 127.0.0.1:2181(CONNECTED) 6] create /zk_test mydata
Created /zk_test
[zk: 127.0.0.1:2181(CONNECTED) 7] ls
[zk: 127.0.0.1:2181(CONNECTED) 8] ls
[zk: 127.0.0.1:2181(CONNECTED) 9] ls /
[zookeeper, zk_test]
[zk: 127.0.0.1:2181(CONNECTED) 10] get /zk_test
mydata
cZxid = 0x2
ctime = Wed Sep 12 14:27:46 CST 2018
mZxid = 0x2
mtime = Wed Sep 12 14:27:46 CST 2018
pZxid = 0x2
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0
[zk: 127.0.0.1:2181(CONNECTED) 11] set /zk_test junk
cZxid = 0x2
ctime = Wed Sep 12 14:27:46 CST 2018
mZxid = 0x3
mtime = Wed Sep 12 14:28:36 CST 2018
pZxid = 0x2
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 0
[zk: 127.0.0.1:2181(CONNECTED) 12] get /zk_test
junk
cZxid = 0x2
ctime = Wed Sep 12 14:27:46 CST 2018
mZxid = 0x3
mtime = Wed Sep 12 14:28:36 CST 2018
pZxid = 0x2
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 0
[zk: 127.0.0.1:2181(CONNECTED) 13] delete /zk_test
[zk: 127.0.0.1:2181(CONNECTED) 14] ld /
ZooKeeper -server host:port cmd args
        stat path [watch]
        set path data [version]
        ls path [watch]
        delquota [-n|-b] path
        ls2 path [watch]
        setAcl path acl
        setquota -n|-b val path
        history 
        redo cmdno
        printwatches on|off
        delete path [version]
        sync path
        listquota path
        rmr path
        get path [watch]
        create [-s] [-e] path data acl
        addauth scheme auth
        quit 
        getAcl path
        close 
        connect host:port
[zk: 127.0.0.1:2181(CONNECTED) 15] ls /
[zookeeper]
[zk: 127.0.0.1:2181(CONNECTED) 16] get /zookeeper

cZxid = 0x0
ctime = Thu Jan 01 08:00:00 CST 1970
mZxid = 0x0
mtime = Thu Jan 01 08:00:00 CST 1970
pZxid = 0x0
cversion = -1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1
[zk: 127.0.0.1:2181(CONNECTED) 17]

2. kafka

> bin/kafka-server-start.sh config/server.properties
[2013-04-22 15:01:47,028] INFO Verifying properties (kafka.utils.VerifiableProperties)
[2013-04-22 15:01:47,051] INFO Property socket.send.buffer.bytes is overridden to 1048576 (kafka.utils.VerifiableProperties)
...

3. topic

[root@thor kafka_2.11-2.0.0]# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
Created topic "test".
[root@thor kafka_2.11-2.0.0]# bin/kafka-topics.sh --list --zookeeper localhost:2181                                                     
test

4. producer

[root@thor kafka_2.11-2.0.0]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>this is a test
>this is a ^Hnothre test
>this is anathor test
>this is another test
>hello
>world
>hehe
>

5. consumer

[root@thor kafka_2.11-2.0.0]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
this is a test
this is anothre test
this is anathor test
this is another test
hello
world
world
hehe

6. windows下使用Kafka

C:\kafka_2.11-2.0.0>.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

C:\kafka_2.11-2.0.0>.\bin\windows\kafka-server-start.bat .\config\server.properties


C:\kafka_2.11-2.0.0\bin\windows>kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic topic1992

C:\kafka_2.11-2.0.0\bin\windows>kafka-topics.bat --list --zookeeper localhost:2181

C:\kafka_2.11-2.0.0\bin\windows>.\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic topic1992 --from-beginning