goka EnsureStreamExists 作用

`goka.EnsureStreamExists` 是 Goka 框架的一个函数，用于确保指定的 Kafka 流存在。在 Goka 应用程序中，输入流和输出流通常需要在应用程序启动前就已经存在，否则应用程序无法正确地读取和发送消息。`goka.EnsureStreamExists` 函数可以帮助应用程序开发者在应用程序启动时自动创建输入流和输出流，从而简化应用程序的部署和维护。

具体来说，`goka.EnsureStreamExists` 函数需要传入一个 `context.Context` 类型的参数，用于指定函数的上下文信息，以及一个 `string` 类型的参数，用于指定要创建的流的名称。如果指定的流不存在，`goka.EnsureStreamExists` 函数会自动使用 Kafka 的 Admin API 创建一个具有默认配置的流，否则函数会直接返回而不进行任何操作。

举个例子，如果你的 Goka 应用程序需要从一个名为 `input` 的流中读取消息，并将处理结果发送到一个名为 `output` 的流中，你可以使用如下代码确保这两个流已经存在：

func main() {
    // 创建 Kafka 连接器
    brokers := []string{"localhost:9092"}
    conn, _ := kafka.Dial(brokers, kafka.NewBrokerConf())

    // 确保流存在
    goka.EnsureStreamExists(context.Background(), conn, "input")
    goka.EnsureStreamExists(context.Background(), conn, "output")

    // 创建 Goka 应用程序
    g := goka.DefineGroup(
        "my-group",
        goka.Input("input", new(codec.String), partitioner),
        goka.Persist(new(codec.String), handler),
        goka.Output("output", new(codec.String)),
    )
    goka.RunProcessor(context.Background(), brokers, g)
}

在上面的代码中，`goka.EnsureStreamExists` 函数会在应用程序启动时自动创建 `input` 和 `output` 两个流，如果这两个流已经存在，则函数不会进行任何操作。这样，应用程序就可以在启动时自动创建所需的流，而不需要手动创建流和配置 Kafka 环境。