本文永久链接 – https://tonybai.com/2024/02/18/some-changes-in-go-1-22

美国时间2024年2月6日，正当中国人民洋溢在即将迎来龙年春节的喜庆祥和的气氛中时，Eli Bendersky代表Go团队在Go官博发文“Go 1.22 is released!”，正式向世界宣告了Go 1.22版本的发布！

注：大家可以从Go官网下载Go 1.22的第一个版本go 1.22.0，也可以在Go playground上选择Go 1.22版本在线体验Go 1.22的语法。

记忆中，这似乎是Eli Bendersky首次代表Go团队撰写Go版本发布的文章，文章短小且言简意赅，会让大家误以为Go 1.22版本没有太多的功能点变更，其实不然。读过我之前写的“Go 1.22新特性前瞻”一文的童鞋都知道Go 1.22中有很多重要且影响深远的值得我们关注的变化。在这篇文章中，我们就再来介绍一下这些变化，供大家参考。

0. 插播“旧闻”：Go再次进入Top10，并刷新有史以来的最高排名

TIOBE编程语言排行榜发布2024年2月编程语言排名的时间恰逢中国人民的传统佳节春节期间，因此它的这次排名发布“淹没”在了“龙年大吉”的喜庆气氛当中了。年后开工，大家翻看这条“旧闻”时，才发现在这次排名中，Go再一次回到Top10，位列第8名，刷新了Go打榜一来的历史最好位次。

单看这一次进入top10似乎没有什么，因为2023年4月份，Go也跻身过top10，排名第10。但如果从Go打榜以来的历史曲线来看，如下图：

我们看到了“翘尾”，我们看到了Go迈过“低谷”后的爬升！这与我在《Go语言第一课专栏》的结课语《和你一起迎接Go的黄金十年》中预判：Go即将迎来自己的黄金十年 愈来愈吻合了！

不过，我在《2023年Go语言盘点：稳中求新，稳中求变》一文中提到过TIOBE index作为世界最知名的编程语言排行榜，却存在其“不靠谱”的特性，比如这一期排名中，上古时代的编程语言Fortran从去年同期的第24位上升至第11位，仅比PHP落后一位，另一门古老的COBOL语言也从去年同期的第30位上升至第19位，仅仅比大热的Rust语言落后一位。

因此，对于TIOBE的排名，大家既要了解，也无需过于看重^_^。

言归正传，我们来说说Go 1.22版本的变化。

1. 语言变化

Go 1.22对语言语法做了两处变更，一个是Go 1.21版本中的试验特性loopvar在Go 1.22中转正落地；另一个也和for循环有关，那就是for range新增了对整型表达式的支持。两者相比较，还是第一个变化loopvar带来的影响更大一些。为什么呢？因此这是Go语言发展历史上第一次真正的填语义层面的“坑”，而且修改的是一个在Go源码中最常用的控制结构的执行语义，这很大可能会带来break change。Go101教程的作者老貘将之成为Go历史上最大的向后兼容性破坏版本。

注：Go 1.21版本有一个对panic(nil)的语义修正，但我估计很少会有人写出panic(nil)这样的代码。

这次语义修改用一句话表达就是：将经典三段式for循环语句以及for range语句中的用短声明形式定义的循环变量从整个循环定义和共享一个，变为每个迭代定义一个。

这里借用Go官博文章中那个例子再说明一下这个语义变化：

// go1.22-examples/lang/loopvar/main.go
package main

import (
    "fmt"
    "time"
)

func main() {
    done := make(chan bool)

    values := []string{"a", "b", "c"}
    for _, v := range values {
        go func() {
            time.Sleep(time.Second)
            fmt.Println(v)
            done <- true
        }()
    }

    // wait for all goroutines to complete before exiting
    for _ = range values {
        <-done
    }
}

我们用Go 1.22.0版本之前的版本，比如Go 1.21.0，来运行该示例：

$go run main.go
c
c
c

我们看到：由于v是整个循环中各个迭代共享的一个变量，所以在每个迭代新创建的goroutine中输出的v都是循环结束后v的最终值c。

如果我们用go 1.22.0来运行上述示例，我们将得到：

// 输出的值的顺序与goroutine调度有关
$go run main.go
b
c
a

注：关于Go 1.22版本之前的for range的坑，我的极客时间专栏《Go语言第一课》专栏有图文并茂的原理讲解，欢迎订阅阅读。

那么，loopvar这一语义填“坑”究竟会对你的代码造成怎样的影响呢？在Russ Cox关于loopvar语义变更的设计文档中提到了：只有go.mod中的go version在go 1.22.0及以后的时候才会生效，这是一个渐进式过渡的过程，因此目前无论是开源项目还是商业项目，只要go.mod中的go version还没有更新为大于等于go 1.22.0，那么for循环依然会保留短声明定义的变量的原语义，这样这些项目都不会受到影响。

不过，如果是直接在脚本中通过go run xxx.go形式运行某个go源码的，且当前工作目录以及父目录下没有go.mod文件的，go 1.22.0会采用新的loopvar语义，这点大家要注意了。

此外，当你将go.mod中的go version升级到go 1.22.0或更高版本时，也要注意语义变更可能带来的问题。在升级go version之前，可以用Go 1.22版本之前的go vet对项目源码进行一次静态分析，对于go vet提示：“loop variable v captured by func literal”的地方务必注意逐个确认。

注：Go 1.22版本中的go vet已经移除了在go version >= 1.22.0时，对“loop variable v captured by func literal”情况进行警告的功能。

