Kafka常用术语
Broker：Kafka的服务端即Kafka实例，Kafka集群由一个或多个Broker组成，主要负责接收和处理客户端的请求
Topic：主题，Kafka承载消息的逻辑容器，每条发布到Kafka的消息都有对应的逻辑容器，工作中多用于区分业务
Partition：分区，是物理概念，代表有序不变的消息序列，每个Topic由一个或多个Partion组成
Replica：副本，Kafka中同一条消息拷贝到多个地方做数据冗余，这些地方就是副本，副本分为Leader和Follower，角色不同作用不同，副本是对Partition而言的，每个分区可配置多个副本来实现高可用
Record：消息，Kafka处理的对象
Offset：消息位移，分区中每条消息的位置信息，是单调递增且不变的值
Producer：生产者，向主题发送新消息的应用程序
Consumer：消费者，从主题订阅新消息的应用程序
Consumer Offset：消费者位移，记录消费者的消费进度，每个消费者都有自己的消费者位移
Consumer Group：消费者组，多个消费者组成一个消费者组，同时消费多个分区来实现高可用（组内消费者的个数不能多于分区个数以免浪费资源）
Reblance：重平衡，消费组内消费者实例数量变更后，其他消费者实例自动重新分配订阅主题分区的过程
下面用一张图展示上面提到的部分概念（用PPT画的图，太费劲了，画了老半天，有好用的画图工具欢迎推荐）
消息生产流程
先来个KafkaProducer的小demo
public static void main(String[] args) throws ExecutionException, InterruptedException { if (args.length != 2) { throw new IllegalArgumentException("usage: com.ding.KafkaProducerDemo bootstrap-servers topic-name"); } Properties props = new Properties(); // kafka服务器ip和端口，多个用逗号分割 props.put("bootstrap.servers", args[0]); // 确认信号配置 // ack=0 代表producer端不需要等待确认信号，可用性最低 // ack=1 等待至少一个leader成功把消息写到log中，不保证follower写入成功，如果leader宕机同时follower没有把数据写入成功 // 消息丢失 // ack=all leader需要等待所有follower成功备份，可用性最高 props.put("ack", "all"); // 重试次数 props.put("retries", 0); // 批处理消息的大小，批处理可以增加吞吐量 props.put("batch.size", 16384); // 延迟发送消息的时间 props.put("linger.ms", 1); // 用来换出数据的内存大小 props.put("buffer.memory", 33554432); // key 序列化方式 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // value 序列化方式 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 创建KafkaProducer对象，创建时会启动Sender线程 Producer<String, String> producer = new KafkaProducer<>(props); for (int i = 0; i < 100; i++) { // 往RecordAccumulator中写消息 Future<RecordMetadata> result = producer.send(new ProducerRecord<>(args[1], Integer.toString(i), Integer.toString(i))); RecordMetadata rm = result.get(); System.out.println("topic: " + rm.topic() + ", partition: " + rm.partition() + ", offset: " + rm.offset()); } producer.close(); }实例化
KafkaProducer构造方法主要是根据配置文件进行一些实例化操作
1.解析clientId，若没有配置则由是producer-递增的数字
2.解析并实例化分区器partitioner，可以实现自己的partitioner，比如根据key分区，可以保证相同key分到同一个分区，对保证顺序很有用。若没有指定分区规则，采用默认的规则（消息有key，对key做hash，然后对可用分区取模；若没有key，用随机数对可用分区取模【没有key的时候说随机数对可用分区取模不准确，counter值初始值是随机的，但后面都是递增的，所以可以算到roundrobin】）
3.解析key、value的序列化方式并实例化
4.解析并实例化拦截器
5.解析并实例化RecordAccumulator，主要用于存放消息（KafkaProducer主线程往RecordAccumulator中写消息，Sender线程从RecordAccumulator中读消息并发送到Kafka中）
6.解析Broker地址
7.创建一个Sender线程并启动
...this.sender = newSender(logContext, kafkaClient, this.metadata);this.ioThread = new KafkaThread(ioThreadName, this.sender, true);this.ioThread.start();...消息发送流程
消息的发送入口是KafkaProducer.send方法，主要过程如下
KafkaProducer.sendKafkaProducer.doSend// 获取集群信息KafkaProducer.waitOnMetadata // key/value序列化key\value serialize// 分区KafkaProducer.partion// 创建TopciPartion对象，记录消息的topic和partion信息TopicPartition// 写入消息RecordAccumulator.applend// 唤醒Sender线程Sender.wakeupRecordAccumulator
RecordAccumulator是消息队列用于缓存消息，根据TopicPartition对消息分组
重点看下RecordAccumulator.applend追加消息的流程
// 记录进行applend的线程数appendsInProgress.incrementAndGet();// 根据TopicPartition获取或新建Deque双端队列Deque<ProducerBatch> dq = getOrCreateDeque(tp);...private Deque<ProducerBatch> getOrCreateDeque(TopicPartition tp) { Deque<ProducerBatch> d = this.batches.get(tp); if (d != null) return d; d = new ArrayDeque<>(); Deque<ProducerBatch> previous = this.batches.putIfAbsent(tp, d); if (previous == null) return d; else return previous;}// 尝试将消息加入到缓冲区中// 加锁保证同一个TopicPartition写入有序synchronized (dq) { if (closed) throw new KafkaException("Producer closed while send in progress"); // 尝试写入 RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq); if (appendResult != null) return appendResult;}private RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers, Callback callback, Deque<ProducerBatch> deque) { // 从双端队列的尾部取出ProducerBatch ProducerBatch last = deque.peekLast(); if (last != null) { // 取到了，尝试添加消息 FutureRecordMetadata future = last.tryAppend(timestamp, key, value, headers, callback, time.milliseconds()); // 空间不够，返回null if (future == null) last.closeForRecordAppends(); else return new RecordAppendResult(future, deque.size() > 1 || last.isFull(), false); } // 取不到返回null return null;}public FutureRecordMetadata tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers, Callback callback, long now) { // 空间不够，返回null if (!recordsBuilder.hasRoomFor(timestamp, key, value, headers)) { return null; } else { // 真正添加消息 Long checksum = this.recordsBuilder.append(timestamp, key, value, headers); ... FutureRecordMetadata future = ... // future和回调callback进行关联 thunks.add(new Thunk(callback, future)); ... return future; }}// 尝试applend失败（返回null），会走到这里。如果tryApplend成功直接返回了// 从BufferPool中申请内存空间，用于创建新的ProducerBatchbuffer = free.allocate(size, maxTimeToBlock);synchronized (dq) { // 注意这里，前面已经尝试添加失败了，且已经分配了内存，为何还要尝试添加？ // 因为可能已经有其他线程创建了ProducerBatch或者之前的ProducerBatch已经被Sender线程释放了一些空间，所以在尝试添加一次。这里如果添加成功，后面会在finally中释放申请的空间 RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq); if (appendResult != null) { return appendResult; } // 尝试添加失败了，新建ProducerBatch MemoryRecordsBuilder recordsBuilder = recordsBuilder(buffer, maxUsableMagic); ProducerBatch batch = new ProducerBatch(tp, recordsBuilder, time.milliseconds()); FutureRecordMetadata future = Utils.notNull(batch.tryAppend(timestamp, key, value, headers, callback, time.milliseconds())); dq.addLast(batch); incomplete.add(batch); // 将buffer置为null,避免在finally汇总释放空间 buffer = null; return new RecordAppendResult(future, dq.size() > 1 || batch.isFull(), true);}finally { |
|
收藏
淘帖
踩
