#tar zxvf kafka_2.-2.1..tgz -C /data/kafka/
#ln -s kafka_2.12-2.1.0 kafka
#cd /data/kafka

启动zookeeper服务
#./bin/zookeeper-server-start.sh config/zookeeper.properties
启动kafka服务
#./bin/kafka-server-start.sh config/server.properties

[root@test kafka ::]#netstat -anltp|grep

tcp               0.0.0.0:            0.0.0.0:*               LISTEN      /java

tcp               127.0.1.1:          127.0.0.1:         ESTABLISHED /java

tcp               127.0.0.1:         127.0.1.1:          ESTABLISHED /java

tcp               127.0.1.1:          127.0.0.1:         ESTABLISHED /java

tcp               127.0.0.1:         127.0.1.1:          ESTABLISHED /java

tcp               127.0.1.1:          127.0.0.1:         ESTABLISHED /java

tcp               127.0.0.1:         127.0.1.1:          ESTABLISHED /java

tcp               127.0.0.1:         127.0.1.1:          CLOSE_WAIT  /java

tcp               127.0.0.1:         127.0.1.1:          ESTABLISHED /java

tcp               127.0.1.1:          127.0.0.1:         ESTABLISHED /java

1>>>  创建topic
#./bin/kafka-topics.sh --create --zookeeper localhost: --replication-factor  --partitions  --topic yjt

Created topic "yjt".
参数解释：
--create：创建。
--zookeeper： 指定zookeeper服务器，这里使用的是kafka自带的zookeeper服务器，所以使用localhost+端口的方式，如果是自己配置的zookeeper服务，这里需要使用主机+端口的方式。
--replication-factor：指定副本数量，主要是为了数据的冗余。
--partitions：指定主题的分区数量。
--topic：指定主题的名字，这里是yjt。

2>>> 查看当前以及存在的主题：

#./bin/kafka-topics.sh --list --zookeeper localhost:2181    #这里使用了--list选项。列出服务器上存在的topic。

__consumer_offsets
  connect-test
  my-topic
  test
  yjt

　3>>> 查看某一个topic的信息。

#./bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic yjt   #这里使用了--describe选项。
  Topic:yjt PartitionCount:1 ReplicationFactor:1 Configs:
  Topic: yjt Partition: 0 Leader: 0 Replicas: 0 Isr: 0

解释：
第一行显示的是所有分区的摘要，其次，每一行显示一个分区，这里分区为1，所有只显示了一行(既第二行)：
Topic：主题名字
PartitionCount：当前主题所有分区数量，刚刚在创建主题的时候，只指定了一个分区，所以，这里显示1。
ReplicationFactor：副本数量。同上。
第二行：
Leader：该节点负责该分区的所有的读和写，每个节点的leader都是随机选择的（注：针对于分区，也就是说，每一个分区都有一个leader，并且leader可能都不一样）。
Replicas：备份的节点列表，无论该节点是否是leader或者目前是否还活着，只是显示。
Isr（a set of in-sync replicas）也叫同步状态的副本集合：“同步备份”的节点列表，也就是活着的节点并且正在同步leader。

#./bin/kafka-console-producer.sh --broker-list localhost: --topic yjt   #使用--broker-list指定kafaka服务器的地址，这里的9092是kafka默认的端口，如果在配置文件里面有改动，这里修改成对应的端口就行。

>My name is yjt.

>I am xxx years old.

#./bin/kafka-console-consumer.sh --bootstrap-server localhost: --topic yjt --from-beginning #使用--bootstrap-server指定服务器，同时使用--from-beginning选项指定从头开始读取。

My name is yjt.

I am xxx years old.

#cp config/server.properties config/server-.properties

#cp config/server.properties config/server-.properties

config/server-.properties:

    broker.id= 　　　　　　　　# broker Id，全局唯一，不能重复。

    listeners=PLAINTEXT://:9093 #监听地址，也可以是listeners=PLAINTEXT://主机或者IP:9093这种方式。
    log.dir=/tmp/kafka-logs-  #日志文件

config/server-.properties: 
　　broker.id= 
　　listeners=PLAINTEXT://:9094  
　　log.dir=/tmp/kafka-logs-