关于Go 1.22中for range支持后面接整型表达式的“语法糖”新特性以及函数迭代器的实验特性，这里就不细说了，大家可以看看“Go 1.22新特性前瞻”一文中的说明。

2. 编译器、运行时与工具链

在编译器、运行时和工具链这些方面，Go 1.22的正式版本与“Go 1.22新特性前瞻”一文中使用的Go 1.22rc1版本几乎没有差异，这里挑主要内容介绍一下，其他一些内容可以参考前瞻一文。

Go 1.22版本继续在编译上优化PGO(profile-guided optimization)， 基于PGO的构建可以比以前版本实现更高比例的调用去虚拟化(devirtualize)。在Go 1.22中，官⽅给出的PGO带来的性能提升数字是2%~14%，这应该是基于Google内部一些典型的Go程序测算出来的。

注：如果你对PGO优化还不是很了解，可以看看“深入理解Profile Guided Optimization(PGO)”这篇文章。

Go 1.22版本编译器现在可以更多运⽤devirtualize和inline。在Go编译器中，devirtualize是一种编译优化技术，旨在消除“虚函数”调用的开销。“虚函数”是指在面向对象编程中，通过基类指针或引用调用的函数。在Go中所谓虚函数调用指的就是通过接口类型变量进行的方法调用。由于是动态调用，基于接口的方法调用需要在运行时进行查找和分派，这可能导致一定的性能损失。

而Go编译器在进行devirtualize优化时，会尝试根据程序的上下文信息和类型信息，确定方法调用的具体对象实例。如果编译器能够确定调用的具体实例，则会将通过接口的方法调用替换为直接调用具体对象实例的方法，从而消除运行时的开销，使得通过接口类型变量进行方法调用的性能得到优化提升。

Go 1.22版本中的运行时可以使基于类型的垃圾收集的元数据更接近每个堆对象，从而将Go程序的CPU性能（延迟或吞吐量）提高了1-3%。这一变化还支持通过重复数据删除冗余元数据，进而将大多数Go程序的内存开销减少了大约1%。

在工具链方面，有三个主要改变这里提一下：

• go work支持vendor

在Go 1.22版本中，通过go work vendor可以将workspace中的依赖放到vendor⽬录下，同时在构建时，如果workspace下有vendor⽬录，那么默认的构建是go build -mod=vendor，即基于vendor的构建。

• go mod init不再care其他vendor工具的配置文件

go mod init不再尝试将其他vendor工具（例如Gopkg.lock ）的配置文件导入到go module依赖文件(go.mod)中了，也就是说从Go 1.22版本开始，go module出现之前的那些gopath时代的依赖管理工具正式退出并成为历史了。

• 改进go test -cover的输出

对于没有自己的测试文件的包，go test -cover在go 1.22版本之前会输出：

? mymod/mypack [no test files]

但在Go 1.22版本之后，会报告覆盖率为0.0%：

mymod/mypack coverage: 0.0% of statements

3. 标准库

这里列举一下标准库值得关注的重大变化，大家可以与前瞻一文相互参考着阅读。

3.1 math/rand/v2：标准库的第一个v2版本包

Go 1.22中新增了math/rand/v2包，这里之所以将它列为Go 1.22版本标准库的⼀次重要变化，是因为这是标准库第一次为某个包建⽴v2版本，按照Russ Cox的说法，这次math/rand/v2包的创建，算是为标准库中的其他可能的v2包“探探路”，找找落地路径。关于math/rand/v2包相对于原math/rand包的变化有很多，具体可以参考issue 61716中的设计与讨论。

3.2 增强http.ServeMux表达能力

在Go 1.22版本中，http.ServeMux的表达能力得到了大幅提升，从原先只支持静态路由，到新版本中支持如下一些特性：

• “/index.html”路由将匹配任何主机和方法的路径”/index.html”；
• “GET /static/”将匹配路径以”/static/”开头的GET请求；
• “example.com/”可以与任何指向主机为”example.com”的请求匹配；
• “example.com/{$}”会匹配主机为”example.com”、路径为”/”的请求，即”example.com/”；
• “/b/{bucket}/o/{objectname…}”匹配第一段为”b”、第三段为”o”的路径。名称”bucket”表示第二段，”objectname”表示路径的其余部分。

并且新版ServeMux在路由匹配性能方面也不输众多开源http路由框架太多，后续建立Go web或api类新项目时，可以考虑首选新版ServeMux来进行路由匹配了，减少对外部的一个依赖。

关于新版http.ServeMux的具体使用方法，其作者Jonathan Amsterdam（也是log/slog的作者）在官博发表了一篇名为“Routing Enhancements for Go 1.22”的文章，大家可以详细参考。

关于标准库的其他一些变化，大家可以参考前瞻一文以及更详细的Go 1.22的发布说明文档。

4. 小结

综上，Go 1.22版本对语言、编译器、工具链、运行时和标准库都有一定程度的改进和创新，遗留代码通过Go 1.22版本的重新编译便可以得到一定程度的性能上的自然提升，这也体现了Go语言在稳中求新、稳中求变的特点。

不过这里还要提醒各位Go开发者，在升级Go 1.22版本时务必注意潜在的向后兼容性问题，尤其是loopvar语义带来的变化影响。

本文涉及的源码可以在这里下载。

