2.1.1 同步和异步发送消息

缺乏、安全感 2022-11-13 01:52 191阅读 0赞

> 消息系统通常由生产者(producer)、 消费者(consumer)和消息代理(broker)三大部分组成，生产者会将消息写入消息代理，消费者会从消息代理中读取消息。对于消息代理而言，生产者和消费者都属于客户端：`生产者和消费者会发送客户端请求给服务端，服务端的处理分别是存储消息和获取消息，最后服务端返回响应结果给客户端`。

客户端和服务端的通信涉及网络中不同的节点，`客户端和服务端都会有一个连接对象，负责数据的发送和接收`：比如客户端会发送请求、接收响应，服务端会接收请求、发送响应。生产者和消费者客户端与服务端完成一次网络通信的具体步骤如下 。

1.  生产者客户端应用程序产生消息 。
2.  客户端连接对象将消息包装到请求中，发送到服务端 。
3.  服务端连接对象负责接收请求，并将消息以文件形式存储 。
4.  服务端返回响应结果给生产者客户端 。
5.  消费者客户端应用程序消费消息 。
6.  客户端连接对象将消费信息也包装到请求中，发送给服务端 。
7.  服务端从文件存储系统中取出消息 。
8.  服务端返回响应结果给消费者客户端 。
9.  客户端将响应结果还原成消息，并开始处理消息 。

> 本章主要关注客户端和服务端的网络通信流程，没有涉及服务端的具体实现。分布式系统通常会自己实现一套负责不同节点之间数据传输的网络层通信机制，也叫作RPC框架。 RPC框架会处理网络通信协议的编解码、客户端和服务端的请求发送和接收等。

协议是由服务端定制的，客户端只要遵循这种协议发送请求，服务端就能够正常地接收并处理客户端请求 。 Kafka初期使用 Scala编写，早期 Scala版本的生产者、消费者和服务端的实现都放在core包下；`而最新的客户端使用了Java重新实现`，放在clients包下 。 本章主要分析新版本的生产者客户端实现，以及服务端的网络连接实现。

--------------------

### 2.1 生产者客户端 ###

> 生产者应用程序使用 KafkaProducer对象代表一个生产者客户端进程。 生产者要发送消息，并不是直接发送给服务端，而是先`在客户端把消息放入队列中，然后由一个消息发送线程从队列中拉取消息`，以批量的方式发送消息给服务端。Kafka的记录收集器(RecordAccumulator)负责缓存生产者客户端产生的消息，发送线程(Sender)负责读取记录收集器的批量消息 ，通过网络发送给服务端。为了保证客户端网络请求的快速响应，Kafka使用选择器(Selector)处理网络连接和读写处理，使用网络连接(NetworkClient)处理客户端网络请求 。

--------------------

#### 2.1.1 同步和异步发送消息 ####

下面的生产者示例代码来向 Kafka源码根目录的examples包 ：

package kafka.examples;
    …………
    public class Producer extends Thread { 
      private final KafkaProducer<Integer, String> producer;
      private final String topic;
      private final Boolean isAsync;
      public Producer(String topic, Boolean isAsync) { 
        Properties props = new Properties();
        props.put("bootstrap.servers",
            KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT);
        props.put("client.id", "DemoProducer");
        props.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producer = new KafkaProducer<>(props);
        this.topic = topic;
        this.isAsync = isAsync;
      }
    
      @Override
      public void run() { 
        int messageNo = 1;
        while (true) { 
          String messageStr = "Message_" + messageNo;
          long startTime = System.currentTimeMillis();
          if (isAsync) { 
            // Send asynchronously。异步发送
            DemoCallBack callBack = new DemoCallBack(startTime, messageNo, messageStr);
            producer.send(new ProducerRecord<>(topic, messageNo, messageStr), callBack);
          } else { 
            // Send synchronously。同步发送
            try { 
              producer.send(new ProducerRecord<>(topic, messageNo, messageStr)).get();
              System.out.println("Sent message: (" + messageNo + ", " + messageStr + ")");
            } catch (InterruptedException | ExecutionException e) { 
              e.printStackTrace();
            }
          }
          ++messageNo;
        }
      }
    }
    
    class DemoCallBack implements Callback { 
    
      private final long startTime;
      private final int key;
      private final String message;
    
      public DemoCallBack(long startTime, int key, String message) { 
        this.startTime = startTime;
        this.key = key;
        this.message = message;
      }
    
      
      @Override
      public void onCompletion(RecordMetadata metadata, Exception exception) { 
        long elapsedTime = System.currentTimeMillis() - startTime;
        if (metadata != null) { 
          System.out.println(
              "message(" + key + ", " + message + ") sent to partition(" + metadata.partition() +
                  "), " +
                  "offset(" + metadata.offset() + ") in " + elapsedTime + " ms");
        } else { 
          exception.printStackTrace();
        }
      }
    }

