教你8步轻松上手kafka

1、配置环境变量
必须要求Java 8及以上版本

openjdk version "1.8.0\_262"  
OpenJDK Runtime Environment (build 1.8.0\_262-b10)  
OpenJDK 64-Bit Server VM (build 25.262-b10, mixed mode)  

2、下载并解压kafka安装文件
kafka下载地址
https://kafka.apache.org/downloads

上传至服务器并解压

kafka\_2.13-3.3.1.tgz  
\[root@localhost kafka\]# tar -xzf ./kafka\_2.13-3.3.1.tgz  
\[root@localhost kafka\]# ls  
kafka\_2.13-3.3.1 kafka\_2.13-3.3.1.tgz  
\[root@localhost kafka\]# cd ./kafka\_2.13-3.3.1/  
\[root@localhost kafka\_2.13-3.3.1\]# ls  
bin config libs LICENSE licenses NOTICE site-docs  
\[root@localhost kafka\_2.13-3.3.1\]#  

3、启动服务
必须按照正确的顺序先启动zookeeper，在启动kafka，启动命令如下。

启动zookeeper

启动kafka（另起一个窗口启动），注意在配置文件里面更改zookeeper.connect的参数值，本机默认为localhost，所以可以不用改。

4、创建topic
使用kafka-topics.sh创建名为local-topic的topic，localhost为kafka部署的主机名，9092为kafka的端口，需根据实际情况更改。

Created topic local-topic.  
\[root@localhost kafka\_2.13-3.3.1\]#  

5、查看topic

Topic: local-topic TopicId: zlnMG\_F9RO6VAeS2hj6rqw PartitionCount: 1 ReplicationFactor: 1 Configs:  
Topic: local-topic Partition: 0 Leader: 0 Replicas: 0 Isr: 0  
\[root@localhost kafka\_2.13-3.3.1\]#  

6、使用kafka-console-producer.sh为local-topic写入event，进入交互式窗口写入event，完成后使用Ctrl-C结束窗口。

\>local-topic 1 test  
\>local-topic 2 test  
\>local-topic 3 test  
\>^C\[root@localhost kafka\_2.13-3.3.1\]#  

7、使用kafka-console-consumer.sh查看local-topic写入的event，使用Ctrl-C结束窗口

local-topic 1 test  
local-topic 2 test  
local-topic 3 test  
^CProcessed a total of 3 messages  
\[root@localhost kafka\_2.13-3.3.1\]#  

8、以文件为例使用kafkaconnect导入数据
编辑connect-standalone.properties，指定plugin.path路径至对应的connect使用的jar包。

\[root@localhost kafka\_2.13-3.3.1\]# cat ./config/connect-standalone.properties | grep plugin.path  
\# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,  
#plugin.path=  
plugin.path=/kafka/kafka\_2.13-3.3.1/libs/connect-file-3.3.1.jar  
\[root@localhost kafka\_2.13-3.3.1\]# 

编辑测试文件test，写入测试数据
[root@localhost kafka_2.13-3.3.1]# echo -e “foo\nbar” > /kafka/test.txt
修改connect-file-source.properties、connect-file-sink.properties中file中的文件路径为要导入数据的文件路径，并根据实际情况修改其他参数值

\[root@localhost kafka\_2.13-3.3.1\]# cat ./config/connect-file-source.properties  
\# Licensed to the Apache Software Foundation (ASF) under one or more  
\# contributor license agreements. See the NOTICE file distributed with  
\# this work for additional information regarding copyright ownership.  
\# The ASF licenses this file to You under the Apache License, Version 2.0  
\# (the "License"); you may not use this file except in compliance with  
\# the License. You may obtain a copy of the License at  
#  
\# http://www.apache.org/licenses/LICENSE-2.0  
#  
\# Unless required by applicable law or agreed to in writing, software  
\# distributed under the License is distributed on an "AS IS" BASIS,  
\# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
\# See the License for the specific language governing permissions and  
\# limitations under the License.  
  
name=local-file-source  
connector.class=FileStreamSource  
tasks.max=1  
file=/kafka/test.txt  
topic=connect-test  
\[root@localhost kafka\_2.13-3.3.1\]#  
  

启动connect进程

查看文件内容读取结果，将此窗口保留，测试实时写入test文件数据后的变化。可以看到数据已经读入kafka并按文件配置创建connect-test的topic，查看test.sink.txt文件内容，跟test一致。

\_\_consumer\_offsets  
connect-test  
local-topic  
\[root@localhost kafka\_2.13-3.3.1\]#  
\[root@localhost kafka\_2.13-3.3.1\]# cat /kafka/test.sink.txt  
foo  
bar  
\[root@localhost kafka\_2.13-3.3.1\]# ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning  
{"schema":{"type":"string","optional":false},"payload":"foo"}  
{"schema":{"type":"string","optional":false},"payload":"bar"}  

