bingyi/gf
1
0
Fork 0
mirror of https://gitee.com/johng/gf synced 2026-06-07 02:12:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

改进gkafka，获取topics使用临时的消费对象

Browse Source
This commit is contained in:
john
2018-10-23 18:13:29 +08:00
parent 1028ffa49f
commit 78cc8533a5
1 changed files with 14 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
g/database/gkafka/gkafka.go
View File

@ -96,23 +96,22 @@ func (client *Client) Close() {
}
// Get all topics from kafka server.
// 这里创建独立的消费客户端获取topics，获取完之后销毁该客户端对象。
func (client *Client) Topics() ([]string, error) {
if client.rawConsumer == nil {
if c, err := sarama.NewConsumer(strings.Split(client.Config.Servers, ","), &client.Config.Config); err != nil {
return nil, err
} else {
client.rawConsumer = c
}
}
if topics, err := client.rawConsumer.Topics(); err == nil {
for k, v := range topics {
if _, ok := ignoreTopics[v]; ok {
topics = append(topics[ : k], topics[k + 1 : ]...)
}
}
return topics, nil
} else {
if c, err := sarama.NewConsumer(strings.Split(client.Config.Servers, ","), &client.Config.Config); err != nil {
return nil, err
} else {
if topics, err := c.Topics(); err == nil {
for k, v := range topics {
if _, ok := ignoreTopics[v]; ok {
topics = append(topics[ : k], topics[k + 1 : ]...)
}
}
c.Close()
return topics, nil
} else {
return nil, err
}
}
}