apache-kafka - kafka : what do 'soTimeout' , 'bufferSize' 和 'minBytes' 对 SimpleConsumer 意味着什么？

Question

我正在使用卡夫卡
0.8.2.1 简单消费者。有人可以澄清 SimpleConsumer 和 FetchRequestBuilder 的一些配置参数的含义吗？由于没有阅读 KAfka 的源代码，我当时找不到任何文档。 (我尝试将此问题发布到 kafka 用户组 - 但没有运气):

-- Q1:在 SimpleConsumer 构造函数的签名中，我看到 Int ' soTimeout' 范围 -
这个超时是什么意思？这是连接到 Kafka 代理的超时吗？从任何 [或特定??] 请求到 Kafka(如 FetchRequest)获得响应时超时？还有什么？

kafka.javaapi.consumer.SimpleConsumer
    (val host: String,
     val port: Int,
     val soTimeout: Int,
     val bufferSize: Int,
     val clientId: String)

-- Q2:同样，SimpleConsumer 构造函数采用 Int 'bufferSize' 参数。它的意义是什么？这是在发出 fetchRequest 时 SimpleConsumer 将读取多少字节？或者它是从 Kafka 每次获取读取的最大字节数 - 如果有更多数据可用，则会发生多次获取？

-- 通过 FetchRequestBuilder 构建 FetchRequest 时(见下文)，我还需要指定' fetchSize ':

FetchRequest req= newFetchRequestBuilder ()
  .clientId(kafkaGroupId)
  .addFetch(topic, partition, offset, fetchSizeInBytes)
  .build();

查看 FetchRequestBuilder 的源代码，我认为(我不是 Scala 专家)这些调用
转换为以下方法调用 - 传递到 FetchRequest 的最终参数称为“ ” minBytes '，暗示这可能不是确切的提取大小？ .这是否意味着它甚至不会获取任何东西，除非至少有“minBytes”的数据可用？

class FetchRequestBuilder():
    def addFetch(topic: String, partition: Int, offset: Long, fetchSize: Int)

    def build() = {
      val fetchRequest= FetchRequest(versionId, correlationId.getAndIncrement, clientId, replicaId, maxWait, minBytes, requestMap.toMap)

FetchRequest(versionId: Short = FetchRequest.CurrentVersion,
    correlationId: Int = FetchRequest.DefaultCorrelationId,
    clientId: String = ConsumerConfig.DefaultClientId,
    replicaId: Int = Request.OrdinaryConsumerId,
    maxWait: Int = FetchRequest.DefaultMaxWait,
    **minBytes: Int = FetchRequest.DefaultMinBytes**,
...)

所以，我的最后一个问题是:

-- Q3: 怎么办 ' 缓冲区大小 ' 和 ' fetchSize/minBytes ' 相关？他们究竟定义了什么？我必须做吗
确定一个比另一个更小或更大？

谢谢，

码头

Answer 1

soTimeout 是等待连接到给定代理的时间(以毫秒为单位)。我不知道连接会发生什么特别的事情，除了你得到验证，那里有一个代理准备执行一些后续操作。

我相信构造函数中使用的 bufferSize 是客户端套接字用于接收代理发送的数据的缓冲区的大小。

对于您的最后一个问题，如果 fetch 请求由于某种原因返回的总字节数大于请求的套接字缓冲区大小，则需要多个较低级别的调用来检索所有数据，即使有单个更高级别的 fetch 调用。