Golang 与Python 连接kafka

127 阅读 0 评论 84 点赞

我是靠谱客的博主负责保温杯，这篇文章主要介绍Golang 与Python 连接kafka，现在分享给大家，希望可以做个参考。

1. 安装

python	golang
pip install kafka-python	git clone https://github.com/edenhill/librdkafka.git cd librdkafka ./configure --prefix=/usr make && make install profile: export PKG_CONFIG_PATH=/usr/lib/pkgconfig go get -u github.com/confluentinc/confluent-kafka-go/kafka

python

golang

pip install kafka-python

git clone https://github.com/edenhill/librdkafka.git

cd librdkafka

./configure --prefix=/usr

make && make install

profile:

export PKG_CONFIG_PATH=/usr/lib/pkgconfig

go get -u github.com/confluentinc/confluent-kafka-go/kafka

编程Producer 连接

Python 编程 Golang编程

复制代码

1
2
3
4
5
6
7
8
9
from kafka import KafkaProducer
import github.com/confluentinc/confluent-kafka-go/kafka
p = KafkaProducer(
p,err := kafka.NewProducer(&kafka.ConfigMap{
bootstrap_servers="master:9092"
"bootstrap.servers": "master"
)
})

Producer发送消息:

Python编程 Golang编程(异步发送)

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
msg = "send msg"
msg := "send msg"
topic = "job"
topic := "job"
p.send(topic, msg)
diverchan := make(chan kafka.Event)
p.Produce(&kafka.Message{
TopicPartition: kafka.TopicPartition{
Topic: &topic,
Partition:kafka.PartitionAny
},
Value: []byte(msg)
}, diverchan)
e := <- diverchan
et := e.(*kafka.Message)
if et.TopicPartition.Error == nil {
"send successfully"
}
close(diverchan)

Consumer编程:

Python编程 Golang编程

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from kafka import KafkaConsumer
c,er:=kafka.NewConsumer(&kafka.ConfigMap{
c = KafkaConsumer("job",
"bootstrap.servers": "master",
bootstrap_servers=["master:9092"]
"auto.offset.reset": "earliest"
)
})
for msg in c:
c.SubcribeTopics([]string{"job"},nil)
print msg
while true:
data := c.ReadMessage(-1)
print data
c.Close()