#./bin/kafka-server-start.sh config/server-.properties &  #以后台启动，但是好像没什么卵用，控制台还是啪啪啪的输出一大堆信息，害的我开了好几个终端。

#./bin/kafka-server-start.sh config/server-.properties & > /dev/null 2>&1  #这种方式启动好像也是一样，不过没试nohup。

#./bin/kafka-topics.sh --create --zookeeper localhost: --replication-factor  --partitions  --topic yjt1

#./bin/kafka-topics.sh --describe --zookeeper localhost: --topic yjt1

Topic:yjt1    PartitionCount:    ReplicationFactor:    Configs:

Topic: yjt1    Partition:     Leader:     Replicas:     Isr:

Topic: yjt1    Partition:     Leader:     Replicas:     Isr:

Topic: yjt1    Partition:     Leader:     Replicas:     Isr:

Topic: yjt1    Partition:     Leader:     Replicas:     Isr: 2
上述信息具体参考上面的解释。

#pwd

/data/kafka/kafka
#echo -e "foo\nbar" > test.txt

#./bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

#cat test.sink.txt

foo

bar

#./bin/kafka-console-consumer.sh --bootstrap-server localhost: --topic connect-test --from-beginning

{"schema":{"type":"string","optional":false},"payload":"foo"}

{"schema":{"type":"string","optional":false},"payload":"bar"}

# egrep -v "(^#|^$)" zoo.cfg

tickTime=

initLimit=

syncLimit=

dataDir=/data1/hadoop/data/zookeeper/data

dataLogDir=/data1/hadoop/data/zookeeper/logs

clientPort=

server.=master::

server.=slave1::

server.=slave2::

#sed -i "s/zookeeper.connect=localhost:2181/zookeeper.connect=master:2181,slave1:2181,slave2:2181/g" config/server.properties

config/server.properties:

　　broker.id=0    #必须修改这个值，全局唯一。其他的值像端口，日志存储等，都可以使用默认的配置，像上面的在一台服务器启动多个实例的时候，需要修改下面的两个配置(端口和日志路径，防止冲突)，但是在完全分布式模式下，也可以不用修改。

　　listeners=PLAINTEXT://:9092

　　log.dir=/tmp/kafka-logs
    host.name=master   #主机名，每个server都不同，需要修改。

//当前机器在集群中的唯一标识，和zookeeper的myid性质一样

broker.id=

//当前kafka对外提供服务的端口默认是9092

port=

//这个参数默认是关闭的，在0.8.1有个bug，DNS解析问题，失败率的问题。

host.name=master

//这个是borker进行网络处理的线程数

num.network.threads=

//这个是borker进行I/O处理的线程数

num.io.threads=

//发送缓冲区buffer大小，数据不是一下子就发送的，先回存储到缓冲区了到达一定的大小后在发送，能提高性能

socket.send.buffer.bytes=

//kafka接收缓冲区大小，当数据到达一定大小后在序列化到磁盘

socket.receive.buffer.bytes=

//这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数，这个值不能超过java的堆栈大小

socket.request.max.bytes=

//消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的num.io.threads要大于这个目录的个数这个目录，

//如果配置多个目录，新创建的topic他把消息持久化的地方是，当前以逗号分割的目录中，那个分区数最少就放那一个

log.dirs=/tmp/kafka-logs

//默认的分区数，一个topic默认1个分区数

num.partitions=

//每个数据目录用来日志恢复的线程数目

num.recovery.threads.per.data.dir=

//默认消息的最大持久化时间，168小时，7天

log.retention.hours=

//这个参数是：因为kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件

log.segment.bytes=

//每隔300000毫秒去检查上面配置的log失效时间

log.retention.check.interval.ms=

//是否启用log压缩，一般不用启用，启用的话可以提高性能

log.cleaner.enable=false

//设置zookeeper的连接端口

zookeeper.connect=master:,slave1:,slave2:

//设置zookeeper的连接超时时间

zookeeper.connection.timeout.ms=

metadata.broker.list=master:9092,slave1:9092,slave2:9092

zookeeper.connect=master:2181,slave1:2181,slave2:2181