其中Producer类首先根据配置文件创建了一个生产者客户端对象 KafkaProducer。在构造函数中，isAsync参数表示生产者发送消息的模式，值为 true表示异步，值为 false表示同步。每一条消息都会包装成一条生产者记录(ProducerRecord)，并传给生产者客户端对象的send发送方法，表示生产者客户端产生的一条消息通过生产者客户端对象发送给Kafka服务端集群。

异步发送模式还传递了一个Callback匿名回调类，当客户端发送完一条消息，并且这条消息成功地存储至Kafka集群后， 服务端会调用匿名回调类的回调方法(onCompletion)。异步发送指的是生产者发送完一条消息后，不需要关心服务端处理完了没有，可以接着发送下一条消息。服务端在处理完每一条消息后，会自动触发回调函数，返回响应结果给客户端。如果是以同步模式发送消息，生产者客户端应用程序不需要提供回调函数。 同步模式下，生产者发送完一条消息后，必须等待服务端返回响应结果，然后才能发送下一条消息。

KafkaProducer只用了一个send方法，就可以完成同步和异步两种模式的消息发迭，这是因为 send方法返回的是一个Future。基于Future，我们可以实现同步或异步的消息发送语义。

*  同步。`调用 send返回Future时，需要立即调用get，因为Future.get在没有返回结果时会一直阻塞`。
 *  异步。提供一个回调，调用 send后可以继续发送消息而不用等待。 当有结果运回时，会向动执行回调函数。

生产者客户端对象KafkaProducer的send方法的处理逻辑是 ： 首先序列化消息的key和value(消息必须序列化成二进制流的形式才能在网络中传输)，然后为每一条消息息选择对应的分区（表示要将消息存储至Kafka集群的哪个节点上），最后通知发送线程发送消息 。 相关代码如下：

@Override
        public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) { 
            // intercept the record, which can be potentially modified; this method does not throw exceptions
            ProducerRecord<K, V> interceptedRecord = this.interceptors == null ? record : this.interceptors.onSend(record);
            return doSend(interceptedRecord, callback);
        }
    
        /** * Implementation of asynchronously send a record to a topic. */
        private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) { 
            TopicPartition tp = null;
            try { 
                // first make sure the metadata for the topic is available
                ClusterAndWaitTime clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs);
                long remainingWaitMs = Math.max(0, maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);
                Cluster cluster = clusterAndWaitTime.cluster;
                byte[] serializedKey;
                try { 
                    serializedKey = keySerializer.serialize(record.topic(), record.headers(), 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");
                }
                byte[] serializedValue;
                try { 
                    serializedValue = valueSerializer.serialize(record.topic(), record.headers(), 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");
                }
                int partition = partition(record, serializedKey, serializedValue, cluster);
                tp = new TopicPartition(record.topic(), partition);
    
                setReadOnly(record.headers());
                Header[] headers = record.headers().toArray();
    
                int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(apiVersions.maxUsableProduceMagic(),
                        compressionType, serializedKey, serializedValue, headers);
                ensureValidRecordSize(serializedSize);
                long timestamp = record.timestamp() == null ? time.milliseconds() : record.timestamp();
                log.trace("Sending record {} with callback {} to topic {} partition {}", record, callback, record.topic(), partition);
                // producer callback will make sure to call both 'callback' and interceptor callback
                Callback interceptCallback = this.interceptors == null ? callback : new InterceptorCallback<>(callback, this.interceptors, tp);
    
                if (transactionManager != null && transactionManager.isTransactional())
                    transactionManager.maybeAddPartitionToTransaction(tp);
    
                RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey,
                        serializedValue, headers, interceptCallback, remainingWaitMs);
                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);
                    this.sender.wakeup();
                }
                return result.future;
                // handling exceptions and record the errors;
                // for API exceptions return them in the future,
                // for other exceptions throw directly
            } catch (ApiException e) { 
                log.debug("Exception occurred during message send:", e);
                if (callback != null)
                    callback.onCompletion(null, e);
                this.errors.record();
                if (this.interceptors != null)
                    this.interceptors.onSendError(record, tp, e);
                return new FutureFailure(e);
            } catch (InterruptedException e) { 
                this.errors.record();
                if (this.interceptors != null)
                    this.interceptors.onSendError(record, tp, e);
                throw new InterruptException(e);
            } catch (BufferExhaustedException e) { 
                this.errors.record();
                this.metrics.sensor("buffer-exhausted-records").record();
                if (this.interceptors != null)
                    this.interceptors.onSendError(record, tp, e);
                throw e;
            } catch (KafkaException e) { 
                this.errors.record();
                if (this.interceptors != null)
                    this.interceptors.onSendError(record, tp, e);
                throw e;
            } catch (Exception e) { 
                // we notify interceptor about all exceptions, since onSend is called before anything else in this method
                if (this.interceptors != null)
                    this.interceptors.onSendError(record, tp, e);
                throw e;
            }
        }

