文档章节

Kafka 环境部署与配置详解

hming
 hming
发布于 2017/03/27 17:00
字数 2974
阅读 494
收藏 4

什么是Kafka

Kafka是一种高吞吐量 的分布式发布订阅消息系统,有如下特性:
    1>.通过O(1)的磁盘数据结构提供消息的持久化,这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。
    2>.高吞吐量:即使是非常普通的硬件Kafka也可以支持每秒数百万的消息。
    3>.支持通过Kafka服务器和消费机集群来分区消息。
    4>.支持Hadoop并行数据加载。

Kafka常用术语

Broker

Kafka集群包含一个或多个服务器,这种服务器被称为broker

Topic

每条发布到Kafka集群的消息都有一个类别,这个类别被称为Topic。(物理上不同Topic的消息分开存储,逻辑上一个Topic的消息
虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处)

Partition

Partition是物理上的概念,每个Topic包含一个或多个Partition.为了更好的实现负载均衡和消息的顺序性,kafka的producer在
分发消息时可以通过分发策略发送给指定的partition。实现分发的程序是需要制定消息的key值,而kafka通过key进行策略分发。

Producer

负责发布消息到Kafka broker

Consumer

消息消费者,向Kafka broker读取消息的客户端。

Consumer Group

每个Consumer属于一个特定的ConsumerGroup(可为每个Consumer指定group name,若不指定group name则属于默认的group)。

Kafka官网在哪?

http://kafka.apache.org/

Kafka安装与配置

JDK安装

下载JDK  (注意:直接复制到下载工具进行下载,版本请使用JDK7,JDK8有可能不兼容kafka_2.11-0.8.2.1)
cd /usr/local/src
 wget http://download.oracle.com/otn-pub/java/jdk/7u79-b15/jdk-7u79-linux-x64.rpm
 chmod +x jdk-7u79-linux-x64.rpm # 添加执行权限
 rpm -ivh jdk-7u79-linux-x64.rpm #安装
 cd /usr/java/ #安装完成之后,可以cd /usr/java/ 到安装目录查看到安装目录查看
 添加JDK到系统环境变量
 vim /etc/profile  #编辑,在最后添加以下代码
 export JAVA_HOME=/usr/java/jdk1.7.0_79
 export PATH=$PATH:$JAVA_HOME/bin:/usr/bin:/usr/sbin:/bin:/sbin:/usr/X11R6/bin
 export CLASSPATH=.:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar
 wq! #保存退出
 source /etc/profile  #使配置文件立即生效
 java -version #出现下述信息 则安装成功 可进行下一步
   java version "1.7.0_79"
   Java(TM) SE Runtime Environment (build 1.7.0_79-b15)
   Java HotSpot(TM) 64-Bit Server VM (build 24.79-b02, mixed mode)
   You have new mail in /var/spool/mail/root

kafka服务配置

cd /usr/local/src
wget http://archive.apache.org/dist/kafka/0.8.2.1/kafka_2.11-0.8.2.1.tgz 
#(kafka_2.11-0.8.2.1.tgz版本是已经编译好的版本,解压就能使用)
tar -xzvf kafka_2.11-0.8.2.1.tgz #解压
mv kafka_2.11-0.8.2.1 /usr/local/kafka #移动到安装目录

mkdir /usr/local/kafka/log/kafka #创建kafka日志目录
mkdir /usr/local/kafka/zookeeper #创建zookeeper目录
mkdir /usr/local/kafka/log/zookeeper #创建zookeeper日志目录
cd /usr/local/kafka/config #进入配置目录
vim server.properties #编辑修改相应的参数
 ---------------------------------------------server.properties-------------------------------------
 broker.id=0
 port=9092                                              #端口号
 host.name=192.168.0.11                                 #服务器IP地址,修改为自己的服务器IP
 log.dirs=/usr/local/kafka/log/kafka                    #日志存放路径,上面创建的目录
 zookeeper.connect=localhost:2181                       #zookeeper地址和端口,单机配置部署,localhost:2181
 log.cleaner.enable=false                               #topic是否可删除
 zookeeper.connection.timeout.ms=6000                   #超时设置
 -----------------------------------------------------------------------------------------------------
 vim zookeeper.properties #编辑修改相应的参数
 ---------------------------------------------zookeeper.properties------------------------------------
 dataDir=/usr/local/kafka/zookeeper #zookeeper数据目录
 dataLogDir=/usr/local/kafka/log/zookeeper #zookeeper日志目录
 clientPort=2181
 maxClientCnxns=100
 tickTime=2000
 initLimit=10
 syncLimit=5
 ------------------------------------------------------------------------------------------------------
 cd /usr/local/kafka
 vim kafkastart.sh #创建启动脚本
 ---------------------------------------------kafkastart.sh--------------------------------------------
 #!/bin/sh
 #启动zookeeper
 /usr/local/kafka/bin/zookeeper-server-start.sh /usr/local/kafka/config/zookeeper.properties &
 sleep 3 #等3秒后执行
 #启动kafka
 /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &
 -------------------------------------------------------------------------------------------------------
 vim kafkastop.sh #创建停止脚本
 ----------------------------------------------kafkastop.sh---------------------------------------------
 #!/bin/sh
 #启动zookeeper
 /usr/local/kafka/bin/zookeeper-server-stop.sh /usr/local/kafka/config/zookeeper.properties &
 sleep 3 #等3秒后执行
 #启动kafka
 /usr/local/kafka/bin/kafka-server-stop.sh /usr/local/kafka/config/server.properties &
 -------------------------------------------------------------------------------------------------------
 chmod +x kafkastart.sh
 chmod +x kafkastop.sh

