博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
客户端发送消息时,源码运行的大致流程
阅读量:311 次
发布时间:2019-03-03

本文共 3202 字,大约阅读时间需要 10 分钟。

send消息:

@Override    public Future
send(ProducerRecord
record, Callback callback) {
// 如果拦截器为null,就直接拿封装的ProducerRecord //否则就动用拦截器,对发送的消息进行拦截 ProducerRecord
interceptedRecord = this.interceptors == null ? record : this.interceptors.onSend(record); return doSend(interceptedRecord, callback); }

如果我们自定义了拦截器,就调用我们实现的onSend方法,走我们的逻辑。

接下来要调用同步阻塞方法waitOnMetadata()方法,去等待获取到目标Topic对应的元数据。如果客户端尚未缓存到元数据,就一定要发送网络请求到Broker,去拉取那个Topic的元数据回来。

//等待获取元数据long waitedOnMetadataMs = waitOnMetadata(record.topic(), this.maxBlockTimeMs);//剩余可阻塞的时间:如果等待获取元数据时,已经阻塞了一段时间了。那么接下来的阻塞,时间就会变短long remainingWaitMs = Math.max(0, this.maxBlockTimeMs - waitedOnMetadataMs);

要想往一个Topic里发消息,必须知道这个Topic的元数据:这个Topic有哪些分区、根据Partitioner组件选择一个分区、这个分区对应的Leader在哪个Broker上,然后才能和Broker建立连接、发送消息。

这一次获取到元数据,再以后就可以直接根据缓存来发送了。

序列化key和value:

//将key序列化成字节数组            byte[] serializedKey;            try {
serializedKey = keySerializer.serialize(record.topic(), record.key()); } catch (ClassCastException cce) {
throw new SerializationException("Can't convert key of class " + record.key().getClass().getName() + " to class " + producerConfig.getClass(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG).getName() + " specified in key.serializer"); } //将value序列化成字节数组 byte[] serializedValue; try {
serializedValue = valueSerializer.serialize(record.topic(), record.value()); } catch (ClassCastException cce) {
throw new SerializationException("Can't convert value of class " + record.value().getClass().getName() + " to class " + producerConfig.getClass(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG).getName() + " specified in value.serializer"); }

key、value可以是各种类型的:String、Double、Boolean…或者是自定义的对象。但是如果想发送消息给Broker,就必须将key、value都序列化成字节数组

然后基于获取到的Topic元数据,根据Partitioner组件获取消息对应的分区

//使用Partitioner组件获取消息对应的分区int partition = partition(record, serializedKey, serializedValue, metadata.fetch());

然后对我们要发送的这条消息进行安全检查,是否超出了请求的最大值、内存缓冲的最大值:

//检查要发送的这条消息,是否超出了请求的最大大小、内存缓冲的最大大小ensureValidRecordSize(serializedSize);

根据拦截器来确定Callback:

//如果我们自定义了拦截器,就为拦截器创建Callback函数Callback interceptCallback = this.interceptors == null ? callback : new InterceptorCallback<>(callback, this.interceptors, tp);

这个callback就是我们在调用producer.send()方法时,传进来的:

producer.send(record,new Callback(){......})

在消息发送完成之后,回调我们自定义实现的Callback函数,通知我们消息发送的结果。

然后进行最关键的一步:将消息添加到内存缓冲里去(RecordAccumulator组件负责的)

//将消息添加到内存缓冲里去,RecordAccumulator组件负责的RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey, serializedValue, interceptCallback, remainingWaitMs);

如果某个分区对应的batch满了、或者有一个新batch被创建出来了,就唤醒Sender线程去工作(发送batch):

//如果batch已经满了,或者有一个新batch被创建了            if (result.batchIsFull || result.newBatchCreated) {
log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), partition); //唤醒Sender线程去工作 this.sender.wakeup(); }

转载地址:http://cdbq.baihongyu.com/

你可能感兴趣的文章