Gopher部落知识星球在2024年将继续致力于打造一个高品质的Go语言学习和交流平台。我们将继续提供优质的Go技术文章首发和阅读体验。同时，我们也会加强代码质量和最佳实践的分享，包括如何编写简洁、可读、可测试的Go代码。此外，我们还会加强星友之间的交流和互动。欢迎大家踊跃提问，分享心得，讨论技术。我会在第一时间进行解答和交流。我衷心希望Gopher部落可以成为大家学习、进步、交流的港湾。让我相聚在Gopher部落，享受coding的快乐! 欢迎大家踊跃加入！

著名云主机服务厂商DigitalOcean发布最新的主机计划，入门级Droplet配置升级为：1 core CPU、1G内存、25G高速SSD，价格5$/月。有使用DigitalOcean需求的朋友，可以打开这个链接地址：https://m.do.co/c/bff6eed92687 开启你的DO主机之路。

Gopher Daily(Gopher每日新闻) – https://gopherdaily.tonybai.com

我的联系方式：

• 微博(暂不可用)：https://weibo.com/bigwhite20xx
• 微博2：https://weibo.com/u/6484441286
• 博客：tonybai.com
• github: https://github.com/bigwhite
• Gopher Daily归档 – https://github.com/bigwhite/gopherdaily

商务合作方式：撰稿、出书、培训、在线课程、合伙创业、咨询、广告合作。

本文永久链接 – https://tonybai.com/2024/01/08/go-unit-testing-deps-on-kafka

Kafka是Apache基金会开源的一个分布式事件流处理平台，是Java阵营(最初为Scala)中的一款杀手级应用，其提供的高可靠性、高吞吐量和低延迟的数据传输能力，让其到目前为止依旧是现代企业级应用系统以及云原生应用系统中使用的重要中间件。

在日常开发Go程序时，我们经常会遇到一些依赖Kafka的代码，如何对这些代码进行测试，尤其是单测是摆在Go开发者前面的一个现实问题！

有人说用mock，是个路子。但看过我的《单测时尽量用fake object》一文的童鞋估计已经走在了寻找kafka fake object的路上了！Kafka虽好，但身形硕大，不那么灵巧。找到一个合适的fake object不容易。在这篇文章中，我们就来聊聊如何测试那些依赖kafka的代码，再往本质一点说，就是和大家以找找那些合适的kafka fake object。

1. 寻找fake object的策略

在《单测时尽量用fake object》一文中，我们提到过，如果测试的依赖提供了tiny版本或某些简化版，我们可以直接使用这些版本作为fake object的候选，就像etcd提供了用于测试的自身简化版的实现(embed)那样。

但Kafka并没有提供tiny版本，我们也只能选择《单测时尽量用fake object》一文提到的另外一个策略，那就是利用容器来充当fake object，这是目前能搞到任意依赖的fake object的最简单路径了。也许以后WASI(WebAssembly System Interface)成熟了，让wasm脱离浏览器并可以在本地系统上飞起，到时候换用wasm也不迟。

下面我们就按照使用容器的策略来找一找适合的kafka container。

2. testcontainers-go

我们第一站就来到了testcontainers-go。testcontainers-go是一个Go语言开源项目，专门用于简化创建和清理基于容器的依赖项，常用于Go项目的单元测试、自动化集成或冒烟测试中。通过testcontainers-go提供的易于使用的API，开发人员能够以编程方式定义作为测试的一部分而运行的容器，并在测试完成时清理这些资源。

注：testcontainers不仅提供Go API，它还覆盖了主流的编程语言，包括：Java、.NET、Python、Node.js、Rust等。

在几个月之前，testcontainers-go项目还没有提供对Kafka的直接支持，我们需要自己使用testcontainers.GenericContainer来自定义并启动kafka容器。2023年9月，以KRaft模式运行的Kafka容器才被首次引入testcontainers-go项目。

目前testcontainers-go使用的kafka镜像版本是confluentinc/confluent-local:7.5.0。Confluent是在kafka背后的那家公司，基于kafka提供商业化支持。今年初，Confluent还收购了Immerok，将apache的另外一个明星项目Flink招致麾下。

confluent-local并不是一个流行的kafka镜像，它只是一个使用KRaft模式的零配置的、包含Confluent Community RestProxy的Apache Kafka，并且镜像是实验性的，仅应用于本地开发工作流，不应该用在支持生产工作负载。

生产中最常用的开源kafka镜像是confluentinc/cp-kafka镜像，它是基于开源Kafka项目构建的，但在此基础上添加了一些额外的功能和工具，以提供更丰富的功能和更易于部署和管理的体验。cp-kafka镜像的版本号并非kafka的版本号，其对应关系需要cp-kafka镜像官网查询。

另外一个开发领域常用的kafka镜像是bitnami的kafka镜像。Bitnami是一个提供各种开源软件的预打包镜像和应用程序栈的公司。Bitnami Kafka镜像是基于开源Kafka项目构建的，是一个可用于快速部署和运行Kafka的Docker镜像。Bitnami Kafka镜像与其内部的Kakfa的版本号保持一致。

下面我们就来看看如何使用testcontainers-go的kafka来作为依赖kafka的Go单元测试用例的fake object。

这第一个测试示例改编自testcontainers-go/kafka module的example_test.go：

// testcontainers/kafka_setup/kafka_test.go

package main

import (
    "context"
    "fmt"
    "testing"

    "github.com/testcontainers/testcontainers-go/modules/kafka"
)

