Kafka-Broker-Replica-Controller-Producer-Consumer

producer

0

返回

KafkaApis.scalaApiKeys.LIST_OFFSETS => handleListOffsetRequest(request)

3

TIMEINDEX

 private def processStartup(): Unit = {    zkClient.registerZNodeChangeHandlerAndCheckExistence(controllerChangeHandler)    elect()           //选主  }

1，外界如何通信内部数据如何写磁盘

app

getAssignorInstances分配的策略

Paration：Leader

有

createReplicaFetcherManager

1

ControllerEventManager

 new ConsumerCoordinator

Consumer.commitAsync()

groupCoordinator.handleJoinGroup

如果read这个事件是常注册handler是多线程多线程：乱序问题

TopicCommond.scala

subscriptions.fetchablePartitions

为了更新各分区的offset的位置

font color=\"#ff9800\

4

// send any new fetches (won't resend pending fetches)        fetcher.sendFetches();

MetadataCache

如果offset  是 Long 8个字节base：基地址 8个字节的long相对offset ： 4个字节的int用base + 相对offset得到真正的msg的offset

Replica同步机制KafkaApis.scala、KafkaServer.scala

Kafka服务重启，内存是空的，GroupCoordinator如何初始化的数据

readFromLocalLog

如果调用了flush，一般不会调用的log.flush()offsetIndex.flush()timeIndex.flush()txnIndex.flush()调用native的fsync（）channel.force(true);fsync() ，fdatasync()mmap.force()

sendResponseCallback

channel.write(buffer);只写到了内核pagecache

Consumer

GroupCoordinator组协调器

client.font color=\"#f44336\

doAutoCommitOffsetsAsync()

jvmbrokerapp

serverChannelnioSelector

请求

socketqueue

leader

delayedProducePurgatory

 super(ApiKeys.JOIN_GROUP);

RequestFuture<Void> future = sendOffsetCommitRequest(offsets);

ACK=0/1/-1

1、newConnections

Leader：respones内容先返回HW：3DATA: 4返回结果后Leader的HW更新为4 （根据follower请求过来的LEO修改）Leader@HW：4

maybeAutoCommitOffsetsAsync(timer.currentTimeMs())

trySend(timer.currentTimeMs());

scalable

RequestChannelrequestQueue（带着processor）

一个broker上只有一个GroupCoordinator，会维护不同组的offset元数据，之间是隔离的持久化写入log

new ConsumerMetadata

Consumer.commitSync()

locallog

val brokerInfo = createBrokerInfoval brokerEpoch = zkClient.registerBroker(brokerInfo)

initiateJoinGroup

maybeIncrementLeaderHW处理HW

ReplicaManager

handleCompletedReceives

kafka，依赖内核kernel的脏页刷写机制LogFlushSchedulerIntervalMs = Long.MaxValue

GroupCoordinatorgroup-1的元数据

多部门多项目组多客户端并发

nextFetchOffset = lastRecord.offset() + 1;

2，responseQueuewrite 事件的注册

if (completedFetch.nextFetchOffset == position.offset) 

 new Fetcher<>

2

kernel

log

consumer2

sendJoinGroupRequest发送加入组的请求，并处理返回响应

进入了ReplicaManager中的同步机制

Processor

Topic1：P3

def partitionFor(groupId: String): Int = Utils.abs(groupId.hashCode) % groupMetadataTopicPartitionCount获得:offset的partition ID，根据partition id获取此分区所在leader的broker，然后将broker的EndPoint的内容返回，以便consumer去建立连接，并在broker上建立GroupCoordinator

集群多进程

fetcher.resetOffsetsIfNeeded()

ACK为-1所有副本同步

P分区：offset

consumer1

LEO

partition

producer-01

KafkaApiscase ApiKeys.LEADER_AND_ISR => handleLeaderAndIsrRequest(request)

0/1

KafkaApis.scalacase ApiKeys.JOIN_GROUP => handleJoinGroupRequest(request)

queue

delayedFetchPurgatory.tryCompleteElseWatch

segment

maybeExpandIsr

Group -1

ApiKeys.OFFSET_COMMIT

broker

acceptor

KafkaServer.scalagroupCoordinator.startup(() => zkClient.getTopicPartitionCount(Topic.GROUP_METADATA_TOPIC_NAME).getOrElse(config.offsetsTopicPartitions))GROUP_METADATA_TOPIC_NAME = \"__consumer_offsets\"  公共的topic，名下有50个分区

异步提交

Topic：test    有两个分区p0、p1consumer GroupId组名为AAhashcode后为21122112%50 取模后12，也就是12号分区 所以Group组名AA的offset保存在Topic：_consumer_offset的12号分区上span style=\"font-size: inherit;\

brocker

有没有获取到数据

OFFSETINDEX （base 基地址）4B(相对offset)：4B(log文件offset)

!ensureActiveGroup(waitForJoinGroup ? timer : time.timer(0L))上面建立真正的连接后，此处开始给Group Coordinator发送消息，要加入组，谁先入组，谁是leader

processFetchRequest

producer-01：生产一条数据LastOffset：3requireOffset：4LEO：4

LOG

Broker

response处理

Controller

C

KafkaApis.scalacase ApiKeys.FIND_COORDINATOR => handleFindCoordinatorRequest(request)

doSyncGroup

Topic1：P1

因为LEO发生变化所以producer-01的LEO也会变化为5

客户端

producer-02

appendRecords

newKafkaConsumer

自身分区有序性

For循环selector.completedReceives

轮询分配

根据各个分区的offset来拉取数据

case ApiKeys.SYNC_GROUP => handleSyncGroupRequest(request)

new ConsumerNetworkClient

producer-01：生产一条数据LastOffset：3requireOffset：4LEO：5

OffsetIndex

返回结果future.addListener为onSuccess

fetchFromLeader

源码入口文件：KafkaApis.scalaKafkaServer.scala

read 常注册，读一次后，mute-> 把read事件取消注册

分区数 > 节点数1.业务膨胀：预分区 方便扩缩容2.业务隔离性：AKF

updateAssignmentMetadataIfNeededMetadata处理

broker进行network read进行mute、unmute操作一条条的读取

分布式

responseData.forKeyValue等待返回结果

updateFetchPositions(timer)

Paration：Leader处理Fetch

Set<SelectionKey> readyKeys = this.nioSelector.selectedKeys();

2-1

3-3

shouldRollsegment文件形成的因素删除7天的segment文件：依据使用场景配置1，size > rollParams.maxSegmentBytes - rollParams.messagesSize当前段文件的使用量是不是大于了允许的使用量，如果大于，那就需要创建新的段文件2，(size > 0 && reachedRollMs)当前段文件有数据，空间够，但是寿命已尽3， offsetIndex.isFull  10MB4，timeIndex.isFull  10MB5，!canConvertToRelativeOffset(rollParams.maxOffsetInMessages)      不能造成 相对offset 溢出

startup()

if (coordinator != null && !font color=\"#ff9800\

值为50

FIND_COORDINATOR

joinGroupIfNeeded

pollForFetches

handleSyncGroup

commitOffsetsAsync

response的处理

request

adminZkClient.createTopic

groupCoordinator.onElection(partition.partitionId)

HW

follower2

Topic1：P2

5

组内的consumer会选举出一个Leader，来分配分区

KafkaController

相互对应的

request处理

/* setup zookeeper */ initZkClient(time)

follow2：第二次请求带着LEO：4       HW：3接收返回结果后：follow2：LEO：5HW：3

Leader生产时间线

发送给broker

apply

如何根据算法找到__consumer_offset分区的leader所在的broker信息，并返回给consumer以便去建立连接，并在broker上建立GroupCoordinator一个broker上只有一个GroupCoordinator，但是GroupCoordinator里的group的状态和元数据是隔离的一个分区对应若干个组分治（路由、索引、映射）

mmap = raf.getChannel.map(10MB)

LogManager

getBrokerMetadatas

让socket的channel在 selector上的read事件 取消掉

fetchMessages

-1

持久化写入log

1-1

ControllerEventcase classprocess()

redis单线程

connect

read 常注册

channel.setSend(send);

P.....

send发送的Request处理

onControllerFailover

createTopicWithAssignment

kafka的吞吐建立在分布式上

P1

scriptions里包含topic

assignReplicasToBrokersRackUnaware分区分配规则，可以避免热点这种分布式的痛点。该逻辑可以直接拿来使用。

Group-3

follower3

生产者 会在socket中  按顺序 把消息发过来问题是你多线程的handler会乱序，如何保证顺序？在一个socket中，取一个消息放队列，然后handler，然后取第二条消息放入队列。。。要考虑并发的问题，饿死的问题 ，隔离真实环境：一个topic 有多个生产者，且生产者在不同的物理进程，所以，会有多个连接进来且都想写一个分区，那kafka可以保障顺序吗？ 不可以

V.S

partitionStates

for (TopicPartition partition : fetchablePartitions()) 

sendAlterIsrRequest

FindCoordinatorRequest.Builder requestBuilder =                new FindCoordinatorRequest.Builder(                        new FindCoordinatorRequestData()                            .setKeyType(CoordinatorType.GROUP.id())                            .setKey(font color=\"#f44336\

partitionfollower

DelayedProduce的tryComplete

通过GroupCoordinator知道要消费哪个leader分区上的数据

接受

ACK

Leader：respones内容先返回HW：3DATA: 3返回结果后Leader的HW更新为3（根据follower请求过来的LEO修改）Leader@HW：3

先跟任意一台broker建立连接；再根据算法找到真正的leader的broker，创建Group Coordinator并建立连接；然后加入Group Coordinator选出consumer的Leader，并分配策略（同组的consumer Leader会进行partition的分配，哪个consumer连接哪个partition)consumer leader将分配策略同步给Group Coordinator，以便consumer的follow去Group Coordinator上同步获取策略

啊 mute 啊 mute 啊 mute  啊

nextInLineFetch = initializeCompletedFetch(records);

协商 空间的问题

Kafka 2.8.0kafka-broker-kafkaserver

tradeoff有序性保证整体的节奏

返回的response

ConsumerCoordinator消费者协调器

将send又封装了一层

partitionleader

handleListOffsetRequestV1AndAbove(request)

nioSelector

setAssigments参数不一样而已

partition.checkEnoughReplicasReachOffset(status.requiredOffset)

多线程中顺序保障问题

为了在内核堵住socket的下一条消息

lookupCoordinator()

groupMetadataTopicPartitionCount值为50

follow2：第一次请求带着LEO：3接收返回结果后：follow2：LEO：4HW：3

如何实现ctrl+c 关闭服务的操作：      Runtime.getRuntime().addShutdownHook(new Thread(\"kafka-shutdown-hook\") {        override def run(): Unit = kafkaServerStartable.shutdown()      })

三次握手

replicaManager = createReplicaManager(isShuttingDown)

mute静音

new SubscriptionState订阅了哪些topic

Group-2

持久化同步、异步、自动提交最终都是同步给GroupCoordinator

ControllerEventThread

sendFindCoordinatorRequest(node)

!ensureCoordinatorReady(timer)为了建立真正的连接

ApiKeys.LEADER_AND_ISR

减轻当下master无磁盘的RDB传输

commitAsync(font color=\"#4caf50\

 fetchPartitionData.fetchOffset == currentFetchState.fetchOffset

自动提交

ReplicaFetcherThreadmaybeFetch()

client.sendQueue up the given request for sending. Requests can only be sent out to ready nodes

生产者有多个，且对着一个分区有序产生消息由程序控制多实例的执行顺序

RequestHandlerHelper.onLeadershipChange(font color=\"#ff9800\

FileRecords

groupCoordinatorl来处理

sendListOffsetRequest给指定的broker发送ListOffsetRequest请求，获取分区和目标时间戳

ReplicaFetcherManager

onJoinLeader(JoinGroupResponse joinResponse)consumer Leader会进行partition分配，然后向Group coordinator提交分配结果，以便follower同步分区分配的结果.setAssignments(groupAssignmentList)

replicaManager

redis我们要不要配置，或者设计实例大小

mmap.putInt(只写到了内核pagecache

request的处理

LOG的类型是FlieRecords

Leader：respones内容先返回HW：4DATA:  炼狱中delay返回结果后Leader的HW更新为5（根据follower请求过来的LEO修改）Leader@HW：5

KafkaApis.scalaApiKeys.OFFSET_COMMIT => handleOffsetCommitRequest(request)

为了匹配receive中的repsone

producer-02：生产一条数据LastOffset：4requireOffset：5LEO：5

ConsumerCoordinator消费者协调器每个组有一个

// call coordinator to handle commit offset        font color=\"#ff9800\

channel

updateFollowerFetchState

KafkaRequestHandler

// Increment the log end offset. updateLogEndOffset(appendInfo.lastOffset + 1)

Follower请求时间线

Topic：__consumer_offset名下50个分区

kafkaserver

follow2：第三次请求带着LEO：5       HW：3Leader的数据在炼狱中，等待返回

算法Utils.abs(groupId.hashCode) % groupMetadataTopicPartitionCount

SocketServer

selector.poll

commitSync(Duration.ofMillis(defaultApiTimeoutMs));

是在consumer.poll中处理的

RequestFuture<ClientResponse> future = client.font color=\"#f44336\

attemptRead(channel);

responseCallback

if (leaderId.isEmpty)      leaderId = Some(member.memberId)谁先进来，谁就是leader

3-2

Create-Topic

#Returns true if this broker is the current controller def isActive: Boolean = activeControllerId == config.brokerId

// call replica manager to append the group message副本管理器添加组的信息   font color=\"#ff9800\

Paration：Follower发送请求：

log.info(\"Resetting offset for partition {} to position {}.\

3个Broker的__consumer_offsets的分区及leader，和副本分布的详情

readFromLog

GroupCoordinatorgroup-3的元数据

系统IO

Node node = this.client.leastLoadedNode();与某个broker建立socket连接

CompletedFetch records = completedFetches.peek();

Broker  server

P0

3-1

if (!coordinator.font color=\"#ff9800\

responseCallback(produceResponseStatus)

同步提交按poll的批次提交offset

respone处理

groupManager.font color=\"#4caf50\

int numReadyKeys = select(timeout);

1-2

3，事件poll

无

Old 知识：1，输入 read，常注册2，输出 write，有数据再注册

Handler

FetchPosition position = subscriptions.position(completedFetch.partition);

真正的发送

KafkaApis

loadGroupsAndOffsets

consumer3

acquireAndEnsureOpen();只能在一个线程poll

groupCoordinator = font color=\"#4caf50\

问题： 在分布式的情况，消费者与broker如何治理/协调的？1. 需要有Coordinator协调器2. 如何在Broker上分配GroupCoordinator3. Consumer端每个组需要有leader来分配消费那个分区4. 分配策略5. offset：（公共的topic：_consumer_offset含有50个分区，所有的consumer消费的offset都存在这50个分区中）6. 消费：同步、异步、性能优化（IOThreads-workerThreads）---------------------------------------------------------------Kafka Consumer主要做的事情：1. 连接外界2. 拉取数据3. 持久化1和3步都是需要Consumer Coordinator来处理。

follower1

创建super(ApiKeys.SYNC_GROUP);同步组的builder，去请求Group coordinator

Ack：-1

ControllerContext

new NetworkClient

Consumerpoll方法

onJoinFollower()

new font color=\"#4caf50\

世界上为什么有这么多的中间件一个中间件搞定所有的事情NOkafka，mq，对数据不加工redis，kv，集合操作，如果多线程加锁排斥

Group -2

KafkaServer.startup()

!coordinator.font color=\"#f44336\

AdminUtils.assignReplicasToBrokers

GroupCoordinatorgroup-2的元数据

ApiKeys.Fetch

resetOffsetsAsync(offsetResetTimestamps);

for循环处理fetchRequestMap，发送Api.Fetch的request，接收response成功后写入completedFetches队列

KafkaServerstartup()

Group-1

收发数据

commitSync(font color=\"#ff9800\

流程与同步提交类似

segment段默认1G

分治

发送数据

2-3

makeFollowersreplicaFetcherManager.addFetcherForPartitions(partitionsToMakeFollowerWithLeaderAndOffset)

this.transportLayer.addInterestOps(SelectionKey.OP_WRITE);只有在真正发送的时候，才会触发写事件。公共的写法！！！！！！

主从，主挂了有/没有持久化必然触发全同步

validatePositionsOnMetadataChange();

appendToLocalLog

ACK-1

2-2