server.properties 配置说明

name	默认值	描述
broker.id  none  每一个boker都有一个唯一的id作为它们的名字.这就允许boker切换到别的主机/端口上,consumer依然知道
enable.zookeeper	true	允许注册到zookeeper
log.flush.interval.messages	Long.MaxValue	在数据被写入到硬盘和消费者可用前最大累积的消息的数量
log.flush.interval.ms	Long.MaxValue	在数据被写入到硬盘前的最大时间
log.flush.scheduler.interval.ms	Long.MaxValue	检查数据是否要写入到硬盘的时间间隔。
log.retention.hours	168	控制一个log保留多长个小时
log.retention.bytes	-1	控制log文件最大尺寸
log.cleaner.enable	false	是否log cleaning
log.cleanup.policy  delete  delete还是compat.其它控制参数还包括
log.cleaner.threads,log.cleaner.io.max.bytes.per.second,
log.cleaner.dedupe.buffer.size,log.cleaner.io.buffer.size,log.cleaner.io.buffer.load.factor,
log.cleaner.backoff.ms,log.cleaner.min.cleanable.ratio,log.cleaner.delete.retention.ms
log.dir	/tmp/kafka-logs	指定log文件的根目录
log.segment.bytes	110241024*1024	单一的log segment文件大小
log.roll.hours	24 * 7	开始一个新的log文件片段的最大时间
message.max.bytes	1000000 + MessageSet.LogOverhead	一个socket 请求的最大字节数
num.network.threads	3	处理网络请求的线程数
num.io.threads	8	处理IO的线程数
background.threads	10	后台线程序
num.partitions	1	默认分区数
socket.send.buffer.bytes	102400	socket SO_SNDBUFF参数
socket.receive.buffer.bytes	102400	socket SO_RCVBUFF参数
zookeeper.connect	localhost:2182/kafka	指定zookeeper连接字符串, 格式如hostname:port/chroot。
chroot是一个namespace 
zookeeper.connection.timeout.ms	6000	指定客户端连接zookeeper的最大超时时间
zookeeper.session.timeout.ms	6000	连接zk的session超时时间
zookeeper.sync.time.ms	2000	zk follower落后于zk leader的最长时间

producer.properties 配置说明