func TestKafkaSetup(t *testing.T) {
    ctx := context.Background()

    kafkaContainer, err := kafka.RunContainer(ctx, kafka.WithClusterID("test-cluster"))
    if err != nil {
        panic(err)
    }

    // Clean up the container
    defer func() {
        if err := kafkaContainer.Terminate(ctx); err != nil {
            panic(err)
        }
    }()

    state, err := kafkaContainer.State(ctx)
    if err != nil {
        panic(err)
    }

    if kafkaContainer.ClusterID != "test-cluster" {
        t.Errorf("want test-cluster, actual %s", kafkaContainer.ClusterID)
    }
    if state.Running != true {
        t.Errorf("want true, actual %t", state.Running)
    }
    brokers, _ := kafkaContainer.Brokers(ctx)
    fmt.Printf("%q\n", brokers)
}

在这个例子中，我们直接调用kafka.RunContainer创建了一个名为test-cluster的kafka实例，如果没有通过WithImage向RunContainer传入自定义镜像，那么默认我们将启动一个confluentinc/confluent-local:7.5.0的容器（注意：随着时间变化，该默认容器镜像的版本也会随之改变）。

通过RunContainer返回的kafka.KafkaContainer我们可以获取到关于kafka容器的各种信息，比如上述代码中的ClusterID、kafka Broker地址信息等。有了这些信息，我们后续便可以与以容器形式启动的kafka建立连接并做数据的写入和读取操作了。

我们先来看这个测试的运行结果，与预期一致：

$ go test
2023/12/16 21:45:52 github.com/testcontainers/testcontainers-go - Connected to docker:
  ... ...
  Resolved Docker Host: unix:///var/run/docker.sock
  Resolved Docker Socket Path: /var/run/docker.sock
  Test SessionID: 19e47867b733f4da4f430d78961771ae3a1cc66c5deca083b4f6359c6d4b2468
  Test ProcessID: 41b9ef62-2617-4189-b23a-1bfa4c06dfec
2023/12/16 21:45:52 Creating container for image docker.io/testcontainers/ryuk:0.5.1
2023/12/16 21:45:53 Container created: 8f2240042c27
2023/12/16 21:45:53 Starting container: 8f2240042c27
2023/12/16 21:45:53 Container started: 8f2240042c27
2023/12/16 21:45:53 Waiting for container id 8f2240042c27 image: docker.io/testcontainers/ryuk:0.5.1. Waiting for: &{Port:8080/tcp timeout:<nil> PollInterval:100ms}
2023/12/16 21:45:53 Creating container for image confluentinc/confluent-local:7.5.0
2023/12/16 21:45:53 Container created: a39a495aed0b
2023/12/16 21:45:53 Starting container: a39a495aed0b
2023/12/16 21:45:53 Container started: a39a495aed0b
["localhost:1037"]
2023/12/16 21:45:58 Terminating container: a39a495aed0b
2023/12/16 21:45:58 Container terminated: a39a495aed0b
PASS
ok      demo    6.236s

接下来，在上面用例的基础上，我们再来做一个Kafka连接以及数据读写测试：

// testcontainers/kafka_consumer_and_producer/kafka_test.go

package main

import (
    "bytes"
    "context"
    "errors"
    "net"
    "strconv"
    "testing"
    "time"

    "github.com/testcontainers/testcontainers-go/modules/kafka"

    kc "github.com/segmentio/kafka-go" // kafka client
)

func createTopics(brokers []string, topics ...string) error {
    // to create topics when auto.create.topics.enable='false'
    conn, err := kc.Dial("tcp", brokers[0])
    if err != nil {
        return err
    }
    defer conn.Close()

    controller, err := conn.Controller()
    if err != nil {
        return err
    }
    var controllerConn *kc.Conn
    controllerConn, err = kc.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
    if err != nil {
        return err
    }
    defer controllerConn.Close()

    var topicConfigs []kc.TopicConfig
    for _, topic := range topics {
        topicConfig := kc.TopicConfig{
            Topic:             topic,
            NumPartitions:     1,
            ReplicationFactor: 1,
        }
        topicConfigs = append(topicConfigs, topicConfig)
    }

    err = controllerConn.CreateTopics(topicConfigs...)
    if err != nil {
        return err
    }

    return nil
}

func newWriter(brokers []string, topic string) *kc.Writer {
    return &kc.Writer{
        Addr:                   kc.TCP(brokers...),
        Topic:                  topic,
        Balancer:               &kc.LeastBytes{},
        AllowAutoTopicCreation: true,
        RequiredAcks:           0,
    }
}

func newReader(brokers []string, topic string) *kc.Reader {
    return kc.NewReader(kc.ReaderConfig{
        Brokers:  brokers,
        Topic:    topic,
        GroupID:  "test-group",
        MaxBytes: 10e6, // 10MB
    })
}

