Liftbridge Golang 客户端使用指南

Liftbridge 是一个轻量级的容错消息流处理系统，构建在 NATS 之上，提供了类似 Kafka 的日志流功能。下面我将介绍如何在 Golang 中使用 Liftbridge 客户端库。

安装

首先安装 Liftbridge Go 客户端：

go get github.com/liftbridge-io/go-liftbridge/v2

基本使用

1. 连接到 Liftbridge 服务器

package main

import (
	"context"
	"log"
	"time"

	"github.com/liftbridge-io/go-liftbridge/v2"
)

func main() {
	// 创建客户端连接
	client, err := liftbridge.Connect([]string{"localhost:9292"})
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	// 其他操作...
}

2. 创建流

// 创建流
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

err := client.CreateStream(ctx, "example-stream", "example-subject")
if err != nil && err != liftbridge.ErrStreamExists {
	log.Fatal(err)
}

3. 发布消息

// 发布消息
ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

_, err = client.Publish(ctx, "example-stream", []byte("Hello, Liftbridge!"))
if err != nil {
	log.Fatal(err)
}

4. 订阅消息

// 创建订阅
ctx, cancel = context.WithCancel(context.Background())
defer cancel()

err = client.Subscribe(ctx, "example-stream", func(msg *liftbridge.Message, err error) {
	if err != nil {
		log.Printf("Subscription error: %v", err)
		return
	}
	log.Printf("Received message: %s", string(msg.Value()))
}, liftbridge.StartAtEarliestReceived())
if err != nil {
	log.Fatal(err)
}

// 保持程序运行
select {}

高级特性

1. 分区支持

// 创建分区流
err := client.CreateStream(ctx, "partitioned-stream", "partitioned-subject",
	liftbridge.Partitions(3))
if err != nil && err != liftbridge.ErrStreamExists {
	log.Fatal(err)
}

// 发布到特定分区
_, err = client.Publish(ctx, "partitioned-stream", []byte("Partitioned message"),
	liftbridge.Partition(1))
if err != nil {
	log.Fatal(err)
}

2. 消息确认

// 发布需要确认的消息
ack, err := client.Publish(ctx, "example-stream", []byte("Message requiring ack"),
	liftbridge.AckPolicyAll())
if err != nil {
	log.Fatal(err)
}

// 等待确认
select {
case <-ack:
	log.Println("Message was acked by all replicas")
case <-time.After(5 * time.Second):
	log.Println("Timed out waiting for ack")
}

3. 消费者组

// 使用消费者组订阅
err = client.Subscribe(ctx, "example-stream", func(msg *liftbridge.Message, err error) {
	// 处理消息
}, 
	liftbridge.StartAtLatestReceived(),
	liftbridge.ConsumerGroup("my-group"))
if err != nil {
	log.Fatal(err)
}

容错处理

// 带重试的连接
client, err := liftbridge.Connect([]string{"localhost:9292"},
	liftbridge.MaxReconnects(5),
	liftbridge.ReconnectDelay(2 * time.Second))
if err != nil {
	log.Fatal(err)
}

// 订阅时处理错误
err = client.Subscribe(ctx, "example-stream", func(msg *liftbridge.Message, err error) {
	if err != nil {
		if err == liftbridge.ErrUnavailable {
			log.Println("Server unavailable, will retry...")
			return
		}
		log.Printf("Fatal error: %v", err)
		return
	}
	// 处理消息
})

完整示例

package main

import (
	"context"
	"log"
	"time"

	"github.com/liftbridge-io/go-liftbridge/v2"
)

func main() {
	// 连接
	client, err := liftbridge.Connect([]string{"localhost:9292"})
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	// 创建流
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	
	err = client.CreateStream(ctx, "demo-stream", "demo.subject")
	if err != nil && err != liftbridge.ErrStreamExists {
		log.Fatal(err)
	}

	// 订阅
	ctx, cancel = context.WithCancel(context.Background())
	defer cancel()
	
	go func() {
		err := client.Subscribe(ctx, "demo-stream", func(msg *liftbridge.Message, err error) {
			if err != nil {
				log.Printf("Error: %v", err)
				return
			}
			log.Printf("Received: %s", string(msg.Value()))
		}, liftbridge.StartAtEarliestReceived())
		
		if err != nil {
			log.Fatal(err)
		}
	}()

	// 发布消息
	for i := 0; i < 10; i++ {
		_, err = client.Publish(ctx, "demo-stream", []byte("Message "+string(i)))
		if err != nil {
			log.Fatal(err)
		}
		time.Sleep(1 * time.Second)
	}

	// 等待
	time.Sleep(10 * time.Second)
}

总结

Liftbridge 提供了轻量级的消息流处理能力，特别适合需要 Kafka 类似功能但希望更简单部署的场景。Go 客户端库提供了简洁的 API 来实现流创建、消息发布和订阅等核心功能，并内置了容错机制。

关键优势：

• 轻量级，易于部署
• 与 NATS 生态集成
• 支持分区和消费者组
• 提供消息确认机制
• 简单的 Go API

对于需要更复杂功能的场景，可以考虑结合 NATS JetStream 或其他消息系统使用。