文章目录

• 前言
• 一、Kafka相关Shell命令
• • （一）创建并查询Topic
  • （二）删除Topic
  • （三）增加Topic的分区
  • （四）生产数据到Topic
  • （五）从Topic消费数据
• 二、Java API
• • （一）需求
  • （二）准备工作
  • （三）代码开发 -- 生产者
  • （四）代码开发 -- 消费者
• 总结

成都创新互联服务项目包括江达网站建设、江达网站制作、江达网页制作以及江达网络营销策划等。多年来，我们专注于互联网行业，利用自身积累的技术优势、行业经验、深度合作伙伴关系等，向广大中小型企业、政府机构等提供互联网行业的解决方案，江达网站推广取得了明显的社会效益与经济效益。目前，我们服务的客户以成都为中心已经辐射到江达省份的部分城市，未来相信会继续扩大服务区域并继续获得客户的支持与信任！

---

前言

#博学谷IT学习技术支持#
本篇文章主要介绍Kafka相关的Shell命令和Java API，通过Shell命令和Java API可以进一步了解Kafka的操作，从而加深对Kafka的认识。

一、Kafka相关Shell命令

使用Kafka相关命令需要在linux服务器上进到Kafka解压的bin文件夹中，以下相关Shell命令的前提是linux服务器以及进入到Kafka的bin文件夹中，同时还需要确保Zookeeper集群的正常运行，因为Kafka集群依赖Zookeeper，相关环境安装可以参考：(1) Zookeeper集群搭建(2) Kafka集群搭建

（一）创建并查询Topic

1. 创建Topic

./kafka-topics.sh 
--bootstrap-server node1:9092,node2:9092,node3:9092 
--create 
--topic kafkaDemo2 
--partitions 3 
--replication-factor 2

解释：
./kafka-topics.sh  kafka集群中操作Topic的Shell脚本，该脚本位于Kafka安装环境的bin文件夹中
--bootstrap-server  指定该集群依赖的Zookeeper地址
--create  用于创建Topic管道
--topic  指定本次创建的Topic的名称
--partitions  设置该Topic的分区
--replication-factor  设置副本数
注意：如果没有设置分区和副本数，分区和副本数默认都为1

2. 查询Topic

./kafka-topics.sh 
--bootstrap-server node1:9092,node2:9092,node3:9092 
--list

解释：
--list  列出当前存在的Topic列表

3. 查询指定Topic

./kafka-topics.sh 
--bootstrap-server node1:9092,node2:9092,node3:9092 
--describe 
--topic kafkaDemo

解释：
--describe  查询Topic的信息
--topic 指定具体的Topic

（二）删除Topic

./kafka-topics.sh 
 --bootstrap-server node1:9092,node2:9092,node3:9092 
 --delete 
 --topic kafkaDemo2

解释：
--delete  删除Topic
--topic  指定Topic的名称

（三）增加Topic的分区

./kafka-topics.sh 
--bootstrap-server node1:9092,node2:9092,node3:9092 
--alter 
--topic kafkaDemo2 
--partitions 4

解释：
--alter  修改Topic相关信息
--topic  指定具体的Topic
--partitions  修改分区的数量

（四）生产数据到Topic

生产数据使用的bin文件夹下的kafka-console-producer.sh脚本，同时还需要使用相关命令才能生产数据到指定的Topic中，接着就是等待消费者到Topic进行数据的消费。

./kafka-console-producer.sh 
--broker-list node1:9092,node2:9092,node3:9092 
--topic kafkaDemo

解释：
--broker-list  指定Zookeeper集群地址
--topic  指定具体的Topic

（五）从Topic消费数据

消费指定Topic中的数据使用的是kafka-console-consumer.sh脚本，该脚本一运行，如果Topic中已经产生数据，则会自动消费该Topic中的数据。

./kafka-console-consumer.sh 
 --bootstrap-server node1:9092,node2:9092,node3:9092 
 --topic kafkaDemo

解释：
--bootstrap-server  指定Zookeeper集群地址
--topic  指定具体的Topic

二、Java API

通过Java的API可以实现kafka中数据的生产和消费，本次使用一个需求来介绍Java API

（一）需求

通过编写Java程序，将1-100的数字消息写入到Kafka中，并从Kafka中将数据读取出来

（二）准备工作

1. 第一步就是导API相关的jar包，该项目中使用Maven进行jar包管理，所以需要在pom文件夹中导入相关jar包
   （1）设置代码库url地址
   
   （2）导入项目所需jar包
   

（三）代码开发 – 生产者

以下为生产者的代码，将1-100写入到Topic【test】中，需要注意的是该类中相关jar包的导入

（四）代码开发 – 消费者

消费者不断从Kafka的Topic【test】中消费数据

---

总结

通过Kafka的shell命令以及相关Java API可以进一步加深对Kafka的认识，Kafka不仅具有很高的可靠性，还具备很优越的性能，对于发布和消费消息都具有高吞吐量，即使面对TB级的消息数据，其性能也是相当稳定。

你是否还在寻找稳定的海外服务器提供商？创新互联www.cdcxhl.cn海外机房具备T级流量清洗系统配攻击溯源，准确流量调度确保服务器高可用性，企业级服务器适合批量采购，新人活动首月15元起，快前往官网查看详情吧

网站标题：大数据之KafkaShell命令和JavaAPI-创新互联
文章起源：http://kswjz.com/article/cesojs.html