blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-12-13 16:47:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

kafka msg data conversion

Browse Source
This commit is contained in:
pluto 2023-08-18 17:59:41 +08:00
parent b336241271
commit 9578ce4b21
1 changed files with 17 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
test/data-conversion/kafka-conversation.go
View File

@ -17,6 +17,7 @@ package data_conversion
import ( import (
"fmt" "fmt"
"github.com/Shopify/sarama" "github.com/Shopify/sarama"
"sync"
) )
var ( var (
@ -27,6 +28,7 @@ var (
var consumer sarama.Consumer var consumer sarama.Consumer
var producer sarama.SyncProducer var producer sarama.SyncProducer
var wg sync.WaitGroup
func init() { func init() {
@ -71,15 +73,23 @@ func GetMessage() {
fmt.Printf("fail to get list of partition:err%v\n", err) fmt.Printf("fail to get list of partition:err%v\n", err)
} }
fmt.Println(partitionList) fmt.Println(partitionList)
for partition := range partitionList { // iterate over all partitions for partition := range partitionList {
// Create a corresponding partition consumer for each partition
pc, err := consumer.ConsumePartition(topic, int32(partition), sarama.OffsetNewest) pc, err := consumer.ConsumePartition(topic, int32(partition), sarama.OffsetNewest)
if err != nil { if err != nil {
fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err) panic(err)
}
// Asynchronously consume information from each partition
for msg := range pc.Messages() {
fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)
} }
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\n", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
}
}(pc)
wg.Wait()
consumer.Close()
} }
} }