生产者客户端发送消息给Kafka服务端集群，应该能够同时往多个服务端节点写数据，从而最大地提高生产者客户端的性能。 Kafka的一个主题会有多个分区，分区作为并行任务的最小单位，为消息选择分区要根据消息是存含有键来判断。

通常，消息是没有key这个概念的。 比如要发送一条消息，一般直接给出这条消息的内容。 由于消息负载要分布到不同的服务端节点，不能让消息老是发送到同一个节点，以免那个节点负担太重（读取的时候也会有问题）， 一般这种没有键的消息会采用round-robin方式，均衡地分发到不同的分区。

如果指定了消息的键，为消息选择分区的算法是：对键进行散列化后，再与分区的数量－取模运算得到分区编号 。 因为对一个不变的键进行散列化的结果永远是同一个值，所以只要分区数不变，相同键的所有消息总是会被写到同一个分区中。

**注意**：包含键的消息通常是要取出来做进一步处理的。 相同键的消息有多条数据，并且都保存在一个分区内。 如果要读取指定键的消息，也只需要和一个分区联系，就可以取出这个键的所有数据。

1.  为消息选择分区

如果没有提前创建消息所属的主题，默认情况下主题的分区数量只有一个 。 一个主题只有一个分区时，会导致同一个主题的所有消息都只会保存到一个节点上。 一般我们会提前创建主题，指定更多的分区数，这样同一个主题的所有消息就会分散在不同的节点上 。

Partitioninfo对象表示一个分区的分布信息、它的成员变量有主题名称、分区编号、所在的主副本节点、所有的副本、 ISR列表。把消息和 Partitioninfo组合起来，就能表示消息被发送到哪个主题的哪个分区 。 相关代码如下：

public class PartitionInfo { 
    
        private final String topic;
        private final int partition;
        private final Node leader;
        private final Node[] replicas;
        private final Node[] inSyncReplicas;
    …………
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

> 以1.4.2节的my-replicated-topic3主题(5个分区、3个副本)为例，假设我们在本机启动了4个代理节点。如图2-1（左）所示，分区信息、PartitionInfo记录了每个分区的基本信息。图2-1（右）列举了每个代理节点上的分区数量、主副本的数量和分区集。  
> ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]

