Kafka 安装部署-单节点

这篇具有很好参考价值的文章主要介绍了Kafka 安装部署-单节点。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

       Kafka强依赖ZK,如果想要使用Kafka,就必须安装ZK,Kafka中的消费偏置信息、kafka集群、topic信息会被存储在ZK中。有人可能会说我在使用Kafka的时候就没有安装ZK,那是因为Kafka内置了一个ZK,一般我们不使用它。

部署说明

本文使用kafka单节点安装及配置,并使用kafka自带的zookeeper。一般kafka需要起三个kafka构成集群,可以连单独的zookeeper,本文不涉及。

kafka下载

根据需要下载对应版本的安装包,下载地址:
https://archive.apache.org/dist/kafka/
上传安装包并解压重命名(路径自定义):
如:上传到 /opt 路径下
解压和重命名:

cd opt
tar -zxvf kafka_2.12-2.5.0.tgz
mv kafka_2.12-2.5.0 kafka

修改zookeeper配置

vim kafka/config/zookeeper.properties
dataDir=/opt/kafka
# the port at which the clients will connect
clientPort=2181
# disable the per-ip limit on the number of connections since this is a non-production config
maxClientCnxns=100
# Disable the adminserver by default to avoid port conflicts.
# Set the port to something non-conflicting if choosing to enable this
admin.enableServer=false
# admin.serverPort=8080

修改kafka配置

vim kafka/config/server.properties
broker.id=0
listeners=PLAINTEXT://192.168.1.1:9092
advertised.listeners=PLAINTEXT://192.168.1.1:9092
log.dirs=/opt/kafka/kafka-logs
zookeeper.connect=192.168.1.1:2181
# 本机ip
host.name=192.168.1.1

 启动zookeeper

./zookeeper-server-start.sh ../config/zookeeper.properties &

 启动kafka

./kafka-server-start.sh ../config/server.properties &

第二种

Kafka 单节点部署

下载安装
下载Zookeeper并解压到指定目录

ZooKeeper官网:http://zookeeper.apache.org/

[root@spark1 soft]# tar -zxvf zookeeper-3.4.5.tar.gz -C /application/

配置ZK

配置zoo.cfg

拷贝zoo_sample.cfg文件重命名为zoo.cfg,然后修改dataDir属性

[root@spark1 soft]# cd /application/zookeeper-3.4.5/conf/
[root@spark1 conf]# cat zoo.cfg
....
# 数据的存放目录
dataDir=/application/zookeeper-3.4.5/data
 
# 端口,默认就是2181
clientPort=2181
....
[root@spark1 conf]# 

配置环境变量

[root@spark1 ~]# cat /etc/profile
......
#ZooKeeper环境变量
export ZOOKEEPER_HOME=/application/zookeeper-3.4.5
export PATH=$PATH:$ZOOKEEPER_HOME/bin
......

Zookeeper 启动停止命令

$ zkServer.sh start
$ zkServer.sh stop

Kafka 单节点单Broker部署及使用

部署架构

kafka单节点部署,kafka,java,分布式

安装Kafka

Apache Kafka

tar -zxvf kafka_2.10-0.10.2.1.tgz -C /application/
 
#然后分别在各机器上创建软连
cd /application/
ln -s kafka_2.10-0.10.2.1 kafka

说明:

    1)版本最好注意下;在Java调用kafka的时候可能有问题就是版本造成的;

配置Kafka

配置server.properties

进入kafka的config目录下,有一个server.properties,添加如下配置

# broker的全局唯一编号,不能重复
broker.id=0
#这个最好配置,否则Java调用报错
advertised.listeners=PLAINTEXT://192.168.2.1:9092
# 日志目录
log.dirs=/application/kafka/logs/kafka-logs
# 配置zookeeper的连接(如果不是本机,需要该为ip或主机名)
zookeeper.connect=localhost:2181

启动Kafka

启动Zookeeper

zkServer.sh start

启动Kafka

