概念

• producer : 消息生产者，发布消息到 kafka 集群的终端或服务。

• consumer : 从 kafka 集群中消费消息的终端或服务。

• Consumer group : high-level consumer API 中，每个 consumer 都属于一个 consumer group，每条消息只能被 consumer group 中的一个 Consumer 消费，但可以被多个 consumer group 消费。

• topic： 每条发布到 kafka 集群的消息属于的类别，即 kafka 是面向 topic 的。

• broker：kafka 集群中包含的服务器。

• partition：partition 是物理上的概念，每个 topic 包含一个或多个 partition。kafka 分配的单位是 partition。

• replica： partition 的副本，保障 partition 的高可用。

kafka27140952.png

kafka配置文件

# 核心配置
broker.id=0

############### Socket Server Settings ##############
# 监听端口
listeners=PLAINTEXT://:9092
# 配置服务提供远端访问能力
advertised.listeners=PLAINTEXT://192.168.199.101:9092
# 配置Https的连接
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
# server用来处理网络请求的网络线程数目；一般你不需要更改这个属性。
num.network.threads=3
# server用来处理请求的I/O线程的数目；这个线程数目至少要等于硬盘的个数。
num.io.threads=8
# SO_SNDBUFF 缓存大小，server进行socket 连接所用
socket.send.buffer.bytes=102400
# SO_RCVBUFF缓存大小，server进行socket连接时所用
socket.receive.buffer.bytes=102400
# server允许的最大请求尺寸； 这将避免server溢出，它应该小于Java heap size
socket.request.max.bytes=10485760

################ Log Basics ####################
# 此目录每次重启会被清理，测试用就不改了
log.dirs=/tmp/kafka-logs
# 如果创建topic时没有给出划分partitions个数，这个数字将是topic下partitions数目的默认数值。
num.partitions=1
num.recovery.threads.per.data.dir=1

################# Internal Topic Settings #################
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

################# Log Flush Policy ###################
# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000
# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000

################# Log Retention Policy ##################
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

################# Zookeeper #################
zookeeper.connect=localhost:2181
# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000

############ Group Coordinator Settings ############
group.initial.rebalance.delay.ms=0

例子 （kafka_2.11-0.10.1.1）

# Licensed to the Apache Software Foundation (ASF) 
############ Server Basics ###################

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1

# Switch to enable topic deletion or not, default value is false
#delete.topic.enable=true

############## Socket Server Settings ##################

# The address the socket server listens on. It will get the value returned from 
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = security_protocol://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://192.168.199.101:9091

# The number of threads handling network requests
num.network.threads=3

# The number of threads doing disk I/O
num.io.threads=8

# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400

# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600

############# Log Basics ####################

# A comma seperated list of directories under which to store log files
log.dirs=/home/kfk/ka1_logs

# the brokers.
num.partitions=1

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

################# Log Flush Policy ###################

# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000

############## Log Retention Policy #################
# The minimum age of a log file to be eligible for deletion
log.retention.hours=168

# The maximum size of a log segment file.
log.segment.bytes=1073741824

# to the retention policies
log.retention.check.interval.ms=300000

################ Zookeeper ################
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# root directory for all kafka znodes.
zookeeper.connect=192.168.199.101:2181,192.168.199.101:2182,192.168.199.101:2183

# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000

启动命令

#!/bin/bash JMX_PORT=9981 bin/kafka-server-start.sh config/server1.properties >/dev/null 2>&1 &

停止命令

ps -ef |grep kafka |grep java|awk '{print $2}' |xargs kill -9

---

常用操作

创建topic test1

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

查看topic列表

bin/kafka-topics.sh --list --zookeeper localhost:2181

查看一个 topic 的分区及副本状态信息。

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test1

启动生产者

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test1

启动消费者

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test1 --from-beginning

**在生产者发送的消息，会在消费者接收到，用于测试基础功能

创建3个副本的topic

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

将已有的topic修改为3个分区

kafka-topics.sh --alter --zookeeper localhost:2181 --topic test1 --partitions 3

zookeeper

配置文件实例 (zookeeper-3.4.14)

tickTime=2000

initLimit=10

syncLimit=5

dataDir=/home/zkdata

dtaLogDir=/home/zkdata/logs

clientPort=2181

server.1=192.168.199.101:2881:3881
server.2=192.168.199.101:2882:3882
server.3=192.168.199.101:2883:3883

启动命令

sh /opt/zookeeper1/bin/zkServer.sh start

start/status/stop/restart

连接服务器

zkCli.sh -server 127.0.0.1:2181

常用命令

1.ls -- 查看某个目录包含的所有文件，例如：
[zk: 127.0.0.1:2181(CONNECTED) 1] ls /2.ls2 -- 查看某个目录包含的所有文件，
与ls不同的是它查看到time、version等信息，例如：
[zk: 127.0.0.1:2181(CONNECTED) 1] ls2 /3.create -- 创建znode，并设置初始内容，例如：
[zk: 127.0.0.1:2181(CONNECTED) 1] create /test "test" 
Created /test创建一个新的 znode节点“ test ”以及与它关联的字符串4.get -- 获取znode的数据，
如下：
[zk: 127.0.0.1:2181(CONNECTED) 1] get /test5.set -- 修改znode内容，例如：
[zk: 127.0.0.1:2181(CONNECTED) 1] set /test "ricky"6.delete -- 删除znode，例如：
[zk: 127.0.0.1:2181(CONNECTED) 1] delete /test7.quit -- 退出客户端8.help -- 帮助命令

zookeeper 四字命令 (服务器)

conf  输出相关服务配置的详细信息 
cons  列出所有连接到服务器的客户端的完全的连接 / 会话的详细信息。
      包括“接受 / 发送”的包数量、会话 id 、操作延迟、最后的操作执行等等信息 
dump  列出未经处理的会话和临时节点。 
envi  输出关于服务环境的详细信息（区别于 conf 命令）。 
reqs  列出未经处理的请求 ruok 测试服务是否处于正确状态。
      如果确实如此，那么服务返回“ imok ”，否则不做任何相应 stat 输出关于性能和连接的客户端的列表。 
wchs  列出服务器 watch 的详细信息 
wchc  通过 session 列出服务器 watch 的详细信息，它的输出是一个与 watch 相关的会话的列表 
      wchp 通过路径列出服务器 watch 的详细信息。它输出一个与 session 相关的路径