提示:以下是本篇文章正文内容,下面案例可供参考
2.使用Xftp将压缩包上传至Linux root目录下
压缩包解压至 opt目录下,并重命名方便后续操作
代码如下(示例):
[root@localhost ~]# ls
anaconda-ks.cfg jdk-8u131-linux-x64.tar.gz kafka_2.12-2.2.0.tgz zookeeper-3.4.6.tar.gz
[root@localhost ~]# tar zxvf kafka_2.12-2.2.0.tgz -C /opt
kafka_2.12-2.2.0/
kafka_2.12-2.2.0/LICENSE
kafka_2.12-2.2.0/NOTICE
然后重命名
[root@localhost ~]# cd /opt
[root@localhost opt]# ls
kafka_2.12-2.2.0[root@localhost opt]# mv kafka_2.12-2.2.0/ kafka
[root@localhost opt]# ls
kafka
[root@localhost opt]#
然后修改config目录下的server.properties文件
[root@localhost kafka]# ls
bin config kafkaStart.sh libs LICENSE log logs NOTICE site-docs zookeeper
[root@localhost kafka]# cd config/[root@localhost config]# ls
config connect-console-source.properties connect-file-sink.properties connect-log4j.properties consumer.properties producer.properties tools-log4j.properties
connect-console-sink.properties connect-distributed.properties connect-file-source.properties connect-standalone.properties log4j.properties server.properties zookeeper.properties
[root@localhost config]# vi server.properties
主要修改如下配置:
log.dirs=/opt/kafka/log
不要忘记创建log文件夹
[root@localhost config]# cd ..[root@localhost kafka]# mkdir log
[root@localhost kafka]# ls
bin config libs LICENSE log NOTICE site-docs
[root@localhost kafka]# pwd
/opt/kafka
[root@localhost kafka]#
修改kafka中config目录下zookeeper.properties
vi zookeeper.properties
dataDir=/opt/kafka/zookeeper/data/dataDir
dataLogDir=/opt/kafka/zookeeper/data/dataLogDir
#theport at which the clients will connect
clientPort=2181#disablethe per-ip limit on the number of connections since this is a non-production config
maxClientCnxns=100
tickTime=2000
initLimit=10
接下来创建刚才提到的文件夹
命令如下:
[root@localhost config]# cd ..[root@localhost kafka]# mkdir zookeeper
[root@localhost kafka]# ls
bin config libs LICENSE log NOTICE site-docs zookeeper
[root@localhost kafka]# cd zookeeper/[root@localhost zookeeper]# mkdir data
[root@localhost zookeeper]# ls
data
[root@localhost zookeeper]# cd data/[root@localhost data]# mkdir dataDir
[root@localhost data]# ls
dataDir
[root@localhost data]# mkdir dataLogDir
[root@localhost data]# ls
dataDir dataLogDir
[root@localhost data]#
编写启动脚本:
在kafka目录下 vi kafkaStart.sh
然后添加如下内容
#!/bin/bash
#启动zookeeper
/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper.properties &
sleep 3 #默默等3秒后执行
#启动kafka
/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties &
添加脚本执行权限:
[root@localhost kafka]# chmod +x kafkaStart.sh
注意:因为脚本中已经包含zookeeper的开启,因此不需要单独开zookeeper
[root@localhost kafka]# ./kafkaStart.sh
成功开启kafka
接下来在bin目录创建topic:
创建Topic命令:
[root@localhost bin]# ./kafka-topics.sh --create --zookeeper localhost:2181--replication-factor 1--partitions 1--topic test
test是topic名称 可以随变换
出现Created topic test即创建成功
查询创建的top主题 出现test则为正常
查询Topic命令:
./kafka-topics.sh --list --zookeeper localhost:2181
删除topic:
删除topic命令:
./kafka-topics.sh --delete --zookeeper localhost:2181--topic test
创建Producer命令:
[root@localhost bin]# ./kafka-console-producer.sh --broker-list localhost:9092--topic test
创建消费者:
在XShell中开启另一个会话:
创建Consumer命令:
[root@localhost bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092--topic test --from-beginning
生产者发送消息:
消费者收到消息:
发送消息后,消费者会话框会变成叹号
关闭的话需要jps查出
然后再关闭kafka,之后是zookeeper,要先关kafka
版权归原作者 楚天阔. 所有, 如有侵权,请联系我们删除。