func TestProducerAndConsumer(t *testing.T) {
    ctx := context.Background()

    kafkaContainer, err := kafka.RunContainer(ctx, kafka.WithClusterID("test-cluster"))
    if err != nil {
        t.Fatalf("want nil, actual %v\n", err)
    }

    // Clean up the container
    defer func() {
        if err := kafkaContainer.Terminate(ctx); err != nil {
            t.Fatalf("want nil, actual %v\n", err)
        }
    }()

    state, err := kafkaContainer.State(ctx)
    if err != nil {
        t.Fatalf("want nil, actual %v\n", err)
    }

    if state.Running != true {
        t.Errorf("want true, actual %t", state.Running)
    }

    brokers, err := kafkaContainer.Brokers(ctx)
    if err != nil {
        t.Fatalf("want nil, actual %v\n", err)
    }

    topic := "test-topic"
    w := newWriter(brokers, topic)
    defer w.Close()
    r := newReader(brokers, topic)
    defer r.Close()

    err = createTopics(brokers, topic)
    if err != nil {
        t.Fatalf("want nil, actual %v\n", err)
    }
    time.Sleep(5 * time.Second)

    messages := []kc.Message{
        {
            Key:   []byte("Key-A"),
            Value: []byte("Value-A"),
        },
        {
            Key:   []byte("Key-B"),
            Value: []byte("Value-B"),
        },
        {
            Key:   []byte("Key-C"),
            Value: []byte("Value-C"),
        },
        {
            Key:   []byte("Key-D"),
            Value: []byte("Value-D!"),
        },
    }

    const retries = 3
    for i := 0; i < retries; i++ {
        ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
        defer cancel()

        // attempt to create topic prior to publishing the message
        err = w.WriteMessages(ctx, messages...)
        if errors.Is(err, kc.LeaderNotAvailable) || errors.Is(err, context.DeadlineExceeded) {
            time.Sleep(time.Millisecond * 250)
            continue
        }

        if err != nil {
            t.Fatalf("want nil, actual %v\n", err)
        }
        break
    }

    var getMessages []kc.Message
    for i := 0; i < len(messages); i++ {
        m, err := r.ReadMessage(context.Background())
        if err != nil {
            t.Fatalf("want nil, actual %v\n", err)
        }
        getMessages = append(getMessages, m)
    }

    for i := 0; i < len(messages); i++ {
        if !bytes.Equal(getMessages[i].Key, messages[i].Key) {
            t.Errorf("want %s, actual %s\n", string(messages[i].Key), string(getMessages[i].Key))
        }
        if !bytes.Equal(getMessages[i].Value, messages[i].Value) {
            t.Errorf("want %s, actual %s\n", string(messages[i].Value), string(getMessages[i].Value))
        }
    }
}

我们使用segmentio/kafka-go这个客户端来实现kafka的读写。关于如何使用segmentio/kafka-go这个客户端，可以参考我之前写的《Go社区主流Kafka客户端简要对比》。

这里我们在TestProducerAndConsumer这个用例中，先通过testcontainers-go的kafka.RunContainer启动一个Kakfa实例，然后创建了一个topic: “test-topic”。我们在写入消息前也可以不单独创建这个“test-topic”，Kafka默认启用topic自动创建，并且segmentio/kafka-go的高级API：Writer也支持AllowAutoTopicCreation的设置。不过topic的创建需要一些时间，如果要在首次写入消息时创建topic，此次写入可能会失败，需要retry。

向topic写入一条消息(实际上是一个批量Message，包括四个key-value pair)后，我们调用ReadMessage从上述topic中读取消息，并将读取的消息与写入的消息做比较。

注：近期发现kafka-go的一个可能导致内存暴涨的问题，在kafka ack返回延迟变大的时候，可能触发该问题。

下面是执行该用例的输出结果：

$ go test
2023/12/17 17:43:54 github.com/testcontainers/testcontainers-go - Connected to docker:
  Server Version: 24.0.7
  API Version: 1.43
  Operating System: CentOS Linux 7 (Core)
  Total Memory: 30984 MB
  Resolved Docker Host: unix:///var/run/docker.sock
  Resolved Docker Socket Path: /var/run/docker.sock
  Test SessionID: f76fe611c753aa4ef1456285503b0935a29795e7c0fab2ea2588029929215a08
  Test ProcessID: 27f531ee-9b5f-4e4f-b5f0-468143871004
2023/12/17 17:43:54 Creating container for image docker.io/testcontainers/ryuk:0.5.1
2023/12/17 17:43:54 Container created: 577309098f4c
2023/12/17 17:43:54 Starting container: 577309098f4c
2023/12/17 17:43:54 Container started: 577309098f4c
2023/12/17 17:43:54 Waiting for container id 577309098f4c image: docker.io/testcontainers/ryuk:0.5.1. Waiting for: &{Port:8080/tcp timeout:<nil> PollInterval:100ms}
2023/12/17 17:43:54 Creating container for image confluentinc/confluent-local:7.5.0
2023/12/17 17:43:55 Container created: 1ee11e11742b
2023/12/17 17:43:55 Starting container: 1ee11e11742b
2023/12/17 17:43:55 Container started: 1ee11e11742b
2023/12/17 17:44:15 Terminating container: 1ee11e11742b
2023/12/17 17:44:15 Container terminated: 1ee11e11742b
PASS
ok      demo    21.505s

我们看到默认情况下，testcontainer能满足与kafka交互的基本需求，并且testcontainer提供了一系列Option(WithXXX)可以对container进行定制，以满足一些扩展性的要求，但是这需要你对testcontainer提供的API有更全面的了解。

除了开箱即用的testcontainer之外，我们还可以使用另外一种方便的基于容器的技术：docker-compose来定制和启停我们需要的kafka image。接下来，我们就来看看如何使用docker-compose建立fake kafka object。

3. 使用docker-compose建立fake kafka

3.1 一个基础的基于docker-compose的fake kafka实例模板

这次我们使用bitnami提供的kafka镜像，我们先建立一个“等价”于上面“testcontainers-go”提供的kafka module的kafka实例，下面是docker-compose.yml：

// docker-compose/bitnami/plaintext/docker-compose.yml

version: "2"

