试着理解Kafka的scala语法

8dtrkrch 于 2021-06-05 发布在 Kafka

关注(0)|答案(3)|浏览(315)

我一直在看Kafka的源代码 Log 在Kafka项目的核心模块类，但我仍然是新的scala。我遇到了一个很难理解的语法。以下是代码片段：
代码段1：

// Now do a second pass and load all the log and index files.
    // We might encounter legacy log segments with offset overflow (KAFKA-6264). We need to split such segments. When
    // this happens, restart loading segment files from scratch.
    retryOnOffsetOverflow {
      // In case we encounter a segment with offset overflow, the retry logic will split it after which we need to retry
      // loading of segments. In that case, we also need to close all segments that could have been left open in previous
      // call to loadSegmentFiles().
      logSegments.foreach(_.close())
      segments.clear()
      loadSegmentFiles()
    }

代码段2：

private[log] def retryOnOffsetOverflow[T](fn: => T): T = {
    while (true) {
      try {
        return fn// what is fn here in context of code snippet 1?
      } catch {
        case e: LogSegmentOffsetOverflowException =>
          info(s"Caught segment overflow error: ${e.getMessage}. Split segment and retry.")
          splitOverflowedSegment(e.segment)//##!!!1.return a List[Segement], but where does this return goes?
      }
    }
    throw new IllegalStateException()
  }

我发现很难理解的是这个方法 retryOnOffsetOverflow 是在代码段1中调用的，传递给它的参数是什么？我知道你的情人 retryOnOffsetOverflow 是一个函数，但在这个代码段中，传递给这个函数的参数是什么？
我也不清楚什么是回报 retryOnOffsetOverflow 在这里？回报是 T 哪种是通用的？我不知道你的回报是什么 retryOnOffsetOverflow 在这里，它是否会根据捕获异常的事实而有所不同？如果是的话，具体回报是多少？
非常感谢您的解释，如果我错过了回答问题所需的代码，请告诉我。
更新：我会纠正我自己的param retryOnOffsetOverflow 是一个按名称参数，除非在方法体的某个地方引用了它，否则不会对其求值。

scala apache-kafka

来源：https://stackoverflow.com/questions/61885638/try-to-understand-kafka-scala-syntax

3条答案

按热度按时间

mtb9vblg1#

def retryOnOffsetOverflow[T](fn: => T): T
``` `retryOnOffsetOverflow()` 是一个采用单个“by name”参数的方法，这意味着除非在方法体的某个地方引用了该参数，否则不会计算该参数。
所以呢 `fn` 可能是单个值，也可能是多行代码，在这里之前它将保持未计算（未执行）状态 `try { return fn }` ，它被 Package 在 `try` .
这个 `fn` 评估将产生某种类型的值。这种类型其实并不重要（我们称之为 `T` )但是 `retryOnOffsetOverflow()` 必须返回相同的类型。

赞(0）回复(0）举报 2021-06-06

hm2xizp92#

我发现很难理解的是，在代码段1中如何调用retryonoffsetoverflow方法，以及将什么作为其param的参数传递给它？我知道retryonoffsetoverflow的param是一个函数，但是在这个片段中，传递给这个函数的参数是什么？
考虑下面的简化示例

def printTillSuccess[T](fn: => T): T = {
  while (true) {
    val result = fn
    println(result)
    return result
  }
  throw new Exception("dead end")
}

printTillSuccess("a")
printTillSuccess({ "a" })
printTillSuccess { "a" }
``` `fn: => T` 不是函数，而是按名称参数。它将在每一个参考上进行评估 `val result = fn` . scala中的函数具有 `apply` 方法，这里不是这样。
可以通过将值传递到方法中 `()` ，这是在示例中完成的 `printTillSuccess("a")` . 
scala允许用 `{}` 最后一条语句将用作块的结果。因此， `{"a"}` 与相同 `"a"` .
所以，现在你可以通过了 `{"a"}` 变成方法，因此 `printTillSuccess({ "a" })` 是有效呼叫。
最后，scala允许替换 `()` 使用块定义 `{}` 在方法中，这将打开语法 `printTillSuccess { "a" }` .
我也不清楚这里的retryonoffsetoverflow的返回是什么？返回值是t，这是一种泛型？我不确定retryonoffsetoverflow在这里的返回是什么，它是否会根据捕获异常的事实而有所不同？如果是的话，具体回报是多少？
返回类型是by name参数的类型，它是 `T` . 和声明 `return fn` 是唯一能定义 `T` . 
万一 `LogSegmentOffsetOverflowException` ，将调用catch并 `splitOverflowedSegment` 执行。这将改变一些内部状态和下一次迭代 `while(true)` 将再次计算by name参数。因此，异常不会更改返回类型，但允许进行下一次迭代。
这个 `while` 循环只能在 `fn` 计算成功或引发其他异常。在任何情况下，回报，如果发生，将是重要的 `T` .

赞(0）回复(0）举报 2021-06-05

qij5mzcb3#

upd.：略微更改了最后一部分，因为看起来它将在下一个循环迭代中加载“拆分”的文件。
参数 retryOnOffsetOverflow 下面是花括号中的所有内容-基本上，这三行代码-这就是函数 (fn: => T) ，它接受。 retryOnOffsetOverflow 正在尝试执行传递的函数，如果执行成功，则返回其答案。有一部分有点难以理解-当有例外的时候， splitOverflowedSegment 不是因为它的返回类型，而是因为它在 replaceSegments 功能。在下一次循环迭代重新启动时，将读取这些段 loadSegmentFiles 功能。

赞(0）回复(0）举报 2021-06-05

我来回答

试着理解Kafka的scala语法

3条答案

相关问题

热门标签

最新问答