Kafka通过将主题分成多个分区的语义来实现并行处理，生产者可以将一批消息分成多个分区，每个分区写入不同的服务端节点 。 如图2-2所示，消息集的每条消息都会选择一个分区编号，不同的分区可以同时向分区的主副本节点发送生产请求 。 生产者客户端采用这种分区并行发送的方式，从而提升生产者客户端的写人性能。 分区对消费者也有好处 ， 消费者指定获取一个主题的消息，它也可以同时从多个分区读取消息，从而提升消费者客户端的读取性能。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]  
partition()方法会为消息、选择一个分区编号。为了保证消息负载均衡地分布到各个服务端节点，对于没有键的消息，通过计数器自增轮询的方式依次将消息分配到不同的分区上；对于有键的消息，对键计算散列值，然后和主题的分区数进行取模得到分区编号。 相关代码如下：

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { 
            List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
            int numPartitions = partitions.size();
            if (keyBytes == null) { 
                int nextValue = nextValue(topic);
                List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
                if (availablePartitions.size() > 0) { 
                    int part = Utils.toPositive(nextValue) % availablePartitions.size();
                    return availablePartitions.get(part).partition();
                } else { 
                    // no partitions are available, give a non-available partition
                    return Utils.toPositive(nextValue) % numPartitions;
                }
            } else { 
                // hash the keyBytes to choose a partition
                return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
            }
        }

如图 2-3所示，假设Topic主题有4个分区， 则总共有4个对应的分区信息对象。一个分区有多个副本，灰色矩形表示主副本 ， 白色矩形表示备份副本，备份副本的数据会和主副本保持同步。假设分区P4 （图中Topic 1-Partition4的简写，下同 ）还没有选举出主副本，那么没有键的消息不会被分配到分区P4 。 选择分区时，计数器是递增的，第一条消息写到分区Pl ，第二条消息写到分区P2 ，第三条写到分区P3 ，第四条又会写到分区P4 ， 以此类推 。 只有分区P4可用了，新的消息才会写到分区P4上 。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]

在客户端就为消息选择分区的目的是什么？只有为消息选择分区，我们才能知道应该发送到哪个节点，否则如图2-4所示，只能随便找一个服务端节点，再由那个节点去决定如何将消息、转发给其他正确的节点来保存。 后面这种方式增加了服务端的负担，多了不必要的数据传输。 可以看到，这种方式比在客户端选择分区多了一次消息传输，而且是全量的数据传输，显然不划算。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]  
消息经过序列化，并且要存储的分区编号也己选择，下一步要将消息先缓存在客户端的记录收集器里 。

1.  客户端记录收集器

生产者发迭的消息先在客户端缓存到记录收集器RecordAccumulator中，等到一定时机再由发送线程Sender批量地写入Kafka集群 。生产者每生产一条消息，就向记录收集器中追加一条消息，追加方法的返回值表示批记录（ RecordBatch ）是否满了 ： 如果批记录满了， 则开始发送这一批数据 。 如图 2-5所示，`每个分区都有一个双端队列用来缓存客户端的消息，队列的每个元素是一个批记录` 。 一旦分区的队列中有批记录满了，就会被发送线程发送到分区对应的节点 ； 如果批记录没有满，就会继续等待直到收集到足够的消息 。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]

追加消息时首先要获取分区所属的队列，然后取队列中最后一个批记录，如果队列中不存在批记录或者上一个批记录已经写满，应该创建新的批记录，并且加入队列的尾部。 这里我们把每个批记录看作队列的一个元素，先创建的批记录最先被旧的消息填满，后创建的批记录表示最近的消息，追加消息时总是往最近的批记录中添加 。如图2-6所示，往记录收集器中追加一条记录有多个分支条件和判断。

(1 ）队列中不存在批记录，进入步骤（ 5 ） 。  
(2）如果存在旧的批记录，尝试追加当前一条消息，并判断能不能追加成功 。  
(3）如果追加成功，说明已有的批记录可以容纳当前这条消息，返回结果 。  
(4）如果追加不成功，说明虽然有旧的批记录，但是容纳不下当前这一条消息，进入下一步 。  
(5 ）创建一个新的批记录，并往其中添加当前消息，新的批记录一定能容纳当前这条消息 。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 6]  
记录收集器的作用是缓存客户端的消息，还需要通过消息发送线程才能将消息发送到服务端。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/a728cdc23545491eaf5def66f0af65c2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/b9a16268f56b411eb3cbedaa8d2516db.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/c88778ec38964fcd9afcd2024be4e59f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/0adad2d8a6094ec7956c753b8fc2f27b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]: /images/20221022/fe95b7faa496449ab6df70bcce95a7d5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]: /images/20221022/aedb892f0f92466e85816e3b6652b417.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 6]: /images/20221022/99d5b8dcfba043c2b6b2e01301317f2d.png