[root@spark1 application]# /application/kafka/bin/kafka-server-start.sh  /application/kafka/config/server.properties

后台启动

nohup /application/kafka/bin/kafka-server-start.sh  /application/kafka/config/server.properties &  

创建topic

[root@spark1 ~]# /application/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
Created topic "test".
[root@spark1 ~]# 

说明: 
    1)--zookeeper:指定kafka连接zk的连接url,该值和server.properties文件中的配置项{zookeeper.connect}一样 
    2)--replication-factor:指定副本数量 
    3)--partitions:指定分区数量 
    4)--topic:主题名称

查看所有的topic信息

[root@spark1 ~]# /application/kafka/bin/kafka-topics.sh --list --zookeeper localhost:2181
TestTopic
WordCount
test
[root@spark1 ~]# 

启动生产者

[root@spark1 ~]# /application/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

启动消费者

[root@spark1 ~]# /application/kafka/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

0.90版本之后启动消费者的方法:

[root@spark1 ~]# /application/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic productscanlog --from-beginning

测试

产者生产数据

[root@spark1 ~]# /application/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
aa
bb
hello world
hello kafka  

消费者消费数据

[root@spark1 ~]# /application/kafka/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
aa
bb
hello world
hello kafka

说明:

    1)--from-beginning参数如果有表示从最开始消费数据,旧的和新的数据都会被消费,而没有该参数表示只会消费新产生的数据;

第三种

1.下载

从官网下载kafka,由于测试需要,我下载的是kafka_2.11-0.9.0.0.tgz

2.上传解压

将安装包上传到/opt/software目录,并解压

tar -zxf kafka_2.11-0.9.0.0.tgz -C /opt/module/

3.修改配置文件

创建logs目录

[root@hadoop01 module]# cd kafka_2.11-0.9.0.0/
[root@hadoop01 kafka_2.11-0.9.0.1]# ll
total 28
drwxr-xr-x 3 root root  4096 Feb 12  2016 bin
drwxr-xr-x 2 root root  4096 Feb 12  2016 config
drwxr-xr-x 2 root root  4096 May  8 09:39 libs
-rw-r--r-- 1 root root 11358 Feb 12  2016 LICENSE
-rw-r--r-- 1 root root   162 Feb 12  2016 NOTICE
drwxr-xr-x 2 root root    46 Feb 12  2016 site-docs
[root@hadoop01 kafka_2.11-0.9.0.1]# mkdir logs

修改server.properties配置文件

[root@hadoop01 kafka_2.11-0.9.0.1]# cd config/
You have new mail in /var/spool/mail/root
[root@hadoop01 config]# vim server.properties 

配置以下内容

#broker的全局唯一编号,不能重复
broker.id=0
#删除topic功能使能
delete.topic.enable=true
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘IO的现成数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#kafka运行日志存放的路径    
log.dirs=/opt/module/kafka_2.11-0.9.0.0/logs
#topic在当前broker上的分区个数
num.partitions=1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment文件保留的最长时间,超时将被删除
log.retention.hours=168
#配置连接Zookeeper集群地址
zookeeper.connect=hadoop01:2181

4.配置环境变量

[root@hadoop01]# sudo vim /etc/profile

#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka_2.11-0.9.0.0
export PATH=$PATH:$KAFKA_HOME/bin

[root@hadoop01]# source /etc/profile

5.修改zookeeper配置

[root@hadoop01 config]# vim zookeeper.properties

dataDir=/opt/module/kafka_2.11-0.9.0.0/zookeeper
# the port at which the clients will connect
clientPort=2181
# disable the per-ip limit on the number of connections since this is a non-production config
#maxClientCnxns=0
tickTime=2000
initLimit=10
syncLimit=5
server.1=hadoop01:2888:3888

在zookeeper的数据目录添加myid文件,写入服务broker.id属性值,如这里的目录是/opt/module/kafka_2.11-0.9.0.1/zookeeper

