Benthos：Golang多协议消息流处理插件库

Benthos 是一个强大的流处理工具，用 Go 编写，支持多种消息协议和处理转换操作。它特别适合构建数据管道、ETL流程和消息路由系统。

核心特性

1. 多协议支持：Kafka、RabbitMQ、HTTP、AWS SQS/SNS、NATS等
2. 丰富处理能力：过滤、转换、批处理、缓存等
3. 插件架构：易于扩展新输入/输出和处理组件
4. 监控：内置Prometheus指标和健康检查

安装与基本使用

# 安装Benthos
go get github.com/benthosdev/benthos/v4

示例配置

下面是一个简单的YAML配置示例，从STDIN读取，处理后输出到STDOUT：

input:
  stdin: {}

pipeline:
  processors:
    - bloblang: |
        root = this
        root.message = this.content.uppercase()
        root.metadata.link_count = this.links.length()

output:
  stdout: {}

Go代码集成示例

package main

import (
	"context"
	"fmt"
	"os"

	"github.com/benthosdev/benthos/v4/public/service"

	// 导入所有标准组件
	_ "github.com/benthosdev/benthos/v4/public/components/all"
)

func main() {
	// 定义自定义Bloblang处理器
	processorConfig := `
bloblang: |
  root = this
  root.timestamp = now().format_timestamp("2006-01-02T15:04:05Z07:00")
  root.processed = true
`

	// 创建Benthos流构建器
	builder := service.NewStreamBuilder()

	// 设置输入
	err := builder.AddInputYAML(`
kafka:
  addresses: [ "localhost:9092" ]
  topics: [ "input-topic" ]
  consumer_group: "benthos-group"
`)
	if err != nil {
		fmt.Printf("Failed to add input: %v\n", err)
		os.Exit(1)
	}

	// 添加处理器
	err = builder.AddProcessorYAML(processorConfig)
	if err != nil {
		fmt.Printf("Failed to add processor: %v\n", err)
		os.Exit(1)
	}

	// 设置输出
	err = builder.AddOutputYAML(`
http_client:
  url: http://localhost:8080/ingest
  verb: POST
  headers:
    Content-Type: application/json
`)
	if err != nil {
		fmt.Printf("Failed to add output: %v\n", err)
		os.Exit(1)
	}

	// 构建流
	stream, err := builder.Build()
	if err != nil {
		fmt.Printf("Failed to build stream: %v\n", err)
		os.Exit(1)
	}

	// 运行流
	ctx := context.Background()
	if err := stream.Run(ctx); err != nil {
		fmt.Printf("Stream error: %v\n", err)
		os.Exit(1)
	}
}

高级功能示例

条件路由

input:
  kafka:
    addresses: [ "localhost:9092" ]
    topics: [ "input-topic" ]
    consumer_group: "benthos-group"

pipeline:
  processors:
    - switch:
      - check: this.type == "A"
        processors:
          - bloblang: 'root = this.message.uppercase()'
      - check: this.type == "B"
        processors:
          - bloblang: 'root = this.message.lowercase()'

output:
  switch:
    cases:
      - check: this.type == "A"
        output:
          kafka:
            addresses: [ "localhost:9092" ]
            topic: "output-topic-a"
      - check: this.type == "B"
        output:
          kafka:
            addresses: [ "localhost:9092" ]
            topic: "output-topic-b"

批处理

input:
  http_client:
    url: http://example.com/api/data
    verb: GET

pipeline:
  processors:
    - batch:
        count: 10
        processors:
          - archive:
              format: lines
          - compress:
              algorithm: gzip

output:
  s3:
    bucket: my-bucket
    path: "data/${!timestamp("2006-01-02")}/${!uuid_v4()}.json.gz"

自定义插件开发

Benthos允许开发自定义组件。以下是自定义处理器的示例：

package main

import (
	"context"
	"fmt"

	"github.com/benthosdev/benthos/v4/public/service"
)

func init() {
	configSpec := service.NewConfigSpec().
		Field(service.NewStringField("prefix"))

	err := service.RegisterProcessor(
		"prefix_adder",
		configSpec,
		func(conf *service.ParsedConfig, mgr *service.Resources) (service.Processor, error) {
			prefix, err := conf.FieldString("prefix")
			if err != nil {
				return nil, err
			}
			return &prefixAdder{prefix: prefix}, nil
		})
	if err != nil {
		panic(err)
	}
}

type prefixAdder struct {
	prefix string
}

func (p *prefixAdder) Process(ctx context.Context, msg *service.Message) (service.MessageBatch, error) {
	content, err := msg.AsBytes()
	if err != nil {
		return nil, err
	}

	newContent := fmt.Sprintf("%s%s", p.prefix, content)
	msg.SetBytes(newContent)

	return service.MessageBatch{msg}, nil
}

func (p *prefixAdder) Close(ctx context.Context) error {
	return nil
}

func main() {
	service.RunCLI()
}

监控与管理

Benthos提供内置的HTTP API用于监控和管理：

http:
  address: 0.0.0.0:4195
  enabled: true
  debug_endpoints: false

可以通过/metrics端点获取Prometheus指标，/ready用于健康检查。

Benthos是一个功能强大且灵活的工具，特别适合构建复杂的数据处理流水线。它的插件架构和丰富的内置组件使其能够轻松集成到各种系统中。