分享
  1. 首页
  2. 文章

golang kafka

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

golang kafka – hello world

https://github.com/Shopify/sarama

https://shopify.github.io/sarama/

consumer.go

package main
import (
"fmt"
"github.com/Shopify/sarama"
"log"
"os"
"strings"
"sync"
)
var (
wg sync.WaitGroup
logger = log.New(os.Stderr, "[srama]", log.LstdFlags)
)
func main() {
sarama.Logger = logger
consumer, err := sarama.NewConsumer(strings.Split("localhost:9092", ","), nil)
if err != nil {
logger.Println("Failed to start consumer: %s", err)
}
partitionList, err := consumer.Partitions("hello")
if err != nil {
logger.Println("Failed to get the list of partitions: ", err)
}
for partition := range partitionList {
pc, err := consumer.ConsumePartition("hello", int32(partition), sarama.OffsetNewest)
if err != nil {
logger.Printf("Failed to start consumer for partition %d: %s\n", partition, err)
}
defer pc.AsyncClose()
wg.Add(1)
go func(sarama.PartitionConsumer) {
defer wg.Done()
for msg := range pc.Messages() {
fmt.Printf("Partition:%d, Offset:%d, Key:%s, Value:%s", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
fmt.Println()
}
}(pc)
}
wg.Wait()
logger.Println("Done consuming topic hello")
consumer.Close()
}

producer.go

package main
import (
"github.com/Shopify/sarama"
"log"
"os"
"strings"
)
var (
logger = log.New(os.Stderr, "[srama]", log.LstdFlags)
)
func main() {
sarama.Logger = logger
config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll
config.Producer.Partitioner = sarama.NewRandomPartitioner
msg := &sarama.ProducerMessage{}
msg.Topic = "hello"
msg.Partition = int32(-1)
msg.Key = sarama.StringEncoder("key")
msg.Value = sarama.ByteEncoder("你好, 世界!")
producer, err := sarama.NewSyncProducer(strings.Split("localhost:9092", ","), config)
if err != nil {
logger.Println("Failed to produce message: %s", err)
os.Exit(500)
}
defer producer.Close()
partition, offset, err := producer.SendMessage(msg)
if err != nil {
logger.Println("Failed to produce message: ", err)
}
logger.Printf("partition=%d, offset=%d\n", partition, offset)
}

此条目发表在Golang, Linux分类目录。将固定链接加入收藏夹。

有疑问加站长微信联系(非本文作者)

本文来自:博客园

感谢作者:simbadan

查看原文:golang kafka

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

关注微信
21196 次点击
暂无回复
添加一条新回复 (您需要 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传

用户登录

没有账号?注册
或
(追記) (追記ここまで)

今日阅读排行

    加载中
(追記) (追記ここまで)

一周阅读排行

    加载中

关注我

  • 扫码关注领全套学习资料 关注微信公众号
  • 加入 QQ 群:
    • 192706294(已满)
    • 731990104(已满)
    • 798786647(已满)
    • 729884609(已满)
    • 977810755(已满)
    • 815126783(已满)
    • 812540095(已满)
    • 1006366459(已满)
    • 692541889

  • 关注微信公众号
  • 加入微信群:liuxiaoyan-s,备注入群
  • 也欢迎加入知识星球 Go粉丝们(免费)

给该专栏投稿 写篇新文章

每篇文章有总共有 5 次投稿机会

收入到我管理的专栏 新建专栏