[root@hadoop01 kafka_2.11-0.9.0.0]# cd /opt/module/kafka_2.11-0.9.0.1/zookeeper
[root@hadoop01 zookeeper]# 
[root@hadoop01 zookeeper]# touch myid
[root@hadoop01 zookeeper]# echo 1 > myid
[root@hadoop01 zookeeper]# cat myid 
1

6.启动kafka

启动kafka之前要先启动zookeeper,这里使用的是kafka自带的zk

[root@hadoop01 ~]# cd /opt/module/kafka_2.11-0.9.0.0/
[root@hadoop01 kafka_2.11-0.9.0.0]# ls
bin  config  libs  LICENSE  logs  NOTICE  site-docs  zookeeper
[root@hadoop01 kafka_2.11-0.9.0.0]# 
[root@hadoop01 kafka_2.11-0.9.0.0]# bin/zookeeper-server-start.sh ./config/zookeeper.properties >./logs/zk.log &
[root@hadoop01 kafka_2.11-0.9.0.0]# jps
2326 Jps
[root@hadoop01 kafka_2.11-0.9.0.0]# bin/zookeeper-server-start.sh ./config/zookeeper.properties >./logs/zk.log &
[1] 2341
[root@hadoop01 kafka_2.11-0.9.0.0]# jps
2341 QuorumPeerMain
2367 Jps
[root@hadoop01 kafka_2.11-0.9.0.0]# 

启动kafka

[root@hadoop01 tools]# cd /opt/module/kafka_2.11-0.9.0.0/
[root@hadoop01 kafka_2.11-0.9.0.0]# bin/kafka-server-start.sh -daemon config/server.properties
[root@hadoop01 kafka_2.11-0.9.0.0]# 
[root@hadoop01 kafka_2.11-0.9.0.0]# jps
33593 QuorumPeerMain
33707 Kafka
33739 Jps
[root@hadoop01 kafka_2.11-0.9.0.1]#

停止kafka

[root@hadoop01 kafka_2.11-0.9.0.1]# bin/kafka-server-stop.sh stop

7.测试

创建topic

[root@hadoop01 kafka_2.11-0.9.0.0]# bin/kafka-topics.sh --create --zookeeper hadoop01:2181 --replication-factor 1 --partitions 1 --topic kafka_streaming_topic

参数说明:
–zookeeper:指定kafka连接zk的连接url,该值和server.properties文件中的配置项{zookeeper.connect}一样
–replication-factor:指定副本数量
–partitions:指定分区数量
–topic:主题名称

查看kafka中所有topic信息

[root@hadoop01 kafka_2.11-0.9.0.0]# bin/kafka-topics.sh --list --zookeeper hadoop01:2181

创建生产者

bin/kafka-console-producer.sh --broker-list hadoop01:9092 --topic kafka_streaming_topic

创建消费者

bin/kafka-console-consumer.sh --zookeeper hadoop01:2181 --topic kafka_streaming_topic --from-beginning

参数说明:
–from-beginning 从头开始消费
–bootstrap-server 指的是目标集群kafka的服务器地址,新版本之后(新版本指的是kafka 0.8.0之后的版本)开始使用 --bootstrap-server代替 --zookeeperkafka单节点部署,kafka,java,分布式文章来源地址https://www.toymoban.com/news/detail-721163.html