services:
  kafka:
    image: docker.io/bitnami/kafka:3.6
    network_mode: "host"
    volumes:
      - "kafka_data:/bitnami"
    environment:
      # KRaft settings
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@localhost:9093
      # Listeners
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://:9092
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=PLAINTEXT
      # borrow from testcontainer
      - KAFKA_CFG_BROKER_ID=0
      - KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR=1
      - KAFKA_CFG_OFFSETS_TOPIC_NUM_PARTITIONS=1
      - KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR=1
      - KAFKA_CFG_GROUP_INITIAL_REBALANCE_DELAY_MS=0
      - KAFKA_CFG_LOG_FLUSH_INTERVAL_MESSAGES=9223372036854775807
volumes:
  kafka_data:
    driver: local

我们看到其中一些配置“借鉴”了testcontainers-go的kafka module，我们启动一下该容器：

$ docker-compose up -d
[+] Running 2/2
 ✔ Volume "plaintext_kafka_data"  Created                                                                                    0.0s
 ✔ Container plaintext-kafka-1    Started                                                                                    0.1s

依赖该容器的go测试代码与前面的TestProducerAndConsumer差不多，只是在开始处去掉了container的创建过程：

// docker-compose/bitnami/plaintext/kafka_test.go

func TestProducerAndConsumer(t *testing.T) {
    brokers := []string{"localhost:9092"}
    topic := "test-topic"
    w := newWriter(brokers, topic)
    defer w.Close()
    r := newReader(brokers, topic)
    defer r.Close()

    err := createTopics(brokers, topic)
    if err != nil {
        t.Fatalf("want nil, actual %v\n", err)
    }
    time.Sleep(5 * time.Second)
    ... ...
}

运行该测试用例，我们看到预期的结果：

go test
write message ok  Value-A
write message ok  Value-B
write message ok  Value-C
write message ok  Value-D!
PASS
ok      demo    15.143s

不过对于单元测试来说，显然我们不能手动来启动和停止kafka container，我们需要为每个用例填上setup和teardown，这样也能保证用例间的相互隔离，于是我们增加了一个docker_compose_helper.go文件，在这个文件中我们提供了一些帮助testcase启停kafka的helper函数：

// docker-compose/bitnami/plaintext/docker_compose_helper.go

package main

import (
    "fmt"
    "os/exec"
    "strings"
    "time"
)

// helpler function for operating docker container through docker-compose command

const (
    defaultCmd     = "docker-compose"
    defaultCfgFile = "docker-compose.yml"
)

func execCliCommand(cmd string, opts ...string) ([]byte, error) {
    cmds := cmd + " " + strings.Join(opts, " ")
    fmt.Println("exec command:", cmds)
    return exec.Command(cmd, opts...).CombinedOutput()
}

func execDockerComposeCommand(cmd string, cfgFile string, opts ...string) ([]byte, error) {
    var allOpts = []string{"-f", cfgFile}
    allOpts = append(allOpts, opts...)
    return execCliCommand(cmd, allOpts...)
}

func UpKakfa(composeCfgFile string) ([]byte, error) {
    b, err := execDockerComposeCommand(defaultCmd, composeCfgFile, "up", "-d")
    if err != nil {
        return nil, err
    }
    time.Sleep(10 * time.Second)
    return b, nil
}

func UpDefaultKakfa() ([]byte, error) {
    return UpKakfa(defaultCfgFile)
}

func DownKakfa(composeCfgFile string) ([]byte, error) {
    b, err := execDockerComposeCommand(defaultCmd, composeCfgFile, "down", "-v")
    if err != nil {
        return nil, err
    }
    time.Sleep(10 * time.Second)
    return b, nil
}

func DownDefaultKakfa() ([]byte, error) {
    return DownKakfa(defaultCfgFile)
}

眼尖的童鞋可能看到：在UpKakfa和DownKafka函数中我们使用了硬编码的“time.Sleep”来等待10s，通常在镜像已经pull到本地后这是有效的，但却不是最精确地等待方式，testcontainers-go/wait中提供了等待容器内程序启动完毕的多种策略，如果你想用更精确的等待方式，可以了解一下wait包。

基于helper函数，我们改造一下TestProducerAndConsumer用例：

// docker-compose/bitnami/plaintext/kafka_test.go
func TestProducerAndConsumer(t *testing.T) {
    _, err := UpDefaultKakfa()
    if err != nil {
        t.Fatalf("want nil, actual %v\n", err)
    }

    t.Cleanup(func() {
        DownDefaultKakfa()
    })
    ... ...
}

我们在用例开始处通过UpDefaultKakfa使用docker-compose将kafka实例启动起来，然后注册了Cleanup函数，用于在test case执行结束后销毁kafka实例。

下面是新版用例的执行结果：

$ go test
exec command: docker-compose -f docker-compose.yml up -d
write message ok  Value-A
write message ok  Value-B
write message ok  Value-C
write message ok  Value-D!
exec command: docker-compose -f docker-compose.yml down -v
PASS
ok      demo    36.402s

使用docker-compose的最大好处就是可以通过docker-compose.yml文件对要fake的object进行灵活的定制，这种定制与testcontainers-go的差别就是你无需去研究testcontiners-go的API。

下面是使用tls连接与kafka建立连接并实现读写的示例。

3.2 建立一个基于TLS连接的fake kafka实例

Kafka的配置复杂是有目共睹的，为了建立一个基于TLS连接，我也是花了不少时间做“试验”，尤其是listeners以及证书的配置，不下点苦功夫读文档还真是配不出来。