#指定kafka节点列表,用于获取metadata,不必全部指定
    metadata.broker.list=192.168.2.105:9092,192.168.2.106:9092

    # 指定分区处理类。默认kafka.producer.DefaultPartitioner,表通过key哈希到对应分区
    #partitioner.class=com.meituan.mafka.client.producer.CustomizePartitioner

    # 是否压缩,默认0表示不压缩,1表示用gzip压缩,2表示用snappy压缩。压缩后消息中会有头来指明消息压缩类型,
    # 故在消费者端消息解压是透明的无需指定。
    compression.codec=none
      
    # 指定序列化处理类(mafka client API调用说明-->3.序列化约定wiki),默认为kafka.serializer.DefaultEncoder,
    # 即byte[]
    serializer.class=com.meituan.mafka.client.codec.MafkaMessageEncoder
    # serializer.class=kafka.serializer.DefaultEncoder
    # serializer.class=kafka.serializer.StringEncoder

    # 如果要压缩消息,这里指定哪些topic要压缩消息,默认empty,表示不压缩。
    #compressed.topics=

    ########### request ack ###############
    #消息的确认模式 默认为0
    # 0:不保证消息的到达确认,只管发送,低延迟但是会出现消息的丢失,在某个server失败的情况下,有点像TCP
    # 1:发送消息,并会等待leader 收到确认后,一定的可靠性
    # -1:发送消息,等待leader收到确认,并进行复制操作后,才返回,最高的可靠性
    request.required.acks=0

    # 在向producer发送ack之前,broker允许等待的最大时间
    # 如果超时,broker将会向producer发送一个error ACK.意味着上一次消息因为某种
    # 原因未能成功(比如follower未能同步成功)
    request.timeout.ms=10000
    ########## end #####################

    # 同步还是异步发送消息,默认“sync”表同步,"async"表异步。异步可以提高发送吞吐量,
    # 也意味着消息将会在本地buffer中,并适时批量发送,但是也可能导致丢失未发送过去的消息
    producer.type=sync

    ############## 异步发送 (以下四个异步参数可选) ####################
    # 在async模式下,当message被缓存的时间超过此值后,将会批量发送给broker,默认为5000ms
    # 此值和batch.num.messages协同工作.
    queue.buffering.max.ms = 5000

    # 在async模式下,producer端允许buffer的最大消息量
    # 无论如何,producer都无法尽快的将消息发送给broker,从而导致消息在producer端大量沉积
    # 此时,如果消息的条数达到阀值,将会导致producer端阻塞或者消息被抛弃,默认为10000
    queue.buffering.max.messages=20000

    # 如果是异步,指定每次批量发送数据量,默认为200
    batch.num.messages=500

    # 当消息在producer端沉积的条数达到"queue.buffering.max.meesages"后
    # 阻塞一定时间后,队列仍然没有enqueue(producer仍然没有发送出任何消息)
    # 此时producer可以继续阻塞或者将消息抛弃,此timeout值用于控制"阻塞"的时间
    # -1: 无阻塞超时限制,消息不会被抛弃
    # 0:立即清空队列,消息被抛弃
    queue.enqueue.timeout.ms=-1
    ################ end ###############

    # 当producer接收到error ACK,或者没有接收到ACK时,允许消息重发的次数
    # 因为broker并没有完整的机制来避免消息重复,所以当网络异常时(比如ACK丢失)
    # 有可能导致broker接收到重复的消息,默认值为3.
    message.send.max.retries=3

    # producer刷新topic metada的时间间隔,producer需要知道partition leader的位置,以及当前topic的情况
    # 因此producer需要一个机制来获取最新的metadata,当producer遇到特定错误时,将会立即刷新
    # (比如topic失效,partition丢失,leader失效等),此外也可以通过此参数来配置额外的刷新机制,默认值600000
    topic.metadata.refresh.interval.ms=60000

consumer.properties 配置说明

# zookeeper连接服务器地址,此处为线下测试环境配置(kafka消息服务-->kafka broker集群线上部署环境wiki)
    # 配置例子:"127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002"
    zookeeper.connect=192.168.2.225:2181,192.168.2.225:2182,192.168.2.225:2183/config/mobile/mq/mafka

    # zookeeper的session过期时间,默认5000ms,用于检测消费者是否挂掉,当消费者挂掉,其他消费者要等该指定时间
    # 才能检查到并且触发重新负载均衡
    zookeeper.session.timeout.ms=5000
    zookeeper.connection.timeout.ms=10000

    # 指定多久消费者更新offset到zookeeper中。注意offset更新时基于time而不是每次获得的消息。一旦在更新zookeeper
    # 发生异常并重启,将可能拿到已拿到过的消息
    zookeeper.sync.time.ms=2000

    #指定消费组
    group.id=xxx

    # 当consumer消费一定量的消息之后,将会自动向zookeeper提交offset信息
    # 注意offset信息并不是每消费一次消息就向zk提交一次,而是现在本地保存(内存),并定期提交,默认为true
    auto.commit.enable=true

    # 自动更新时间。默认60 * 1000
    auto.commit.interval.ms=1000

    # 当前consumer的标识,可以设定,也可以有系统生成,主要用来跟踪消息消费情况,便于观察
    conusmer.id=xxx

    # 消费者客户端编号,用于区分不同客户端,默认客户端程序自动产生
    client.id=xxxx

    # 最大取多少块缓存到消费者(默认10)
    queued.max.message.chunks=50

    # 当有新的consumer加入到group时,将会reblance,此后将会有partitions的消费端迁移到新
    # 的consumer上,如果一个consumer获得了某个partition的消费权限,那么它将会向zk注册
    # "Partition Owner registry"节点信息,但是有可能此时旧的consumer尚没有释放此节点,
    # 此值用于控制,注册节点的重试次数.
    rebalance.max.retries=5

    # 获取消息的最大尺寸,broker不会像consumer输出大于此值的消息chunk
    # 每次feth将得到多条消息,此值为总大小,提升此值,将会消耗更多的consumer端内存
    fetch.min.bytes=6553600

    # 当消息的尺寸不足时,server阻塞的时间,如果超时,消息将立即发送给consumer
    fetch.wait.max.ms=5000
    socket.receive.buffer.bytes=655360

    # 如果zookeeper没有offset值或offset值超出范围。那么就给个初始的offset。有smallest、largest、
    # anything可选,分别表示给当前最小的offset、当前最大的offset、抛异常。默认largest
    auto.offset.reset=smallest

    # 指定序列化处理类(mafka client API调用说明-->3.序列化约定wiki),默认为kafka.serializer.DefaultDecoder,
    # 即byte[]
    derializer.class=com.meituan.mafka.client.codec.MafkaMessageDecoder
