go-zero与Kafka+Avro的实践：构建高性能的交互式数据处理系统

[kafka]
addrs = ["localhost:9092"]
version = "2.0.0"
maxMessageBytes = 10000000

var (
    topic = "test"
)

func (s *Service) Produce(msg []byte) error {
    p, err := kafka.NewProducer(s.cfg.Kafka)
    if err != nil {
        return err
    }
    defer p.Close()

    return p.Send(context.TODO(), &kafka.Message{
        Key:   []byte(topic),
        Value: msg,
    })
}

{
    "namespace": "com.example",
    "type": "record",
    "name": "User",
    "fields": [
        {
            "name": "name",
            "type": "string"
        },
        {
            "name": "age",
            "type": "int"
        }
    ]
}

$ go run github.com/gogo/protobuf/protoc-gen-gogofaster --proto_path=./ example.proto --gogofaster_out

// 创建消费组
group, err := kafka.NewGroup(s.cfg.Kafka, "test", kafka.WithGroupID("test-group"))
if err != nil {
    return nil, err
}
// 创建消费者
consumer, err := group.NewConsumer(context.Background(), []string{"test"})
if err != nil {
    return nil, err
}
// 启动并发协程
for i := 0; i < s.cfg.WorkerNum; i++ {
    go func() {
        for {
            select {
                // 从同步通道中获取新消息
                case msg := <-msgs:
                    if err := s.processMsg(msg); err != nil {
                        log.Errorf("failed to process message(%v): %v", msg.Value, err)
                    }
                }
        }
    }()
}
// 消费数据
for {
    m, err := consumer.FetchMessage(context.Background())
    if err != nil {
        log.Errorf("failed to fetch message: %v", err)
        continue
    }
    // 将新消息发送到同步通道中
    msgs <- m
}