文档章节

key为null时Kafka会将消息发送给哪个分区?

m
 miscellanea
发布于 2015/08/24 19:46
字数 817
阅读 1970
收藏 2
点赞 1
评论 0

当你编写kafka Producer时, 会生成KeyedMessage对象。

KeyedMessage<K, V> keyedMessage = new KeyedMessage<>(topicName, key, message)

这里的key值可以为空,在这种情况下, kafka会将这个消息发送到哪个分区上呢?依据Kafka官方的文档, 默认的分区类会随机挑选一个分区:

The third property  "partitioner.class" defines what class to use to determine which Partition in the Topic the message is to be sent to. This is optional, but for any non-trivial implementation you are going to want to implement a partitioning scheme. More about the implementation of this class later. If you include a value for the key but haven't defined a partitioner.class Kafka will use the default partitioner. If the key is null, then the Producer will assign the message to a random Partition.

但是这句话相当的误导人。

从字面上来讲,这句话没有问题, 但是这里的随机是指在参数"topic.metadata.refresh.ms"刷新后随机选择一个, 这个时间段内总是使用唯一的分区。 默认情况下每十分钟才可能重新选择一个新的分区。 但是相信大部分的程序员和我一样, 都理解成每个消息都会随机选择一个分区。
可以查看相关的代码:

private def getPartition(topic: String, key: Any, topicPartitionList: Seq[PartitionAndLeader]): Int = {
    val numPartitions = topicPartitionList.size    if(numPartitions <= 0)
      throw new UnknownTopicOrPartitionException("Topic " + topic + " doesn't exist")
    val partition =
      if(key == null) {
        // If the key is null, we don't really need a partitioner
        // So we look up in the send partition cache for the topic to decide the target partition
        val id = sendPartitionPerTopicCache.get(topic)
        id match {
          case Some(partitionId) =>
            // directly return the partitionId without checking availability of the leader,
            // since we want to postpone the failure until the send operation anyways
            partitionId          case None => 
            val availablePartitions = topicPartitionList.filter(_.leaderBrokerIdOpt.isDefined)
            if (availablePartitions.isEmpty)
              throw new LeaderNotAvailableException("No leader for any partition in topic " + topic)
            val index = Utils.abs(Random.nextInt) % availablePartitions.size
            val partitionId = availablePartitions(index).partitionId
            sendPartitionPerTopicCache.put(topic, partitionId)
            partitionId        }
      } else
        partitioner.partition(key, numPartitions)
    if(partition < 0 || partition >= numPartitions)
      throw new UnknownTopicOrPartitionException("Invalid partition id: " + partition + " for topic " + topic +
        "; Valid values are in the inclusive range of [0, " + (numPartitions-1) + "]")
    trace("Assigning message of topic %s and key %s to a selected partition %d".format(topic, if (key == null) "[none]" else key.toString, partition))
    partition  }

如果key为null, 它会从sendPartitionPerTopicCache查选缓存的分区, 如果没有,随机选择一个分区,否则就用缓存的分区。

LinkedIn工程师Guozhang Wang在邮件列表中解释了这一问题,
最初kafka是按照大部分用户理解的那样每次都随机选择一个分区, 后来改成了定期选择一个分区, 这是为了减少服务器段socket的数量。不过这的确很误导用户,据称0.8.2版本后又改回了每次随机选取。但是我查看0.8.2的代码还没看到改动。

所以,如果有可能,还是为KeyedMessage设置一个key值吧。

当你编写kafka Producer时, 会生成KeyedMessage对象。

KeyedMessage<K, V> keyedMessage = new KeyedMessage<>(topicName, key, message)

这里的key值可以为空,在这种情况下, kafka会将这个消息发送到哪个分区上呢?依据Kafka官方的文档, 默认的分区类会随机挑选一个分区:

The third property  "partitioner.class" defines what class to use to determine which Partition in the Topic the message is to be sent to. This is optional, but for any non-trivial implementation you are going to want to implement a partitioning scheme. More about the implementation of this class later. If you include a value for the key but haven't defined a partitioner.class Kafka will use the default partitioner. If the key is null, then the Producer will assign the message to a random Partition.

但是这句话相当的误导人。


本文转载自:http://ju.outofmemory.cn/entry/112202

共有 人打赏支持
m
粉丝 6
博文 71
码字总数 22525
作品 0
海淀
kafka java编程

