是整个同步机制的主入口,主要组织的逻辑有:
实现HelixWorkerOnlineOfflineStateModelFactory和OnlineOfflineStateModel,OnlineOfflineStateModel可以理解成是一个监听器。实例在上下线切换时可以监听到。
helixZkManager = HelixManagerFactory.getZKHelixManager(helixClusterName, instanceId, InstanceType.PARTICIPANT, zkServer)
val stateMachineEngine: StateMachineEngine = helixZkManager.getStateMachineEngine()
// register the MirrorMaker worker
val stateModelFactory = new HelixWorkerOnlineOfflineStateModelFactory(instanceId, fetchNum, connectorMap)
stateMachineEngine.registerStateModelFactory("OnlineOffline", stateModelFactory)
helixZkManager.connect()
helixAdmin = helixZkManager.getClusterManagmentTool
class HelixWorkerOnlineOfflineStateModelFactory(final val instanceId: String, final val fetchNum: Int,
final val connectorMap: ConcurrentHashMap[String, KafkaConnector]) extends StateModelFactory[StateModel] {
override def createNewStateModel(partitionName: String) = new OnlineOfflineStateModel(instanceId, connectorMap)
// register mm instance
class OnlineOfflineStateModel(final val instanceId: String, final val connectors: ConcurrentHashMap[String, KafkaConnector]) extends StateModel {
def onBecomeOnlineFromOffline(message: Message, context: NotificationContext) = {
// add topic partition on the instance
connectorMap.get(getFetcherId(message.getResourceName, message.getPartitionName.toInt)).addTopicPartition(message.getResourceName, message.getPartitionName.toInt)
}
def onBecomeOfflineFromOnline(message: Message, context: NotificationContext) = {
// delete topic partition on the instance
connectorMap.get(getFetcherId(message.getResourceName, message.getPartitionName.toInt)).deleteTopicPartition(message.getResourceName, message.getPartitionName.toInt)
}
def onBecomeDroppedFromOffline(message: Message, context: NotificationContext) = {
// do nothing
}
private def getFetcherId(topic: String, partitionId: Int): String = {
"" + Utils.abs(31 * topic.hashCode() + partitionId) % fetchNum
}
}
}
CompactConsumerFetcherThread
是继承自Kafka提供的ShutdownableThread
,ShutdownableThread
内部会在isRunning标志位ok的情况下以spin的形式一直调用doWork
方法。
override def run(): Unit = {
info("Starting ")
try{
while(isRunning.get()){
doWork()
}
} catch{
case e: Throwable =>
if(isRunning.get())
error("Error due to ", e)
}
shutdownLatch.countDown()
info("Stopped ")
}
当doWork方法准备好了FetchRequest实例就要靠processFetchRequest方法来拉数据给partitionInfoMap中的PartitionTopicInfo实例中的队列了。简单过程如下:
手机扫一扫
移动阅读更方便
你可能感兴趣的文章