下面是一个基于bitnami/kafka镜像配置出来的基于TLS安全通道上的kafka实例：

// docker-compose/bitnami/tls/docker-compose.yml

# config doc:  https://github.com/bitnami/containers/blob/main/bitnami/kafka/README.md

version: "2"

services:
  kafka:
    image: docker.io/bitnami/kafka:3.6
    network_mode: "host"
    #ports:
      #- "9092:9092"
    environment:
      # KRaft settings
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@localhost:9094
      # Listeners
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,SECURED://:9093,CONTROLLER://:9094
      - KAFKA_CFG_ADVERTISED_LISTENERS=SECURED://:9093
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,SECURED:SSL,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=SECURED
      # SSL settings
      - KAFKA_TLS_TYPE=PEM
      - KAFKA_TLS_CLIENT_AUTH=none
      - KAFKA_CFG_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM=
      # borrow from testcontainer
      - KAFKA_CFG_BROKER_ID=0
      - KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR=1
      - KAFKA_CFG_OFFSETS_TOPIC_NUM_PARTITIONS=1
      - KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR=1
      - KAFKA_CFG_GROUP_INITIAL_REBALANCE_DELAY_MS=0
      - KAFKA_CFG_LOG_FLUSH_INTERVAL_MESSAGES=9223372036854775807
    volumes:
      # server.cert, server.key and ca.crt
      - "kafka_data:/bitnami"
      - "./kafka.keystore.pem:/opt/bitnami/kafka/config/certs/kafka.keystore.pem:ro"
      - "./kafka.keystore.key:/opt/bitnami/kafka/config/certs/kafka.keystore.key:ro"
      - "./kafka.truststore.pem:/opt/bitnami/kafka/config/certs/kafka.truststore.pem:ro"
volumes:
  kafka_data:
    driver: local

这里我们使用pem格式的证书和key，在上面配置中，volumes下面挂载的kafka.keystore.pem、kafka.keystore.key和kafka.truststore.pem分别对应了以前在Go中常用的名字：server-cert.pem(服务端证书), server-key.pem(服务端私钥)和ca-cert.pem(CA证书)。

这里整理了一个一键生成的脚本docker-compose/bitnami/tls/kafka-generate-cert.sh，我们执行该脚本生成所有需要的证书并放到指定位置(遇到命令行提示，只需要一路回车即可)：

