4.Kafka Producer机制优化-提高发送消息可靠性
2018-03-11 16:55
507 查看
名称解释:
Broker:负责消息的存储和转发,也可以叫消息中介节点Topic:每种消息的分类叫做主题(Topic)。
Partition:每一个Topic被切分为多个Partitions。
背景
Producer构造Message对象时,传入key参数,当Producer发送Message,会根据key确定目标Partition,当Kafka集群中某个Partition所有存活的节点都失效或挂掉。会造成Producer尝试重新发送message.send.max.retries(默认值为3)次后抛出Exception,每次尝试都会休眠一定时间(默认值为100ms)。用户捕捉到异常其结果,停止发送会阻塞,继续发送消息会丢失。解决思路
Kafka中默认发送消息方式不变,给用户提供一种可选择方式,增加一个消息发送失效转移的开关,当Producer发送到目标Partition的(所有副本本来存活的)节点都失效或挂掉,就转发到其他Partition上。M:表示Partition的主分区。
S:表示Partition的从分区。下面图2所示,消息轮询发送到Partition的0-3上。
图1
上图本来一条消息是发送到Partition-0所在Broker的,当Partition-0所在Broker全部失效或挂掉后,如下图所示。
图2
消息(轮询均衡)发送到其他Partition上,待失效的Brokers恢复,发送消息恢复到图1状态。
Producer代码修改
代码修改如下:ProducerConfig.scala类代码中增加一行:
val sendSwitchBrokerEnabled = props.getBoolean(“send.switch.broker.enable”, false)
DefaultEventHandler.scala类增加一行
val isSendSwitchBrokerEnabled = config.sendSwitchBrokerEnabled
DefaultEventHandler.scala类中增加如下方法:
private def getPartition(topic: String, key: Any, topicPartitionList: Seq[PartitionAndLeader], isSendSwitchBrokerEnabled: Boolean = false): Int = { val numPartitions = topicPartitionList.size if(numPartitions <= 0) throw new UnknownTopicOrPartitionException("Topic " + topic + " doesn't exist") var 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 { 4000 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)) if (isSendSwitchBrokerEnabled) { if(partitionsLeaderInTopicsCache.isEmpty()) { val availablePartitions = topicPartitionList.map { partitionAndLeader => if(partitionAndLeader.leaderBrokerIdOpt.isDefined) { partitionsLeaderInTopicsCache.put(TopicAndPartition(topic, partitionAndLeader.partitionId), partitionAndLeader.leaderBrokerIdOpt.get) } } } if(!partitionsLeaderInTopicsCache.containsKey(TopicAndPartition(topic, partition))) { 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 partition = availablePartitions(index).partitionId } } partition }
相关文章推荐
- Kafka Producer机制优化-提高发送消息可靠性
- Android Looper消息发送机制解析
- RocketMQ原理解析-producer 4.发送分布式事物消息
- MFC中发送自定义消息机制,PostMessage和SendMessage方式
- RocketMQ原理解析-producer 4.发送分布式事物消息
- RabbitMQ(python实现)学习之二:Producer发送消息至多个消息队列queue(广播消息)
- kafka producer发送消息 Failed to update metadata after问题
- handler消息发送机制
- android QMI机制---modem消息发送
- MFC 消息发送机制
- 新浪微博发送消息和授权机制原理(WeiboSDK)
- spring activeMQ 整合(二): 重发机制(消息发送失败后的重新发送)
- 消息发送机制的利用
- Kafka producer无法发送消息解决办法
- runtime1 - 消息发送机制
- Strom 消息可靠性保障机制和Ack原理
- adnroid_Handler消息发送处理机制
- rocketmq-producer之发送事物消息
- Kafka Producer 发送消息源码阅读
- APM:拦截器里发送跟踪消息的机制和时机