在test.txt文件中追加信息，如追加test3到/kafka/test.txt，写完后观察前面保留的窗口变化。

\[root@localhost kafka\_2.13-3.3.1\]#

可以看到，新追加的test3会自动打印出来，并且在写入test.sink.txt

{"schema":{"type":"string","optional":false},"payload":"foo"}  
{"schema":{"type":"string","optional":false},"payload":"bar"}  
{"schema":{"type":"string","optional":false},"payload":"test3"}  
  
\[root@localhost kafka\_2.13-3.3.1\]# cat /kafka/test.sink.txt  
foo  
bar  
test3  

附：kafka-topics.sh帮助文档

Option Description  
\------ -----------  
\--alter Alter the number of partitions,  
replica assignment, and/or  
configuration for the topic.  
\--at-min-isr-partitions if set when describing topics, only  
show partitions whose isr count is  
equal to the configured minimum.  
\--bootstrap-server <String: server to REQUIRED: The Kafka server to connect  
connect to> to.  
\--command-config <String: command Property file containing configs to be  
config property file> passed to Admin Client. This is used  
only with --bootstrap-server option  
for describing and altering broker  
configs.  
\--config <String: name=value> A topic configuration override for the  
topic being created or altered. The  
following is a list of valid  
configurations:  
cleanup.policy  
compression.type  
delete.retention.ms  
file.delete.delay.ms  
flush.messages  
flush.ms  
follower.replication.throttled.  
replicas  
index.interval.bytes  
leader.replication.throttled.replicas  
local.retention.bytes  
local.retention.ms  
max.compaction.lag.ms  
max.message.bytes  
message.downconversion.enable  
message.format.version  
message.timestamp.difference.max.ms  
message.timestamp.type  
min.cleanable.dirty.ratio  
min.compaction.lag.ms  
min.insync.replicas  
preallocate  
remote.storage.enable  
retention.bytes  
retention.ms  
segment.bytes  
segment.index.bytes  
segment.jitter.ms  
segment.ms  
unclean.leader.election.enable  
See the Kafka documentation for full  
details on the topic configs. It is  
supported only in combination with --  
create if --bootstrap-server option  
is used (the kafka-configs CLI  
supports altering topic configs with  
a --bootstrap-server option).  
\--create Create a new topic.  
\--delete Delete a topic  
\--delete-config <String: name> A topic configuration override to be  
removed for an existing topic (see  
the list of configurations under the  
\--config option). Not supported with  
the --bootstrap-server option.  
\--describe List details for the given topics.  
\--disable-rack-aware Disable rack aware replica assignment  
\--exclude-internal exclude internal topics when running  
list or describe command. The  
internal topics will be listed by  
default  
\--help Print usage information.  
\--if-exists if set when altering or deleting or  
describing topics, the action will  
only execute if the topic exists.  
\--if-not-exists if set when creating topics, the  
action will only execute if the  
topic does not already exist.  
\--list List all available topics.  
\--partitions <Integer: # of partitions> The number of partitions for the topic  
being created or altered (WARNING:  
If partitions are increased for a  
topic that has a key, the partition  
logic or ordering of the messages  
will be affected). If not supplied  
for create, defaults to the cluster  
default.  
\--replica-assignment <String: A list of manual partition-to-broker  
broker\_id\_for\_part1\_replica1 : assignments for the topic being  
broker\_id\_for\_part1\_replica2 , created or altered.  
broker\_id\_for\_part2\_replica1 :  
broker\_id\_for\_part2\_replica2 , ...>  
\--replication-factor <Integer: The replication factor for each  
replication factor> partition in the topic being  
created. If not supplied, defaults  
to the cluster default.  
\--topic <String: topic> The topic to create, alter, describe  
or delete. It also accepts a regular  
expression, except for --create  
option. Put topic name in double  
quotes and use the '\\' prefix to  
escape regular expression symbols; e.  
g. "test\\.topic".  
\--topic-id <String: topic-id> The topic-id to describe.This is used  
only with --bootstrap-server option  
for describing topics.  
\--topics-with-overrides if set when describing topics, only  
show topics that have overridden  
configs  
\--unavailable-partitions if set when describing topics, only  
show partitions whose leader is not  
available  
\--under-min-isr-partitions if set when describing topics, only  
show partitions whose isr count is  
less than the configured minimum.  
\--under-replicated-partitions if set when describing topics, only  
show under replicated partitions  
\--version Display Kafka version.  
\[root@localhost kafka\_2.13-3.3.1\]#