进入kafka/bin目录下
./kafka-server-start.sh -daemon ../config/server.properties
验证是否启动成功：
进入到zk中的节点看id是 0 的broker有没有存在（上线）
ls /brokers/ids
server.properties核心配置详解：
| Property | Default | Description |
|---|---|---|
| broker.id | 0 | 每个broker都可以⽤⼀个唯⼀的⾮负整数id进⾏标识；这个id可以作为broker的“名字”，你可以选择任意你喜欢的数字作为id，只要id是唯⼀的即可。 |
| log.dirs | /tmp/kafka-logs | kafka存放数据的路径。这个路径并不是唯⼀的，可以是多个，路径之间只需要使⽤逗号分隔即可；每当创建新partition时，都会选择在包含最少partitions的路径下进⾏。 |
| listeners | PLAINTEXT://192.168.65.60:9092 | server接受客户端连接的端⼝，ip配置kafka本机ip即可 |
| zookeeper.connect | localhost:2181 | zooKeeper连接字符串的格式为：hostname:port，此处hostname和port分别是ZooKeeper集群中某个节点的host和port；zookeeper如果是集群，连接⽅式为hostname1:port1, hostname2:port2,hostname3:port3 |
| log.retention.hours | 168 | 每个⽇志⽂件删除之前保存的时间。默认数据保存时间对所有topic都⼀样。 |
| num.partitions | 1 | 创建topic的默认分区数 |
| default.replication.factor | 1 | ⾃动创建topic的默认副本数量，建议设置为⼤于等于2 |
| min.insync.replicas | 1 | 当producer设置acks为-1时，min.insync.replicas指定replicas的最⼩数⽬（必须确认每⼀个repica的写数据都是成功的），如果这个数⽬没有达到，producer发送消息会产⽣异常 |
| delete.topic.enable | false | 是否允许删除主题 |
执行以下命令创建名为“test”的topic，这个topic只有一个partition，并且备份因子也设置为1
注：新版本的kafka，已经不需要依赖zookeeper来创建topic，新版的kafka创建topic指令为下：
./kafka-topics.sh --create --bootstrap-server 192.168.1.106:9092 --replication-factor 1 --partitions 1 --topic test
查看当前kafka内有哪些topic
./kafka-topics.sh --list --bootstrap-server 192.168.1.106:90923. 发送消息
kafka自带了一个producer命令客户端，可以从本地文件中读取内容，或者我们也可以以命令行中直接输入内容，并将这些内容以消息的形式发送到kafka集群中。在默认情况下，每一个行会被当做成一个独立的消息。使用kafka的发送消息的客户端，指定发送到的kafka服务器地址和topic
./kafka-console-producer.sh --broker-list 192.168.1.106:9092 --topic test4.消费消息
对于consumer，kafka同样也携带了一个命令行客户端，会将获取到内容在命令中进行输
出， 默认是消费最新的消息 。使用kafka的消费者消息的客户端，从指定kafka服务器的指定
topic中消费消息
方式一：从最后一条消息的偏移量+1开始消费
./kafka-console-consumer.sh --bootstrap-server 192.168.1.106:9092 --topic test
方式二：从头开始消费
./kafka-console-consumer.sh --bootstrap-server 192.168.1.106:9092 --from-beginning --topic test几个注意点：
消息会被存储消息是顺序存储消息是有偏移量的消费时可以指明偏移量进行消费
