53 lines
1.5 KiB
Go
53 lines
1.5 KiB
Go
|
package kafka
|
|||
|
|
|
|||
|
|
import (
|
|||
|
|
"fmt"
|
|||
|
|
"github.com/1340691923/xwl_bi/model"
|
|||
|
|
"github.com/Shopify/sarama"
|
|||
|
|
"log"
|
|||
|
|
"time"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
//初始化kafka数据
|
|||
|
|
func Init() {
|
|||
|
|
config := sarama.NewConfig()
|
|||
|
|
|
|||
|
|
config.Version = sarama.V2_0_0_0
|
|||
|
|
if model.GlobConfig.Comm.Kafka.Username != "" {
|
|||
|
|
config.Net.SASL.Enable = true
|
|||
|
|
config.Net.SASL.User = model.GlobConfig.Comm.Kafka.Username
|
|||
|
|
config.Net.SASL.Password = model.GlobConfig.Comm.Kafka.Password
|
|||
|
|
config.Net.SASL.Handshake = true
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
config.Consumer.Group.Session.Timeout = 15 * time.Second
|
|||
|
|
config.Consumer.Group.Heartbeat.Interval = 5 * time.Second
|
|||
|
|
|
|||
|
|
conn, err := sarama.NewClusterAdmin(model.GlobConfig.Comm.Kafka.Addresses, config)
|
|||
|
|
if err != nil {
|
|||
|
|
log.Println(fmt.Sprintf("kafka 链接初始化失败:%s", err.Error()))
|
|||
|
|
panic(err)
|
|||
|
|
}
|
|||
|
|
s, err := conn.ListTopics()
|
|||
|
|
for topic := range s {
|
|||
|
|
log.Println("您所拥有的TOPIC为：", topic)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
if _, ok := s[model.GlobConfig.Comm.Kafka.ReportTopicName]; !ok {
|
|||
|
|
detail := sarama.TopicDetail{NumPartitions: model.GlobConfig.Comm.Kafka.NumPartitions, ReplicationFactor: 1}
|
|||
|
|
err = conn.CreateTopic(model.GlobConfig.Comm.Kafka.ReportTopicName, &detail, false)
|
|||
|
|
if err != nil {
|
|||
|
|
log.Println("创建TOPIC失败！", model.GlobConfig.Comm.Kafka.ReportTopicName)
|
|||
|
|
panic(err)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
err = conn.Close()
|
|||
|
|
if err != nil {
|
|||
|
|
panic(err)
|
|||
|
|
}
|
|||
|
|
log.Println("初始化TOPIC完成！", model.GlobConfig.Comm.Kafka.ReportTopicName)
|
|||
|
|
} else {
|
|||
|
|
log.Println("您已拥有该TOPIC：", model.GlobConfig.Comm.Kafka.ReportTopicName)
|
|||
|
|
}
|
|||
|
|
}
|