使用Watermill构建高效事件驱动应用

Watermill是一个Go语言库，专门用于构建事件驱动的应用程序。它提供了简单而强大的抽象来处理消息流，支持多种消息代理（如Kafka、RabbitMQ、Google Pub/Sub等），并内置了重试、限流等生产级特性。

Watermill核心概念

1. 消息(Message)：携带数据和元数据的基本单元
2. 发布者(Publisher)：发送消息的组件
3. 订阅者(Subscriber)：接收消息的组件
4. 路由器(Router)：协调消息流的核心组件

安装Watermill

go get github.com/ThreeDotsLabs/watermill

基本使用示例

1. 创建简单的发布-订阅应用

package main

import (
	"context"
	"log"
	"time"

	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/ThreeDotsLabs/watermill/pubsub/gochannel"
)

func main() {
	// 创建GoChannel发布订阅（内存实现，适合开发和测试）
	pubSub := gochannel.NewGoChannel(
		gochannel.Config{},
		watermill.NewStdLogger(false, false),
	)

	// 创建路由器
	router, err := message.NewRouter(message.RouterConfig{}, watermill.NewStdLogger(false, false))
	if err != nil {
		panic(err)
	}

	// 添加处理程序
	router.AddHandler(
		"handler_example",  // 处理程序名称
		"example.topic",    // 订阅的主题
		pubSub,            // 订阅的消息来源
		"output.topic",     // 发布的目标主题
		pubSub,            // 发布的消息目标
		func(msg *message.Message) ([]*message.Message, error) {
			// 处理消息
			log.Printf("Received message: %s, payload: %s", msg.UUID, string(msg.Payload))

			// 可以返回多个消息
			newMsg := message.NewMessage(watermill.NewUUID(), []byte("processed: "+string(msg.Payload)))
			return []*message.Message{newMsg}, nil
		},
	)

	// 启动路由器
	go func() {
		if err := router.Run(context.Background()); err != nil {
			panic(err)
		}
	}()

	// 等待路由器启动
	<-router.Running()

	// 发布消息
	for i := 0; i < 5; i++ {
		msg := message.NewMessage(watermill.NewUUID(), []byte("Hello, world!"))
		if err := pubSub.Publish("example.topic", msg); err != nil {
			panic(err)
		}
		time.Sleep(time.Second)
	}

	// 等待处理完成
	time.Sleep(5 * time.Second)
	router.Close()
}

2. 使用Kafka作为消息代理

package main

import (
	"context"
	"log"
	"time"

	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/ThreeDotsLabs/watermill/message/router/middleware"
	"github.com/ThreeDotsLabs/watermill-kafka/v2/pkg/kafka"
)

func main() {
	logger := watermill.NewStdLogger(true, true)

	// 创建Kafka发布者
	kafkaPublisher, err := kafka.NewPublisher(
		kafka.PublisherConfig{
			Brokers:   []string{"localhost:9092"},
			Marshaler: kafka.DefaultMarshaler{},
		},
		logger,
	)
	if err != nil {
		panic(err)
	}

	// 创建Kafka订阅者
	kafkaSubscriber, err := kafka.NewSubscriber(
		kafka.SubscriberConfig{
			Brokers:       []string{"localhost:9092"},
			Unmarshaler:   kafka.DefaultMarshaler{},
			ConsumerGroup: "test_group",
		},
		logger,
	)
	if err != nil {
		panic(err)
	}

	// 创建路由器
	router, err := message.NewRouter(message.RouterConfig{}, logger)
	if err != nil {
		panic(err)
	}

	// 添加中间件
	router.AddMiddleware(
		middleware.Recoverer,       // 处理panic
		middleware.Retry{
			MaxRetries:      3,
			InitialInterval: time.Millisecond * 100,
			Logger:         logger,
		}.Middleware,               // 自动重试
		middleware.Throttle(100).Middleware, // 限流
	)

	// 添加处理程序
	router.AddHandler(
		"kafka_handler",
		"input_topic",
		kafkaSubscriber,
		"output_topic",
		kafkaPublisher,
		processMessage,
	)

	// 启动路由器
	go func() {
		if err := router.Run(context.Background()); err != nil {
			panic(err)
		}
	}()

	// 等待路由器启动
	<-router.Running()

	// 发布测试消息
	for i := 0; i < 10; i++ {
		msg := message.NewMessage(watermill.NewUUID(), []byte("Kafka message"))
		if err := kafkaPublisher.Publish("input_topic", msg); err != nil {
			log.Println("cannot publish message:", err)
		}
	}

	// 等待处理完成
	time.Sleep(10 * time.Second)
	router.Close()
}

func processMessage(msg *message.Message) ([]*message.Message, error) {
	log.Printf("Processing Kafka message: %s, payload: %s", msg.UUID, string(msg.Payload))
	
	// 可以添加业务逻辑处理
	newMsg := message.NewMessage(watermill.NewUUID(), []byte("processed: "+string(msg.Payload)))
	return []*message.Message{newMsg}, nil
}

Watermill高级特性

1. 使用装饰器模式

// 创建装饰器
func loggingDecorator(h message.HandlerFunc) message.HandlerFunc {
	return func(msg *message.Message) ([]*message.Message, error) {
		log.Printf("Before processing: %s", msg.UUID)
		messages, err := h(msg)
		log.Printf("After processing: %s", msg.UUID)
		return messages, err
	}
}

// 使用装饰器
router.AddHandler(
	"decorated_handler",
	"decorated_topic",
	pubSub,
	"output_topic",
	pubSub,
	loggingDecorator(processMessage),
)

2. 使用消息组

// 添加消息组中间件
router.AddMiddleware(middleware.NewGroup(10).Middleware)

// 处理程序会按消息组批量处理
router.AddHandler(
	"grouped_handler",
	"grouped_topic",
	pubSub,
	"output_topic",
	pubSub,
	func(msg *message.Message) ([]*message.Message, error) {
		group := middleware.MessageGroupFromCtx(msg.Context())
		log.Printf("Processing message in group %s", group)
		return []*message.Message{msg}, nil
	},
)

最佳实践

1. 错误处理：始终正确处理错误并考虑重试策略
2. 幂等性：确保消息处理是幂等的，避免重复处理导致问题
3. 监控：添加适当的日志和监控以跟踪消息流
4. 测试：使用Watermill的测试工具进行组件测试
5. 性能优化：根据需求调整并发度和批处理大小

Watermill提供了灵活而强大的工具来构建事件驱动系统，通过合理的架构设计，可以创建高效、可靠的消息处理流水线。