go get -u github.com/maczh/mgkafka在MGin微服务模块的main.go中,在mgin.Init()之后，加入一行
//加载Kafka消息队列
mgin.MGin.Use("kafka",mgkafka.Kafka.Init,mgkafka.Kafka.Close,mgkafka.Kafka.Check)go:
config:
used: kafka
prefix:
kafka: kafkago:
data:
kafka:
servers: "59.56.77.23:9092" #集群多个服务器之间用逗号分隔
ack: all #ack模式 no,local,all
auto_commit: true #是否自动提交
partitioner: hash #分区选择模式 hash,random,round-robin
version: 2.8.1 #kafka版本 mgkafka.Kafka.Send("my_topic", "测试消息")- 定义消息处理函数
func handleMsg(msg string) error {
logs.Debug("收到Kafka消息:{}",msg)
return nil
}- 在main.go中添加侦听代码
//侦听kafka消息，说明，一个topic对应一个groupId
err := mgkafka.Kafka.MessageListener("my_group_id","my_topic",handleMsg)
if err != nil {
logs.Error("侦听kafka消息失败")
}
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent