kafka中处理超大消息的一些考虑
时间： 00:38:26
&&&& 阅读：14627
&&&& 评论：
&&&& 收藏：0
标签：Kafka设计的初衷是迅速处理短小的消息，一般10K大小的消息吞吐性能最好（可参见LinkedIn的kafka性能测试）。但有时候，我们需要处理更大的消息，比如XML文档或JSON内容，一个消息差不多有10-100M，这种情况下，Kakfa应该如何处理？针对这个问题，有以下几个建议：& 最好的方法是不直接传送这些大的数据。如果有共享存储，如NAS, HDFS, S3等，可以把这些大的文件存放到共享存储，然后使用Kafka来传送文件的位置信息。& 第二个方法是，将大的消息数据切片或切块，在生产端将数据切片为10K大小，使用分区主键确保一个大消息的所有部分会被发送到同一个kafka分区（这样每一部分的拆分顺序得以保留），如此以来，当消费端使用时会将这些部分重新还原为原始的消息。& 第三，Kafka的生产端可以压缩消息，如果原始消息是XML，当通过压缩之后，消息可能会变得不那么大。在生产端的配置参数中使用compression.codec和commpressed.topics可以开启压缩功能，压缩算法可以使用GZip或Snappy。& & 不过如果上述方法都不是你需要的，而你最终还是希望传送大的消息，那么，则可以在kafka中设置下面一些参数：broker 配置:&&& message.max.bytes (默认:1000000) – broker能接收消息的最大字节数，这个值应该比消费端的fetch.message.max.bytes更小才对，否则broker就会因为消费端无法使用这个消息而挂起。&&& log.segment.bytes (默认: 1GB) – kafka数据文件的大小，确保这个数值大于一个消息的长度。一般说来使用默认值即可（一般一个消息很难大于1G，因为这是一个消息系统，而不是文件系统）。&&& replica.fetch.max.bytes (默认: 1MB) – broker可复制的消息的最大字节数。这个值应该比message.max.bytes大，否则broker会接收此消息，但无法将此消息复制出去，从而造成数据丢失。Consumer 配置:&fetch.message.max.bytes (默认 1MB) – 消费者能读取的最大消息。这个值应该大于或等于message.max.bytes。所以，如果你一定要选择kafka来传送大的消息，还有些事项需要考虑。要传送大的消息，不是当出现问题之后再来考虑如何解决，而是在一开始设计的时候，就要考虑到大消息对集群和主题的影响。性能: 根据前面提到的性能测试，kafka在消息为10K时吞吐量达到最大，更大的消息会降低吞吐量，在设计集群的容量时，尤其要考虑这点。可用的内存和分区数：Brokers会为每个分区分配replica.fetch.max.bytes参数指定的内存空间，假设replica.fetch.max.bytes=1M，且有1000个分区，则需要差不多1G的内存，确保 分区数*最大的消息不会超过服务器的内存，否则会报OOM错误。同样地，消费端的fetch.message.max.bytes指定了最大消息需要的内存空间，同样，分区数*最大需要内存空间 不能超过服务器的内存。所以，如果你有大的消息要传送，则在内存一定的情况下，只能使用较少的分区数或者使用更大内存的服务器。垃圾回收：到现在为止，我在kafka的使用中还没发现过此问题，但这应该是一个需要考虑的潜在问题。更大的消息会让GC的时间更长（因为broker需要分配更大的块），随时关注GC的日志和服务器的日志信息。如果长时间的GC导致kafka丢失了zookeeper的会话，则需要配置zookeeper.session.timeout.ms参数为更大的超时时间。一切的一切，都需要在权衡利弊之后，再决定选用哪个最合适的方案。标签：原文地址：http://www.cnblogs.com/doubletree/p/4264969.html
&&国之画&&&& &&&&chrome插件
版权所有 京ICP备号-2
迷上了代码！专注于数据结构、算法、c/c++、python、数据挖掘等
为什么说Kafka使用磁盘比内存快
学习过[]的同学一定会问为什么Kafka大量使用了磁盘作为传统意义的缓存。
其实Kafka最核心的思想是使用磁盘，而不是使用内存，可能所有人都会认为，内存的速度一定比磁盘快，我也不例外。在看了Kafka的设计思想，查阅了相应资料再加上自己的测试后，发现磁盘的顺序读写速度和内存持平。
而且Linux对于磁盘的读写优化也比较多，包括read-ahead和write-behind，磁盘缓存等。如果在内存做这些操作的时候，一个是JAVA对象的内存开销很大，另一个是随着堆内存数据的增多，JAVA的GC时间会变得很长，使用磁盘操作有以下几个好处：
磁盘缓存由Linux系统维护，减少了程序员的不少工作。磁盘顺序读写速度超过内存随机读写。JVM的GC效率低，内存占用大。使用磁盘可以避免这一问题。系统冷启动后，磁盘缓存依然可用。
没有更多推荐了，
加入CSDN，享受更精准的内容推荐，与500万程序员共同成长！Kafka主要参数详解
#############################
System #############################
#唯一标识在集群中的ID，要求是正数。
broker.id=0
#服务端口，默认9092
#监听地址，不设为所有地址
host.name=debugo01
# 处理网络请求的最大线程数
num.network.threads=2
# 处理磁盘I/O的线程数
num.io.threads=8
# 一些后台线程数
background.threads
# 等待IO线程处理的请求队列最大数
queued.max.requests
#&&socket的发送缓冲区（SO_SNDBUF）
socket.send.buffer.bytes=1048576
# socket的接收缓冲区 (SO_RCVBUF)
socket.receive.buffer.bytes=1048576
# socket请求的最大字节数。为了防止内存溢出，message.max.bytes必然要小于
socket.request.max.bytes
#############################
Topic #############################
# 每个topic的分区个数，更多的partition会产生更多的segment
num.partitions=2
# 是否允许自动创建topic ，若是false，就需要通过命令创建topic
auto.create.topics.enable
# 一个topic ，默认分区的replication个数
，不能大于集群中broker的个数。
default.replication.factor
# 消息体的最大大小，单位是字节
message.max.bytes
#############################
ZooKeeper #############################
# Zookeeper quorum设置。如果有多个使用逗号分割
zookeeper.connect=debugo01:2181,debugo02,debugo03
# 连接zk的超时时间
zookeeper.connection.timeout.ms=1000000
# ZooKeeper集群中leader和follower之间的同步实际
zookeeper.sync.time.ms
#############################
Log #############################
#日志存放目录，多个目录使用逗号分割
log.dirs=/var/log/kafka
# 当达到下面的消息数量时，会将数据flush到日志文件中。默认10000
#log.flush.interval.messages=10000
# 当达到下面的时间(ms)时，执行一次强制的flush操作。interval.ms和interval.messages无论哪个达到，都会flush。默认3000ms
#log.flush.interval.ms=1000
# 检查是否需要将日志flush的时间间隔
log.flush.scheduler.interval.ms
# 日志清理策略（delete|compact）
log.cleanup.policy
# 日志保存时间 (hours|minutes)，默认为7天（168小时）。超过这个时间会根据policy处理数据。bytes和minutes无论哪个先达到都会触发。
log.retention.hours=168
# 日志数据存储的最大字节数。超过这个时间会根据policy处理数据。
#log.retention.bytes=
# 控制日志segment文件的大小，超出该大小则追加到一个新的日志segment文件中（-1表示没有限制）
log.segment.bytes=
# 当达到下面时间，会强制新建一个segment
log.roll.hours
# 日志片段文件的检查周期，查看它们是否达到了删除策略的设置（log.retention.hours或log.retention.bytes）
log.retention.check.interval.ms=60000
# 是否开启压缩
log.cleaner.enable=false
# 对于压缩的日志保留的最长时间
log.cleaner.delete.retention.ms
# 对于segment日志的索引文件大小限制
log.index.size.max.bytes
#y索引计算的一个缓冲区，一般不需要设置。
log.index.interval.bytes
#############################
replica #############################
# partition management controller
与replicas之间通讯的超时时间
controller.socket.timeout.ms
# controller-to-broker-channels消息队列的尺寸大小
controller.message.queue.size=10
# replicas响应leader的最长等待时间，若是超过这个时间，就将replicas排除在管理之外
replica.lag.time.max.ms
# 是否允许控制器关闭broker ,若是设置为true,会关闭所有在这个broker上的leader，并转移到其他broker
controlled.shutdown.enable
# 控制器关闭的尝试次数
controlled.shutdown.max.retries
# 每次关闭尝试的时间间隔
controlled.shutdown.retry.backoff.ms
# 如果relicas落后太多,将会认为此partition
relicas已经失效。而一般情况下,因为网络延迟等原因,总会导致replicas中消息同步滞后。如果消息严重滞后,leader将认为此relicas网络延迟较大或者消息吞吐能力有限。在broker数量较少,或者网络不足的环境中,建议提高此值.
replica.lag.max.messages
#leader与relicas的socket超时时间
replica.socket.timeout.ms=
# leader复制的socket缓存大小
replica.socket.receive.buffer.bytes=64
# replicas每次获取数据的最大字节数
replica.fetch.max.bytes
# replicas同leader之间通信的最大等待时间，失败了会重试
replica.fetch.wait.max.ms
# 每一个fetch操作的最小数据尺寸,如果leader中尚未同步的数据不足此值,将会等待直到数据达到这个大小
replica.fetch.min.bytes
# leader中进行复制的线程数，增大这个数值会增加relipca的IO
num.replica.fetchers
# 每个replica将最高水位进行flush的时间间隔
replica.high.watermark.checkpoint.interval.ms
# 是否自动平衡broker之间的分配策略
auto.leader.rebalance.enable
# leader的不平衡比例，若是超过这个数值，会对分区进行重新的平衡
leader.imbalance.per.broker.percentage
# 检查leader是否不平衡的时间间隔
leader.imbalance.check.interval.seconds
# 客户端保留offset信息的最大空间大小
offset.metadata.max.bytes
#############################Consumer
#############################
# Consumer端核心的配置是group.id、zookeeper.connect
# 决定该Consumer归属的唯一组ID，By setting
the same group id multiple processes indicate that they are all part of the same consumer group.
# 消费者的ID，若是没有设置的话，会自增
consumer.id
# 一个用于跟踪调查的ID ，最好同group.id相同
&group_id&
# 对于zookeeper集群的指定，必须和broker使用同样的zk配置
zookeeper.connect=debugo01:2182,debugo02:2182,debugo03:2182
# zookeeper的心跳超时时间，查过这个时间就认为是无效的消费者
zookeeper.session.timeout.ms
# zookeeper的等待连接时间
zookeeper.connection.timeout.ms
# zookeeper的follower同leader的同步时间
zookeeper.sync.time.ms
# 当zookeeper中没有初始的offset时，或者超出offset上限时的处理方式
# smallest ：重置为最小值
# largest:重置为最大值
# anything else：抛出异常给consumer
auto.offset.reset
# socket的超时时间，实际的超时时间为max.fetch.wait
+ socket.timeout.ms.
socket.timeout.ms=
# socket的接收缓存空间大小
socket.receive.buffer.bytes=64
#从每个分区fetch的消息大小限制
fetch.message.max.bytes
# true时，Consumer会在消费消息后将offset同步到zookeeper，这样当Consumer失败后，新的consumer就能从zookeeper获取最新的offset
auto.commit.enable
# 自动提交的时间间隔
auto.commit.interval.ms
# 用于消费的最大数量的消息块缓冲大小，每个块可以等同于fetch.message.max.bytes中数值
queued.max.message.chunks
# 当有新的consumer加入到group时,将尝试reblance,将partitions的消费端迁移到新的consumer中,
该设置是尝试的次数
rebalance.max.retries
# 每次reblance的时间间隔
rebalance.backoff.ms
# 每次重新选举leader的时间
refresh.leader.backoff.ms
# server发送到消费端的最小数据，若是不满足这个数值则会等待直到满足指定大小。默认为1表示立即接收。
fetch.min.bytes
# 若是不满足fetch.min.bytes时，等待消费端请求的最长等待时间
fetch.wait.max.ms
# 如果指定时间内没有新消息可用于消费，就抛出异常，默认-1表示不受限
consumer.timeout.ms
#############################Producer#############################
# 核心的配置包括：
# metadata.broker.list
# request.required.acks
# producer.type
# serializer.class
# 消费者获取消息元信息(topics, partitions
and replicas)的地址,配置格式是：host1:port1,host2:port2，也可以在外面设置一个vip
metadata.broker.list
#消息的确认模式
# 0：不保证消息的到达确认，只管发送，低延迟但是会出现消息的丢失，在某个server失败的情况下，有点像TCP
# 1：发送消息，并会等待leader 收到确认后，一定的可靠性
# -1：发送消息，等待leader收到确认，并进行复制操作后，才返回，最高的可靠性
request.required.acks
# 消息发送的最长等待时间
request.timeout.ms
# socket的缓存大小
send.buffer.bytes=100*1024
# key的序列化方式，若是没有设置，同serializer.class
key.serializer.class
# 分区的策略，默认是取模
partitioner.class=kafka.producer.DefaultPartitioner
# 消息的压缩模式，默认是none，可以有gzip和snappy
compression.codec
# 可以针对默写特定的topic进行压缩
compressed.topics=null
# 消息发送失败后的重试次数
message.send.max.retries
# 每次失败后的间隔时间
retry.backoff.ms
# 生产者定时更新topic元信息的时间间隔 ，若是设置为0，那么会在每个消息发送后都去更新数据
topic.metadata.refresh.interval.ms
# 用户随意指定，但是不能重复，主要用于跟踪记录消息
client.id=&&
# 异步模式下缓冲数据的最大时间。例如设置为100则会集合100ms内的消息后发送，这样会提高吞吐量，但是会增加消息发送的延时
queue.buffering.max.ms
# 异步模式下缓冲的最大消息数，同上
queue.buffering.max.messages
# 异步模式下，消息进入队列的等待时间。若是设置为0，则消息不等待，如果进入不了队列，则直接被抛弃
queue.enqueue.timeout.ms
# 异步模式下，每次发送的消息数，当queue.buffering.max.messages或queue.buffering.max.ms满足条件之一时producer会触发发送。
batch.num.messages=200
没有更多推荐了，
加入CSDN，享受更精准的内容推荐，与500万程序员共同成长！Kafka是如何实现高吞吐率的 - 简书
Kafka是如何实现高吞吐率的
Kafka是分布式消息系统，需要处理海量的消息，Kafka的设计是把所有的消息都写入速度低容量大的硬盘，以此来换取更强的存储能力，但实际上，使用硬盘并没有带来过多的性能损失
kafka主要使用了以下几个方式实现了超高的吞吐率
顺序读写：
Kafka数据不是实时写入硬盘，采用内存映射文件(分页存储)来利用内存提高I/O效率，利用操作系统的页来实现物理内存映射，映射完后物理内存上的操作会被同步到硬盘上，
零拷贝：减少了系统的两次上下文切换，
原来：文件复制到系统内核空间---复制到用户空间---复制到内核空间--发送网卡，使用socket发送。
现在：直接从内核空间到内核空间---发送给网卡
一个topic队列分成多个partition区，partition区分为过个segment段，每个segment段对应一个文件；
批量发送消息
缓存在内存中，达到阈值或者时间，则发送，减少了网络I/O
GZIP、Snappy，减少了磁盘I/O和网络的I/O
kafka的消息是不断追加到文件中的，这个特性使kafka可以充分利用磁盘的顺序读写性能
顺序读写不需要硬盘磁头的寻道时间，只需很少的扇区旋转时间，所以速度远快于随机读写
Kafka官方给出了测试数据(Raid-5，7200rpm)：
顺序 I/O: 600MB/s
随机 I/O: 100KB/s
先简单了解下文件系统的操作流程，例如一个程序要把文件内容发送到网络
这个程序是工作在用户空间，文件和网络socket属于硬件资源，两者之间有一个内核空间
在操作系统内部，整个过程为：
在Linux kernel2.2 之后出现了一种叫做"零拷贝(zero-copy)"系统调用机制，就是跳过“用户缓冲区”的拷贝，建立一个磁盘空间和内存的直接映射，数据不再复制到“用户态缓冲区”
系统上下文切换减少为2次，可以提升一倍的性能
kafka的队列topic被分为了多个区partition，每个partition又分为多个段segment，所以一个队列中的消息实际上是保存在N多个片段文件中
通过分段的方式，每次文件操作都是对一个小文件的操作，非常轻便，同时也增加了并行处理能力
Kafka允许进行批量发送消息，先将消息缓存在内存中，然后一次请求批量发送出去
比如可以指定缓存的消息达到某个量的时候就发出去，或者缓存了固定的时间后就发送出去
如100条消息就发送，或者每5秒发送一次
这种策略将大大减少服务端的I/O次数
Kafka还支持对消息集合进行压缩，Producer可以通过GZIP或Snappy格式对消息集合进行压缩
压缩的好处就是减少传输的数据量，减轻对网络传输的压力
Producer压缩之后，在Consumer需进行解压，虽然增加了CPU的工作，但在对大数据处理上，瓶颈在网络上而不是CPU，所以这个成本很值得
姓名：周小蓬
转载自：http://blog.csdn.net/YChenFeng/article/details/ [嵌牛导读] Kafka Kafka是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）...
KafkaKafka是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟...
Kafka是分布式消息系统，需要处理海量的消息，Kafka的设计是把所有的消息都写入速度低容量大的硬盘，以此来换取更强的存储能力，但实际上，使用硬盘并没有带来过多的性能损失kafka主要使用了以下几个方式实现了超高的吞吐率 顺序读写 kafka的消息是不断追加到文件中的，这...
Kafka入门经典教程-Kafka-about云开发
http://www.aboutyun.com/thread-.html 1.Kafka独特设计在什么地方？2.Kafka如何搭建及创建topic、发送消息、消费消息？3.如何书写Kafka程序？4....
一、Kafka简介 本文综合了我之前写的kafka相关文章，可作为一个全面了解学习kafka的培训学习资料。 转载请注明出处 :本文链接（http://blog.csdn.net/suifeng3051/article/details/） 1.1 背景历史 ...
随着现代人越来越追求个性化的生活，以往以样板式生产为主的家居产品，如今掀起了定制潮流。所谓全屋定制，即定制家具厂家根据消费者的需求而为其量身定做的家具产品，其可最大限度地满足消费者的需求。然而在我国，定制家具尚属新兴产品，大部分消费者对其缺乏全面了解，不知在选购时应该注意哪...
监控器里的人不过是一个个赌注，被贪婪和恐惧所控制着。这就是这个世界的运作法则。 游戏，从未结束 现场 他从来不信教，但此时此刻他太需要一些精神上的寄托了。他在胸前胡乱地划着十字，右膝盖轻吻地面。 “万能的主啊，上帝，耶稣......各方神明，我知道你们大概不会想保佑我这个无...
8月8日21时19分，四川九寨沟发生了7.0级地震,当时我们正吃晚饭。 妹妹在阳台上，率先感应到地震了，可那时震的很微，就等了会儿，发觉震感越来越强就进来告诉大家。这才发现冰箱旁边的花儿不停地抖动，门也好像有人在摇，侄女还准备去开门，一摸门把手发觉不对才缩了回来。于是大家陆...
在我的家乡县城里，有一家企业，每年都会帮助一些贫困家庭的大学生，会在全县范围内挑出一些家庭经济条件比较困难，且品学兼优的学生，每年资助直到读完大学。 其实老实说这家企业已经算是挺好的了，在邻近的几个县也都这样做，坚持资助也都有十几年了。 每年暑假企业的领导们都会空出一天来，...
二、资产损失扣除政策