自定义系列化方式Encoder kafka自带的序列化方式 DefaultEncoder默认的这个Encoder事实上不做任何处理,接收到什么byte[]就返回什么byte[]: class DefaultEncoder(props: VerifiablePropert...

张欢19933 ⋅ 2016/01/25 ⋅ 0

kafka 文档 (三)producer(生产者)和高级消费者

API Producer API 此处只简介一个procedure的例子 生产类是用来创建新消息的主题和可选的分区。 如果使用Java你需要包括几个包和支持类: import kafka.javaapi.producer.Producer; import k...

q5725827 ⋅ 2014/05/29 ⋅ 0

Kafka分区机制介绍与示例

Kafka中可以将Topic从物理上划分成一个或多个分区(Partition),每个分区在物理上对应一个文件夹, 以”topicName_partitionIndex”的命名方式命名,该文件夹下存储这个分区的所有消息(.log...

HZCoder ⋅ 2016/03/01 ⋅ 0

Kafka实战解惑

Kafka是LinkedIn使用Scala开发的一个分布式消息中间件,它以水平扩展能力和高吞吐率著称,被广泛用于日志处理、ETL等应用场景。Kafka具有以下主要特点: 消息的发布、订阅均具有高吞吐量:据...

量子世界的我 ⋅ 2016/11/16 ⋅ 0

D20 Kafka及其常用命令

一、Kafka客户端常用命令: 启动 nohup bin/kafka-server-start.sh config/server.properties & 停止 bin/kafka-server-stop.sh 创建kafka topic bin/kafka-topics.sh --zookeeper server:21......

u014253445 ⋅ 2017/08/26 ⋅ 0

Learning Apache Kafka 2nd Edition读书笔记

Chap 1 Kafka简介 1.Apache Kafka是一款开源的,分布式的,基于分区、日志提交和订阅推送的消息系统。设计用于: 持久化消息到硬盘,TB级别 高吞吐量,每秒数百M读写 分布式,支持弹性伸缩 ...

GunnerAha ⋅ 04/10 ⋅ 0

架构设计:系统间通信(29)——Kafka及场景应用(中2)

接上文:《架构设计:系统间通信(28)——Kafka及场景应用(中1)》 4-3、复制功能 我们在上文中已经讨论了Kafka使用分区的概念存储消息,一个topic可以有多个分区它们分布在整个Kafka集群的...

yinwenjie ⋅ 2016/05/06 ⋅ 0

Kafka分区分配计算(分区器Partitions)

KafkaProducer在调用send方法发送消息至broker的过程中,首先是经过拦截器Inteceptors处理,然后是经过序列化Serializer处理,之后就到了Partitions阶段,即分区分配计算阶段。在某些应用场景...

u013256816 ⋅ 2017/12/03 ⋅ 0

Kafka学习笔记

Apache Kafka 一、消息队列分类 1.1 点对点 1.2 发布/订阅 二、消息队列对比 2.1 RabbitMQ 2.2 ZeroMQ 2.3 ActiveMQ 2.4 Redis 三、Kafka简介 3.1 Kafka简介 3.2 Kafka特点 四、Kafka架构 五...

凡夫俗子_ ⋅ 2017/06/14 ⋅ 0

Kafka介绍, kafka主题的副本机制

1、Kafka概览 Apache下的项目Kafka(卡夫卡)是一个分布式流处理平台,它的流行是因为卡夫卡系统的设计和操作简单,能充分利用磁盘的顺序读写特性。kafka每秒钟能有百万条消息的吞吐量,因此很...

qq_31598113 ⋅ 2017/04/28 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

前台对中文编码,后台解码

前台:encodeURI(sbzt) 后台:String param = URLDecoder.decode(sbzt,"UTF-8");

west_coast ⋅ 41分钟前 ⋅ 0

VS2015配置并运行汇编(一步一步照图做)【vs2017的链接在最后】

前言 我是上学期学的汇编,因为有vs又不想用课上教的麻烦的dosbox以及masm32,但是一直没找到高亮插件和能调试的(难在运行不了而找不到答案上,出现的错误在最后放出,还请先达们不吝指点)...

simpower ⋅ 50分钟前 ⋅ 0

一起读书《深入浅出nodejs》-node模块机制

node 模块机制 前言 说到node,就不免得提到JavaScript。JavaScript自诞生以来,经历了工具类库、组件库、前端框架、前端应用的变迁。通过无数开发人员的努力,JavaScript不断被类聚和抽象,...

小草先森 ⋅ 53分钟前 ⋅ 0

Java桌球小游戏

其实算不上一个游戏,就是两张图片,不停的重画,改变ball图片的位置。一个左右直线碰撞的,一个有角度碰撞的。 左右直线碰撞 package com.bjsxt.test;import javax.swing.*;import j...

森林之下 ⋅ 今天 ⋅ 0

你真的明白RPC 吗?一起来探究 RPC 的实质

你真的明白RPC 吗?一起来探究 RPC 的实质 不论你是科班出身还是半路转行,这么优秀的你一定上过小学语文,那么对扩句和缩句你一定不陌生。缩句就是去除各种修饰提炼出一句话的核心,而不失基...

AI9o後 ⋅ 今天 ⋅ 0

z-index设置失效?

今天碰到了一个问题,就是在给li设置提示框的时候,有用到遮罩效果,本来想把对应的出现在最顶层,可是不管将li设置的z-index值设为多大,li都没有出现在遮罩层之上。 我在网上查了z-index设...

IrisHunag ⋅ 今天 ⋅ 0

CyclicBarrier、CountDownLatch以及Semaphore使用及其原理分析

CyclicBarrier、CountDownLatch以及Semaphore是Java并发包中几个常用的并发组件,这几个组件特点是功能相识很容易混淆。首先我们分别介绍这几个组件的功能然后再通过实例分析和源码分析其中设...

申文波 ⋅ 今天 ⋅ 0

Java对象的序列化与反序列化

Java对象的序列化与反序列化

Cobbage ⋅ 今天 ⋅ 0

Sqoop

1.Sqoop: 《=》 SQL to Hadoop 背景 1)场景:数据在RDBMS中,我们如何使用Hive或者Hadoop来进行数据分析呢? 1) RDBMS ==> Hadoop(广义) 2) Hadoop ==> RDBMS 2)原来可以通过MapReduce I...

GordonNemo ⋅ 今天 ⋅ 0

全量构建和增量构建的区别

1.全量构建每次更新时都需要更新整个数据集,增量构建只对需要更新的时间范围进行更新,所以计算量会较小。 2.全量构建查询时不需要合并不同Segment,增量构建查询时需要合并不同Segment的结...

无精疯 ⋅ 今天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部