Broker 实现了 river.Handler 中的核心函数 OnEvent, 添加了校验, offset自动存储功能。 其余 String()、OnAlert()、OnClose() 函数委托给 BrokerHandler 实现。 Broker example: broker, err := New(brokerConfig) broker.SetHandler(...) go broker.Consume(func(msg *sarama.ConsumerMessage) error { // consume your event }) err := broker.Pipe(river.River, river.FromFile).