参数参考可见:http://debugo.com/kafka-params/

kafka 服务启动

sh /usr/local/kafka/kafkastart.sh        #kafka启动
sh /usr/local/kafka/kafkastop.sh         #kafka停止

kafka常用命令

创建一个名为test的topic
 bin/kafka-topics.sh --create --zookeeper 192.168.1.3:2181 --replication-factor 1 --partitions 5 --topic b
 查看当前创建的topic
 bin/kafka-topics.sh --zookeeper 192.168.1.3::2181 --list
 验证生产消息成功 
 bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.1.3::9092 --topic boyaa --time -1 
 #(参数解释:--time -1 表示从最新的时间的offset中得到数据条数) 
 #输出结果每个字段分别表示topic、partition、untilOffset

 开启消息生产者
 bin/kafka-console-producer.sh --broker-list 192.168.1.3:9092 --sync --topic b
 开启消费者
 bin/kafka-console-consumer.sh --zookeeper 192.168.201.73:2181 --topic boyaa --from-beginning 
 #(--from-beginning 如果consumer之前没有建立offset,则从producer最开始的数据读取。)

© 著作权归作者所有

hming
粉丝 3
博文 88
码字总数 86648
作品 0
深圳
程序员
私信 提问
apache kafka技术分享系列(目录索引)

目录索引: Kafka使用场景 1.为何使用消息系统 2.我们为何需要搭建ApacheKafka分布式系统 3.消息队列中点对点与发布订阅区别 kafka开发与管理: 1)apachekafka消息服务 2)kafak安装与使用 ...

dannyhe
2015/09/06
453
1
logstash通过kafka传输nginx日志

logstash通过kafka传输nginx日志(三)   单个进程 logstash 可以实现对数据的读取、解析和输出处理。但是在生产环境中,从每台应用服务器运行 logstash 进程并将数据直接发送到 Elastics...

yixinsiyu
2018/08/27
0
0
深入浅出、点金赋能!Hadoop体系大数据开发案例实战

感谢关注天善智能,走好数据之路↑↑↑ 欢迎关注天善智能,我们是专注于商业智能BI,人工智能AI,大数据分析与挖掘领域的垂直社区,学习,问答、求职一站式搞定! 课程名称:深入浅出、点金赋...

天善智能
2018/01/03
0
0
超人学院大数据高薪就业班第七期精彩课程抢先看

超人学院大数据高薪就业班第七期火爆招生中,新的一期,内容部不断增加,还有更多优惠呦,精彩内容抢先看: 1、Linux课程:如果你没有使用过Linux,别担心,本内容让你轻松入门,讲解Linux基...

超人学院
2015/04/30
0
0
消息中间件(Kafka/RabbitMQ)收录集

本篇主要整理工作中遇到的一些消息中间件的相关知识,包括Kafka, RabbitMQ, RocketMQ, ActiveMQ等,不排除收录其他消息中间件的可能。 这里会持续收录相关知识,包括安装、部署、使用示例、监...

u013256816
2017/01/26
0
0

没有更多内容

加载失败,请刷新页面

加载更多

阿里P7架构师:这些技术点没搞懂,我劝你不要跳槽!

阿里P7架构师架构师:这些技术点没搞懂,我劝你不要跳槽! 哪些技术点呢? 废话不多说,技术点全在下面这6张图里面了! 1.怎么看源码? 2.分布式 3.微服务 4.性能优化 5.工程化 粉丝福利:一...

别打我会飞
24分钟前
2
0
易错题

父类必须有一个无参构造,不然会报Implicit super constructor Parent() is undefined. Must explicitly invoke another constructor...

architect刘源源
29分钟前
1
0
使用Json4s 将带有Timestamp的对象转json时 变为所在时区

在有Timestamp属性的对象转json时 最后出来的时间会减小8小时,是因为变成了0时区,需要将隐式转换中添加设置本地时区 import org.json4s.{DefaultFormats, Formats}import org.json4s.jac...

可达鸭Go
31分钟前
1
0
工作六年java程序员的工作感悟,带你走出迷茫

前言 很多年前,刚刚从大学毕业的时候,很多公司来校招。其中最烂俗的一个面试问题是:“你希望你之后三到五年的发展是什么?”。我当时的标准回答是(原话):“成为在某一方面能够独当一面...

java知识分子
36分钟前
3
0
jenkins邮箱发送失败以及解决方案

jenkins邮箱设置报错以及解决方案 Jenkins邮箱设置 我用了 Email Extended Plugin这个插件,这个插件会有更加丰富的邮件内容,建议添加。 1.安装 Email Extended Plugin 2.系统管理-系统设置...

shzwork
45分钟前
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部