$bash kafka-generate-cert.sh
.........++++++
.............................++++++
You are about to be asked to enter information that will be incorporated
into your certificate request.
What you are about to enter is what is called a Distinguished Name or a DN.
There are quite a few fields but you can leave some blank
For some fields there will be a default value,
If you enter '.', the field will be left blank.
-----
Country Name (2 letter code) [XX]:
State or Province Name (full name) []:
Locality Name (eg, city) [Default City]:
Organization Name (eg, company) [Default Company Ltd]:
Organizational Unit Name (eg, section) []:
Common Name (eg, your name or your server's hostname) []:
Email Address []:

Please enter the following 'extra' attributes
to be sent with your certificate request
A challenge password []:
An optional company name []:
Signature ok
subject=/C=XX/L=Default City/O=Default Company Ltd
Getting Private key
.....................++++++
.........++++++
You are about to be asked to enter information that will be incorporated
into your certificate request.
What you are about to enter is what is called a Distinguished Name or a DN.
There are quite a few fields but you can leave some blank
For some fields there will be a default value,
If you enter '.', the field will be left blank.
-----
Country Name (2 letter code) [XX]:
State or Province Name (full name) []:
Locality Name (eg, city) [Default City]:
Organization Name (eg, company) [Default Company Ltd]:
Organizational Unit Name (eg, section) []:
Common Name (eg, your name or your server's hostname) []:
Email Address []:

Please enter the following 'extra' attributes
to be sent with your certificate request
A challenge password []:
An optional company name []:
Signature ok
subject=/C=XX/L=Default City/O=Default Company Ltd
Getting CA Private Key

接下来，我们来改造用例，使之支持以tls方式建立到kakfa的连接：

//docker-compose/bitnami/tls/kafka_test.go

func createTopics(brokers []string, tlsConfig *tls.Config, topics ...string) error {
    dialer := &kc.Dialer{
        Timeout:   10 * time.Second,
        DualStack: true,
        TLS:       tlsConfig,
    }

    conn, err := dialer.DialContext(context.Background(), "tcp", brokers[0])
    if err != nil {
        fmt.Println("creating topic: dialer dial error:", err)
        return err
    }
    defer conn.Close()
    fmt.Println("creating topic: dialer dial ok")
    ... ...
}

func newWriter(brokers []string, tlsConfig *tls.Config, topic string) *kc.Writer {
    w := &kc.Writer{
        Addr:                   kc.TCP(brokers...),
        Topic:                  topic,
        Balancer:               &kc.LeastBytes{},
        AllowAutoTopicCreation: true,
        Async:                  true,
        //RequiredAcks:           0,
        Completion: func(messages []kc.Message, err error) {
            for _, message := range messages {
                if err != nil {
                    fmt.Println("write message fail", err)
                } else {
                    fmt.Println("write message ok", string(message.Topic), string(message.Value))
                }
            }
        },
    }

    if tlsConfig != nil {
        w.Transport = &kc.Transport{
            TLS: tlsConfig,
        }
    }
    return w
}

func newReader(brokers []string, tlsConfig *tls.Config, topic string) *kc.Reader {
    dialer := &kc.Dialer{
        Timeout:   10 * time.Second,
        DualStack: true,
        TLS:       tlsConfig,
    }

    return kc.NewReader(kc.ReaderConfig{
        Dialer:   dialer,
        Brokers:  brokers,
        Topic:    topic,
        GroupID:  "test-group",
        MaxBytes: 10e6, // 10MB
    })
}

func TestProducerAndConsumer(t *testing.T) {
    var err error
    _, err = UpDefaultKakfa()
    if err != nil {
        t.Fatalf("want nil, actual %v\n", err)
    }

    t.Cleanup(func() {
        DownDefaultKakfa()
    })

    brokers := []string{"localhost:9093"}
    topic := "test-topic"

    tlsConfig, _ := newTLSConfig()
    w := newWriter(brokers, tlsConfig, topic)
    defer w.Close()
    r := newReader(brokers, tlsConfig, topic)
    defer r.Close()
    err = createTopics(brokers, tlsConfig, topic)
    if err != nil {
        fmt.Printf("create topic error: %v, but it may not affect the later action, just ignore it\n", err)
    }
    time.Sleep(5 * time.Second)
    ... ...
}

func newTLSConfig() (*tls.Config, error) {
    /*
       // 加载 CA 证书
       caCert, err := ioutil.ReadFile("/path/to/ca.crt")
       if err != nil {
               return nil, err
       }

       // 加载客户端证书和私钥
       cert, err := tls.LoadX509KeyPair("/path/to/client.crt", "/path/to/client.key")
       if err != nil {
               return nil, err
       }

       // 创建 CertPool 并添加 CA 证书
       caCertPool := x509.NewCertPool()
       caCertPool.AppendCertsFromPEM(caCert)
    */
    // 创建并返回 TLS 配置
    return &tls.Config{
        //RootCAs:      caCertPool,
        //Certificates: []tls.Certificate{cert},
        InsecureSkipVerify: true,
    }, nil
}

在上述代码中，我们按照segmentio/kafka-go为createTopics、newWriter和newReader都加上了tls.Config参数，此外在测试用例中，我们用newTLSConfig创建一个tls.Config的实例，在这里我们一切简化处理，采用InsecureSkipVerify=true的方式与kafka broker服务端进行握手，既不验证服务端证书，也不做双向认证(mutual TLS)。

下面是修改代码后的测试用例执行结果：

$ go test
exec command: docker-compose -f docker-compose.yml up -d
creating topic: dialer dial ok
creating topic: get controller ok
creating topic: dial control listener ok
create topic error: EOF, but it may not affect the later action, just ignore it
write message error: [3] Unknown Topic Or Partition: the request is for a topic or partition that does not exist on this broker
write message ok  Value-A
write message ok  Value-B
write message ok  Value-C
write message ok  Value-D!
exec command: docker-compose -f docker-compose.yml down -v
PASS
ok      demo    38.473s

这里我们看到：createTopics虽然连接kafka的各个listener都ok，但调用topic创建时，返回EOF，但这的确不影响后续action的执行，不确定这是segmentio/kafka-go的问题，还是kafka实例的问题。另外首次写入消息时，也因为topic或partition未建立而失败，retry后消息正常写入。

通过这个例子我们看到，基于docker-compose建立fake object有着更广泛的灵活性，如果做好容器启动和停止的精准wait机制的话，我可能会更多选择这种方式。

4. 小结

本文介绍了如何在Go编程中进行依赖Kafka的单元测试，并探讨了寻找适合的Kafka fake object的策略。

对于Kafka这样的复杂系统来说，找到合适的fake object并不容易。因此，本文推荐使用容器作为fake object的策略，并分别介绍了使用testcontainers-go项目和使用docker-compose作为简化创建和清理基于容器的依赖项的工具。相对于刚刚加入testcontainers-go项目没多久的kafka module而言，使用docker-compose自定义fake object更加灵活一些。但无论哪种方法，开发人员都需要对kafka的配置有一个较为整体和深入的理解。

文中主要聚焦使用testcontainers-go和docker-compose建立fake kafka的过程，而用例并没有建立明确的sut(被测目标)，比如针对某个函数的白盒单元测试。

文本涉及的源码可以在这里下载。

“Gopher部落”知识星球旨在打造一个精品Go学习和进阶社群！高品质首发Go技术文章，“三天”首发阅读权，每年两期Go语言发展现状分析，每天提前1小时阅读到新鲜的Gopher日报，网课、技术专栏、图书内容前瞻，六小时内必答保证等满足你关于Go语言生态的所有需求！2024年，Gopher部落将进一步聚焦于如何编写雅、地道、可读、可测试的Go代码，关注代码质量并深入理解Go核心技术，并继续加强与星友的互动。欢迎大家加入！

著名云主机服务厂商DigitalOcean发布最新的主机计划，入门级Droplet配置升级为：1 core CPU、1G内存、25G高速SSD，价格5$/月。有使用DigitalOcean需求的朋友，可以打开这个链接地址：https://m.do.co/c/bff6eed92687 开启你的DO主机之路。

Gopher Daily(Gopher每日新闻) – https://gopherdaily.tonybai.com

我的联系方式：

• 微博(暂不可用)：https://weibo.com/bigwhite20xx
• 微博2：https://weibo.com/u/6484441286
• 博客：tonybai.com
• github: https://github.com/bigwhite
• Gopher Daily归档 – https://github.com/bigwhite/gopherdaily

商务合作方式：撰稿、出书、培训、在线课程、合伙创业、咨询、广告合作。