mostafamoradian/xk6-kafka

xk6-kafka Docker镜像文档

镜像概述和主要用途

xk6-kafka是k6的扩展，允许用户使用生产者（及用于调试的消费者）对Apache Kafka进行负载测试。其核心用途是测试基于Apache Kafka设计的系统，通过自动生成消息并发送到Kafka来验证消费者及整个系统的性能和正确性。本Docker镜像为该扩展提供官方容器化部署方式，简化测试环境配置。

核心功能和特性

• 多格式支持：生产/消费String、JSON、ByteArray、Avro及JSON Schema格式消息
• 模式管理：支持用户在脚本中硬编码Avro/JSON Schema键值模式，或从Schema Registry获取
• 认证机制：支持SASL PLAIN、SCRAM及SSL认证
• 主题操作：可创建、列出和删除Kafka主题
• 消息压缩：支持Gzip、Snappy、Lz4及Zstd压缩算法
• 消费者组：通过group ID从所有分区消费消息
• 指标监控：导出丰富的Kafka指标（如消息数、字节数、错误数等），支持k6阈值设置
• 调试工具：内置消费者用于验证消息发送正确性

使用场景和适用范围

适用于对Kafka生产者、消费者及依赖Kafka的系统进行负载测试的场景，尤其适合：

• 验证系统在高并发消息处理下的吞吐量和延迟
• 测试消息格式、序列化/反序列化逻辑正确性
• 验证认证机制（如SASL）和权限配置有效性
• 开发环境中快速验证Kafka主题配置和消息流

使用方法和配置说明

获取镜像

从Docker Hub拉取官方镜像：

bash
docker pull docker.xuanyuan.run/mostafamoradian/xk6-kafka:latest

基本使用示例

通过挂载本地脚本或标准输入传递测试脚本运行负载测试。

1. 通过标准输入传递脚本

bash
docker run --rm -i docker.xuanyuan.run/mostafamoradian/xk6-kafka:latest run - <本地脚本路径/test_json.js

2. 挂载本地目录运行脚本

将本地脚本目录挂载到容器中，直接运行指定脚本：

bash
docker run --rm -v $(pwd)/scripts:/scripts docker.xuanyuan.run/mostafamoradian/xk6-kafka:latest run /scripts/test_json.js

测试脚本结构

测试脚本遵循k6测试生命周期，包含以下核心阶段：

1. 导入模块

javascript
// 导入完整模块
import * as kafka from "k6/x/kafka";

// 或导入特定类和常量
import { Writer, Reader, Connection, SchemaRegistry, SCHEMA_TYPE_STRING } from "k6/x/kafka";

2. 初始化上下文（init阶段）

创建Kafka连接、生产者、消费者实例，并初始化主题：

javascript
// 创建生产者
const writer = new Writer({
  brokers: ["localhost:9092"],  // Kafka broker地址列表
  topic: "my-test-topic",       // 目标主题
  compression: "snappy"         // 消息压缩方式（可选）
});

// 创建消费者
const reader = new Reader({
  brokers: ["localhost:9092"],
  topic: "my-test-topic",
  groupID: "test-group"         // 消费者组ID（可选）
});

// 创建连接（用于主题管理）
const connection = new Connection({ address: "localhost:9092" });

// 初始化时创建主题（仅首次运行）
if (__VU == 0) {
  connection.createTopic({
    topic: "my-test-topic",
    numPartitions: 3,            // 分区数（可选）
    replicationFactor: 1         // 副本因子（可选）
  });
}

3. VU测试逻辑（default函数）

发送消息、消费消息并验证：

javascript
export default function () {
  // 生产消息
  writer.produce({
    messages: [
      {
        key: schemaRegistry.serialize({ data: "key-1", schemaType: SCHEMA_TYPE_STRING }),
        value: schemaRegistry.serialize({ data: { id: 1, value: "test" }, schemaType: SCHEMA_TYPE_JSON })
      }
    ]
  });

  // 消费消息并验证
  const messages = reader.consume({ limit: 10 });  // 最多消费10条消息
  if (messages.length > 0) {
    console.log(`消费消息: ${JSON.stringify(messages[0].value)}`);
    // 可添加k6 checks验证消息内容、格式等
  }
}

4. 清理资源（teardown阶段）

关闭连接并清理测试主题：

javascript
export function teardown() {
  // 删除测试主题（可选）
  connection.deleteTopic("my-test-topic");
  
  // 关闭所有连接
  writer.close();
  reader.close();
  connection.close();
}

环境配置示例

SASL认证配置

javascript
const writer = new Writer({
  brokers: ["kafka-broker:9092"],
  topic: "sasl-test-topic",
  sasl: {
    mechanism: "SCRAM-SHA-256",  // 认证机制
    username: "test-user",
    password: "test-pass"
  },
  tls: {
    enable: true  // 启用TLS
  }
});

Avro格式消息（带Schema Registry）

javascript
const schemaRegistry = new SchemaRegistry({ url: "http://schema-registry:8081" });

writer.produce({
  messages: [
    {
      key: schemaRegistry.serialize({
        data: "user-1",
        schemaType: "avro",
        schema: { type: "string" }  // 或从Schema Registry获取
      }),
      value: schemaRegistry.serialize({
        data: { id: 1, name: "test" },
        schemaType: "avro",
        schema: { type: "record", name: "User", fields: [{ name: "id", type: "int" }, { name: "name", type: "string" }] }
      })
    }
  ]
});

支持的指标

xk6-kafka导出以下核心指标用于监控测试过程：

示例脚本

官方仓库scripts目录提供多种功能示例，关键脚本包括：

• test_json.js：JSON格式消息生产/消费测试
• test_avro_with_schema_registry.js：使用Schema Registry的Avro消息测试
• test_sasl_auth.js：SASL认证测试
• test_consumer_group.js：消费者组多分区消费测试
• test_topics.js：主题创建/删除/列出操作测试