Kafka笔记
Posted 奋斗的程序猿
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Kafka笔记相关的知识，希望对你有一定的参考价值。
Kafka笔记
安装
安装依赖JDK
Java
Kafka的安装需要java环境
如没有Java 须先安装java
1、检查一下系统中的jdk版本
java –version
显示:
openjdk version "1.8.0_102"
OpenJDK Runtime Environment (build 1.8.0_102-b14)
OpenJDK 64-Bit Server VM (build 25.102-b14, mixed mode)
2、检测jdk安装包
rpm -qa | grep java
3. 卸载openjdk
yum remove *openjdk*
之后再次输入rpm -qa | grep java 查看卸载情况：
4. 安装新的jdk
首先到jdk官网上下载你想要的jdk版本，下载完成之后将需要安装的jdk安装包放到Linux系统指定的文件夹下，并且命令进入该文件夹下：
mkdir -p /usr/lib/jvm
tar -zxvf jdk-8u131-linux-x64.tar.gz -C /usr/lib/jvm
5.环境设置环境
vim /etc/profile
在最前面添加：
export JAVA_HOME=/usr/lib/jvm/jdk1.8.0_131
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH
6、执行profile文件
source /etc/profile
7、检查新安装的jdk
java –version
安装并简单启动kafka
下载
下载: http://kafka.apache.org/downloads
选中https://www.apache.org/dyn/closer.cgi?path=/kafka/1.1.0/kafka_2.12-1.1.0.tgz
解压
下载好kafka安装包后，将其解压到/usr/local目录下，删除压缩包
tar –zxvf kafka_2.11-1.1.0.tgz –C /usr/local
进入目录
cd /usr/local/kafka_2.11-1.1.0
启动zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties
默认使用的2181端口，可在配置文件修改。
启动kafka server
bin/kafka-server-start.sh config/server.properties
(非本地生产者和消费者访问Kafka，记得修改 config/server.properties中的listeners, 例如
listeners=PLAINTEXT://192.168.1.253:9092)
创建topic
创建一个名为"test"的Topic，只有一个分区和一个备份
bin/kafka-topics.sh --create --zookeeper 192.168.1.253:2181 --replication-factor 1 --partitions 1 --topic test
查看topic
bin/kafka-topics.sh --list --zookeeper 192.168.1.253:2181
发送消息
bin/kafka-console-producer.sh --broker-list 192.168.1.253:9092 --topic test
输入消息:
Aaaaaa
消费消息
bin/kafka-console-consumer.sh --zookeeper 192.168.1.253:2181 --topic test --from-beginning
如果你有2台不同的终端上运行上述命令，那么当你在运行生产者时，消费者就能消费到生产者发送的消息。
单机设置多个集群
到目前，我们只是单一的运行一个broker，没什么意思。对于Kafka，一个broker仅仅只是一个集群的大小，所有让我们多设几个broker。
首先为每个broker创建一个配置文件:
> cp config/server.properties config/server-1.properties
> cp config/server.properties config/server-2.properties
现在编辑这些新建的文件，设置以下属性：
config/server-1.properties:
broker.id=1
listeners=PLAINTEXT:// 192.168.1.253:9093
log.dir=/tmp/kafka-logs-1
config/server-2.properties:
broker.id=2
listeners=PLAINTEXT://192.168.1.253:9094
log.dir=/tmp/kafka-logs-2
broker.id是集群中每个节点的唯一且永久的名称，我们修改端口和日志目录是因为我们现在在同一台机器上运行，我们要防止broker在同一端口上注册和覆盖对方的数据。
我们已经运行了zookeeper和刚才的一个kafka节点，所有我们只需要在启动2个新的kafka节点。
> bin/kafka-server-start.sh config/server-1.properties &
...
> bin/kafka-server-start.sh config/server-2.properties &
...
现在，我们创建一个新topic，把备份设置为：3
bin/kafka-topics.sh --create --zookeeper 192.168.1.253:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic
好了，现在我们已经有了一个集群了，我们怎么知道每个集群在做什么呢？运行命令"describe topics"
bin/kafka-topics.sh --describe --zookeeper 192.168.1.253:2181 --topic my-replicated-topic
输出:
Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
Topic: my-replicated-topic Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
输出解释：第一行是所有分区的摘要，其次，每一行提供一个分区信息，因为我们只有一个分区，所以只有一行。
"leader"：该节点负责该分区的所有的读和写，每个节点的leader都是随机选择的。
"replicas"：备份的节点列表，无论该节点是否是leader或者目前是否还活着，只是显示。
"isr"："同步备份"的节点列表，也就是活着的节点并且正在同步leader。
我们运行这个命令，看看一开始我们创建的那个节点：
> bin/kafka-topics.sh --describe --zookeeper 192.168.1.253:2181 --topic test
Topic:test PartitionCount:1 ReplicationFactor:1 Configs:
Topic: test Partition: 0 Leader: 0 Replicas: 0 Isr: 0
这并不奇怪，刚才创建的主题没有Replicas，并且在服务器"0"上，我们创建它的时候，集群中只有一个服务器，所以是"0"。
让我们来发布一些信息在新的topic上：
> bin/kafka-console-producer.sh --broker-list 192.168.1.253:9092 --topic my-replicated-topic
...
my test message 1
my test message 2
^C
现在，消费这些消息。
> bin/kafka-console-consumer.sh --zookeeper 192.168.1.253:2181 --from-beginning --topic my-replicated-topic
...
my test message 1
my test message 2
^C
我们要测试集群的容错，kill掉leader，Broker1作为当前的leader，也就是kill掉Broker1。
> ps | grep server-1.properties
7564 ttys002 0:15.91 /System/Library/Frameworks/JavaVM.framework/Versions/1.6/Home/bin/java...
kill -9 7564
在Windows上使用：
> wmic process where "caption = \'java.exe\' and commandline like \'%server-1.properties%\'" get processid
ProcessId
6016
> taskkill /pid 6016 /f
备份节点之一成为新的leader，而broker1已经不在同步备份集合里了。
bin/kafka-topics.sh --describe --zookeeper 192.168.1.253:2181 --topic my-replicated-topic
但是，消息仍然没丢：
> bin/kafka-console-consumer.sh --zookeeper 192.168.1.253:2181 --from-beginning --topic my-replicated-topic
...
my test message 1
my test message 2
kafka常用命令
启动zookeeper bin/zookeeper-server-start.sh config/zookeeper.properties & 启动kafka bin/kafka-server-start.sh config/server.properties & 停止kafka bin/kafka-server-stop.sh 停止zookeeper bin/zookeeper-server-stop.sh 复制代码 启动 kafka bin/kafka-server-start.sh config/server.properties & 创建节点 bin/kafka-topics.sh --create --zookeeper 192.168.1.253:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic 查看节点 bin/kafka-topics.sh --describe --zookeeper 192.168.1.253:2181 --topic test bin/kafka-topics.sh --describe --zookeeper 192.168.1.253:2181 --topic my-replicated-topic 消费 bin/kafka-console-consumer.sh --zookeeper 192.168.1.253:2181 --from-beginning --topic my-replicated-topic 生产 bin/kafka-console-producer.sh --broker-list 192.168.1.253:9092 --topic my-replicated-topic 以下是kafka常用命令行总结： 0.查看有哪些主题： ./kafka-topics.sh --list --zookeeper 192.168.0.201:12181 1.查看topic的详细信息 ./kafka-topics.sh -zookeeper 127.0.0.1:2181 -describe -topic testKJ1 2、为topic增加副本 ./kafka-reassign-partitions.sh -zookeeper 127.0.0.1:2181 -reassignment-json-file json/partitions-to-move.json -execute 3、创建topic ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testKJ1 4、为topic增加partition ./bin/kafka-topics.sh –zookeeper 127.0.0.1:2181 –alter –partitions 20 –topic testKJ1 5、kafka生产者客户端命令 ./kafka-console-producer.sh --broker-list localhost:9092 --topic testKJ1 ./kafka-console-producer.sh --broker-list localhost:9092 --topic testKJ1 > .\\kafka-console-producer.bat kafka.tools.ConsoleProducer --broker-list 192.168.1.161:9092 --topic proccess_trading_end 6、kafka消费者客户端命令 ./kafka-console-consumer.sh -zookeeper localhost:2181 --from-beginning --topic testKJ1 > .\\kafka-console-consumer.bat -zookeeper 192.168.1.161:2181 --from-beginning --topic proccess_trading_end 7、kafka服务启动 ./kafka-server-start.sh -daemon ../config/server.properties 8、下线broker ./kafka-run-class.sh kafka.admin.ShutdownBroker --zookeeper 127.0.0.1:2181 --broker #brokerId# --num.retries 3 --retry.interval.ms 60 shutdown broker 9、删除topic ./kafka-run-class.sh kafka.admin.DeleteTopicCommand --topic testKJ1 --zookeeper 127.0.0.1:2181 ./kafka-topics.sh --zookeeper localhost:2181 --delete --topic testKJ1 ./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper 192.168.1.161:2181 --group defaultGroup --topic testKJ1 ./kafka-consumer-offset-checker.sh --zookeeper 192.168.0.201:12181 --group group1 --topic group1 10、查看consumer组内消费的offset ./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper 192.168.1.161:2181 --group defaultGroup --topic proccess_trading_end .\\kafka-consumer-groups.bat --new-consumer --bootstrap-server 192.168.1.161:9092 --describe --group defaultGroup ./kafka-consumer-offset-checker.sh --zookeeper 192.168.0.201:12181 --group group1 --topic group1 善于将复杂问题简单化
以上是关于Kafka笔记的主要内容，如果未能解决你的问题，请参考以下文章