到了这里,关于Kafka 安装部署-单节点的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 【软件开发/设计】分布式架构中的组件(如Kafka、MongoDB和Nginx)如何进行容器化部署

    容器化部署是将应用程序及其依赖打包成一个容器镜像,然后在任何支持容器的环境中运行这个镜像的过程。在分布式架构中,像Nginx、MongoDB、Kafka这样的组件通过容器化可以更易于部署、扩展和管理。以下是这些组件容器化部署的一般步骤和原理: 容器化部署的一般步骤

    2024年02月04日
    浏览(49)
  • 分布式消息服务kafka

    什么是消息中间件? 消息中间件是分布式系统中重要的组件,本质就是一个具有接收消息、存储消息、分发消息的队列,应用程序通过读写队列消息来通信。 例如:在淘宝购物时,订单系统处理完订单后,把订单消息发送到消息中间件中,由消息中间件将订单消息分发到下

    2024年02月01日
    浏览(48)
  • 【分布式应用】kafka集群、Filebeat+Kafka+ELK搭建

    主要原因是由于在高并发环境下,同步请求来不及处理,请求往往会发生阻塞。比如大量的请求并发访问数据库,导致行锁表锁,最后请求线程会堆积过多,从而触发 too many connection 错误,引发雪崩效应。 我们使用消息队列,通过异步处理请求,从而缓解系统的压力。消息队

    2024年02月16日
    浏览(53)
  • 【分布式技术】消息队列Kafka

    目录 一、Kafka概述 二、消息队列Kafka的好处 三、消息队列Kafka的两种模式 四、Kafka 1、Kafka 定义 2、Kafka 简介 3、Kafka 的特性 五、Kafka的系统架构 六、实操部署Kafka集群  步骤一:在每一个zookeeper节点上完成kafka部署 ​编辑 步骤二:传给其他节点 步骤三:启动3个节点 kafka管理

    2024年01月23日
    浏览(55)
  • 分布式 - 消息队列Kafka:Kafka 消费者的消费位移

    01. Kafka 分区位移 对于Kafka中的分区而言,它的每条消息都有唯一的offset,用来表示消息在分区中对应的位置。偏移量从0开始,每个新消息的偏移量比前一个消息的偏移量大1。 每条消息在分区中的位置信息由一个叫位移(Offset)的数据来表征。分区位移总是从 0 开始,假设一

    2024年02月12日
    浏览(50)
  • 分布式 - 消息队列Kafka:Kafka消费者的分区分配策略

    Kafka 消费者负载均衡策略? Kafka 消费者分区分配策略? 1. 环境准备 创建主题 test 有5个分区,准备 3 个消费者并进行消费,观察消费分配情况。然后再停止其中一个消费者,再次观察消费分配情况。 ① 创建主题 test,该主题有5个分区,2个副本: ② 创建3个消费者CustomConsu

    2024年02月13日
    浏览(47)
  • 分布式 - 消息队列Kafka:Kafka生产者架构和配置参数

    生产者发送消息流程参考图1: 先从创建一个ProducerRecord对象开始,其中需要包含目标主题和要发送的内容。另外,还可以指定键、分区、时间戳或标头。在发送ProducerRecord对象时,生产者需要先把键和值对象序列化成字节数组,这样才能在网络上传输。 接下来,如果没有显式

    2024年02月13日
    浏览(50)
  • 分布式 - 消息队列Kafka:Kafka生产者发送消息的方式

    不管是把Kafka作为消息队列、消息总线还是数据存储平台,总是需要一个可以往Kafka写入数据的生产者、一个可以从Kafka读取数据的消费者,或者一个兼具两种角色的应用程序。 Kafka 生产者是指使用 Apache Kafka 消息系统的应用程序,它们负责将消息发送到 Kafka 集群中的一个或多

    2024年02月13日
    浏览(44)
  • 分布式 - 消息队列Kafka:Kafka消费者和消费者组

    1. Kafka 消费者是什么? 消费者负责订阅Kafka中的主题,并且从订阅的主题上拉取消息。与其他一些消息中间件不同的是:在Kafka的消费理念中还有一层消费组的概念,每个消费者都有一个对应的消费组。当消息发布到主题后,只会被投递给订阅它的每个消费组中的一个消费者

    2024年02月13日
    浏览(45)
  • 分布式 - 消息队列Kafka:Kafka 消费者消费位移的提交方式

    最简单的提交方式是让消费者自动提交偏移量,自动提交 offset 的相关参数: enable.auto.commit:是否开启自动提交 offset 功能,默认为 true; auto.commit.interval.ms:自动提交 offset 的时间间隔,默认为5秒; 如果 enable.auto.commit 被设置为true,那么每过5秒,消费者就会自动提交 poll() 返

    2024年02月12日
    浏览(48)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包