kafkaUtil

kafkaUtil

a util for Quick to use sarama kafka client

Usage

AsyncProducer

producer, err := NewProducer(
    []string{"172.16.9.207:9092"},
    MarshalMsgFunc(codec.JSONMarshal),
    Async())
if err != nil {
    log.Fatalf("kafka error: [%s]", err)
}

go func() {
    for {
        s := <-producer.AsyncSuccesses()
        fmt.Printf("success: %v \n", s)
    }
}()

go func() {
    for {
        e := <-producer.AsyncErrors()
        fmt.Printf("error: %v \n", e)
    }
}()

for {
    t := <-time.After(200 * time.Millisecond)
    person := &example.Person{
        Name: fmt.Sprintf("%d", t.UnixNano()),
        Age:  uint8(t.Nanosecond()),
    }
    producer.AsyncSendMessages("person", ProducerMsg{
        Headers: map[string]string{"traceId": "123456"},
        Body:    person,
    })
}

SyncProducer

producer, err := NewProducer(
    []string{"172.16.9.207:9092"},
    MarshalMsgFunc(codec.JSONMarshal),
    Sync())
if err != nil {
    log.Fatalf("kafka error: [%s]", err)
}

for {
    t := <-time.After(200 * time.Millisecond)
    person := &example.Person{
        Name: fmt.Sprintf("%d", t.UnixNano()),
        Age:  uint8(t.Nanosecond()),
    }
    send, offset, err := producer.SyncSendMessages("person", ProducerMsg{
        Headers: map[string]string{"traceId": "123456"},
        Body:    person,
    })
    fmt.Println(send, offset, err)
}

Consumer

consumer, err := NewConsumer([]string{"172.16.9.207:9092"}, Newest())
if err != nil {
    log.Fatalf("kafka error: [%s]", err)
}
defer consumer.Close()

for i:=0;i<10;i++ {
    ctx, _ := context.WithTimeout(context.Background(), time.Second*20)
    publications := consumer.GroupConsume(ctx, []string{"person"})
    go func() {
        for e := range consumer.Errors() {
            t.Log(e)
        }
    }()
    for p := range publications {
        t.Log(p.Msg.Offset)
        p.Session.MarkMessage(p.Msg, "")
    }
    t.Log("=======================================")
}
t.Log("finished")