Golang中如何Mock Kafka依赖与Writer编写单元测试

要mock Kafka相关逻辑，核心是面向接口编程，解耦业务逻辑和外部依赖的具体实现，不要在Server方法里硬编码直接调用Kafka的具体操作。

1. 抽象Kafka操作接口 #

把所有需要mock的Kafka相关行为（创建topic、生产消息）统一定义为接口，业务逻辑只依赖接口定义，不依赖具体实现：

type KafkaClient interface {
    CreateTopic(brokerUrl string, topic string) error
    ProduceEvents(key string, val string) error
}

2. 重构Server结构体和相关方法 #

把原来直接依赖的*kafka.Writer、包级函数createKafkaTopic替换为对KafkaClient接口的依赖，同时把原来硬编码在StartServer里的Kafka初始化逻辑挪到真实实现中：

首先实现生产环境用的真实Kafka客户端：

type RealKafkaClient struct {
    writer *kafka.Writer
}

// 实现KafkaClient接口的CreateTopic方法
func (r *RealKafkaClient) CreateTopic(brokerUrl string, topic string) error {
    // 原有createKafkaTopic逻辑放在这里，注意修改原有panic逻辑为返回error
    // 若原有第三方实现会panic，可在此处加recover逻辑捕获后返回error
    return createKafkaTopic(brokerUrl, topic)
}

// 实现KafkaClient接口的ProduceEvents方法
func (r *RealKafkaClient) ProduceEvents(key string, val string) error {
    msg := kafka.Message{
        Key:   []byte(key),
        Value: []byte(val),
    }
    return r.writer.WriteMessages(context.Background(), msg)
}

然后改造Server结构和对应方法：

type Server struct {
    grpcServerPort int
    grpcServer     *grpc.Server
    kafkaClient    KafkaClient // 替换原有的writer字段
}

// NewServer 支持注入KafkaClient依赖，生产传真实实现，测试传mock
func NewServer(port int, kafkaClient KafkaClient) *Server {
    gs := grpc.NewServer()
    return &Server{
        grpcServerPort: port,
        grpcServer:     gs,
        kafkaClient:    kafkaClient,
    }
}

func (s *Server) StartServer() error {
    // 调用接口方法，不再直接依赖硬编码的包级函数
    if err := s.kafkaClient.CreateTopic("brokker_url", "my_topic"); err != nil {
        return fmt.Errorf("create kafka topic failed: %w", err)
    }

    listener, err := net.Listen("tcp", fmt.Sprintf(":%d", s.grpcServerPort))
    if err != nil {
        return fmt.Errorf("failed to listen: %w", err)
    }

    go s.grpcServer.Serve(listener)
    return nil
}

func (s *Server) produceEvents(key string, val string) error {
    return s.kafkaClient.ProduceEvents(key, val)
}

生产环境初始化示例：

func main() {
    realKafka := &RealKafkaClient{
        writer: &kafka.Writer{
            Addr:        kafka.TCP("your_kafka_broker_urls"),
            Topic:       "my_topic",
            Balancer:    &kafka.Hash{},
            MaxAttempts: 1,
            BatchSize:   1,
        },
    }
    s := NewServer(8080, realKafka)
    if err := s.StartServer(); err != nil {
        log.Fatalf("server start failed: %v", err)
    }
    // 后续业务逻辑
}

3. 实现Mock客户端编写单元测试 #

测试时不需要启动真实Kafka，只要实现一个mock版本的KafkaClient即可，还可以自定义返回值、记录调用参数方便断言：

// MockKafkaClient 测试用mock实现
type MockKafkaClient struct {
    // 记录调用状态
    CreateTopicCalled bool
    CalledTopicName   string
    ProduceCalled     bool
    LastProduceKey    string
    LastProduceVal    string
    // 自定义返回的错误
    MockReturnErr error
}

func (m *MockKafkaClient) CreateTopic(brokerUrl string, topic string) error {
    m.CreateTopicCalled = true
    m.CalledTopicName = topic
    return m.MockReturnErr
}

func (m *MockKafkaClient) ProduceEvents(key string, val string) error {
    m.ProduceCalled = true
    m.LastProduceKey = key
    m.LastProduceVal = val
    return m.MockReturnErr
}

测试用例示例：

func TestServer_StartServer(t *testing.T) {
    mockKafka := &MockKafkaClient{MockReturnErr: nil}
    s := NewServer(19999, mockKafka) // 选一个空闲测试端口
    err := s.StartServer()
    if err != nil {
        t.Fatalf("start server unexpected error: %v", err)
    }
    defer s.grpcServer.Stop()

    // 断言CreateTopic被正确调用
    if !mockKafka.CreateTopicCalled {
        t.Error("expected CreateTopic to be called, but not invoked")
    }
    if mockKafka.CalledTopicName != "my_topic" {
        t.Errorf("expected topic name 'my_topic', got '%s'", mockKafka.CalledTopicName)
    }
}

func TestServer_ProduceEvents(t *testing.T) {
    mockKafka := &MockKafkaClient{MockReturnErr: nil}
    s := NewServer(19998, mockKafka)
    testKey, testVal := "order_1", "paid"

    err := s.produceEvents(testKey, testVal)
    if err != nil {
        t.Fatalf("produce events unexpected error: %v", err)
    }

    // 断言生产消息逻辑被正确调用
    if !mockKafka.ProduceCalled {
        t.Error("expected ProduceEvents to be called, but not invoked")
    }
    if mockKafka.LastProduceKey != testKey || mockKafka.LastProduceVal != testVal {
        t.Errorf("produce param mismatch, expect key=%s val=%s, got key=%s val=%s",
            testKey, testVal, mockKafka.LastProduceKey, mockKafka.LastProduceVal)
    }
}

注意事项 #

• 原有createKafkaTopic触发panic属于不合理实现，底层依赖操作应该返回error给上层，由上层决定是重试、熔断还是终止服务，不要直接在底层逻辑panic。
• 如果不想手写mock，可以用mockgen类的工具，基于定义好的KafkaClient接口自动生成功能更完善的mock实现，支持调用次数、参数匹配等更复杂的断言。
• 单元测试不要依赖真实外部服务（Kafka、数据库等），所有外部IO依赖都应该通过接口抽象替换为mock，保证测试用例运行快、无环境依赖。

内容的提问来源于stack exchange，提问作者Ashutosh Pandey