golang连接kafka

dazheng · · 6428 次点击 · · 开始浏览    
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

golang连接kafka的例子好少,而支持offset追踪者更少,但也是基础的要求。”github.com/bsm/sarama-cluster”结合“github.com/Shopify/sarama”满足了此要求。
环境:
golang 1.7
kafka 0.10
centos7.2

package main
import (
    "fmt"
    "os"
    "strings"
    "time"

    "github.com/Shopify/sarama"
    "github.com/bsm/sarama-cluster" //support automatic consumer-group rebalancing and offset tracking
    "github.com/golang/glog"
)

func main() {
    groupID := "group-1"
    topicList := "topic_1"
    config := cluster.NewConfig()
    config.Consumer.Return.Errors = true
    config.Group.Return.Notifications = true
    config.Consumer.Offsets.CommitInterval = 1 * time.Second
    config.Consumer.Offsets.Initial = sarama.OffsetNewest //初始从最新的offset开始

    c, err := cluster.NewConsumer(strings.Split("localhost:9092", ","), groupID, strings.Split(topicList, ","), config)
    if err != nil {
        glog.Errorf("Failed open consumer: %v", err)
        return
    }
    defer c.Close()
    go func() {
        for err := range c.Errors() {
            glog.Errorf("Error: %s\n", err.Error())
        }
    }()

    go func() {
        for note := range c.Notifications() {
            glog.Infof("Rebalanced: %+v\n", note)
        }
    }()

    for msg := range c.Messages() {
        fmt.Fprintf(os.Stdout, "%s/%d/%d\t%s\n", msg.Topic, msg.Partition, msg.Offset, msg.Value)
        c.MarkOffset(msg, "") //MarkOffset 并不是实时写入kafka,有可能在程序crash时丢掉未提交的offset
    }
}

参考:
http://pastebin.com/9ZsnP2eU
https://github.com/Shopify/sarama
https://github.com/bsm/sarama-cluster

本文来自:CSDN博客

感谢作者:dazheng

查看原文:golang连接kafka

入群交流(和以上内容无关):Go中文网 QQ 交流群:798786647 或加微信入微信群:274768166 备注:入群;关注公众号:Go语言中文网

6428 次点击  
加入收藏 微博
被以下专栏收入,发现更多相似内容
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传