You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

90 lines
3.4 KiB

package kafka
import (
"fmt"
"time"
"github.com/IBM/sarama"
)
var (
//brokers = flag.String("brokers", "150.158.237.39:50004", "The Kafka brokers to connect to, as a comma separated list")
// userName = flag.String("username", "admin", "The SASL username")
// passwd = flag.String("passwd", "root", "The SASL password")
// algorithm = flag.String("algorithm", "", "The SASL SCRAM SHA algorithm sha256 or sha512 as mechanism")
// topic = flag.String("topic", "test", "The Kafka topic to use")
// certFile = flag.String("certificate", "", "The optional certificate file for client authentication")
// keyFile = flag.String("key", "", "The optional key file for client authentication")
// caFile = flag.String("ca", "", "The optional certificate authority file for TLS client authentication")
// tlsSkipVerify = flag.Bool("tls-skip-verify", false, "Whether to skip TLS server cert verification")
// useTLS = flag.Bool("tls", false, "Use TLS to communicate with the cluster")
// mode = flag.String("mode", "produce", "Mode to run in: \"produce\" to produce, \"consume\" to consume")
// logMsg = flag.Bool("logmsg", false, "True to log consumed messages to console")
// logger = log.New(os.Stdout, "[Producer] ", log.LstdFlags)
)
type msgConsumerGroup struct{}
func (m msgConsumerGroup) Setup(_ sarama.ConsumerGroupSession) error { return nil }
func (m msgConsumerGroup) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (m msgConsumerGroup) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
fmt.Printf("Message topic:%q partition:%d offset:%d value:%s\n", msg.Topic, msg.Partition, msg.Offset, string(msg.Value))
// 其他数据落库操作
// 标记,sarama会自动进行提交,默认间隔1秒
sess.MarkMessage(msg, "")
}
return nil
}
var ConsumerGroup msgConsumerGroup
func InitSynProducer(brokers []string) sarama.SyncProducer {
conf := sarama.NewConfig()
conf.Consumer.Offsets.AutoCommit.Enable = true
conf.Consumer.Offsets.AutoCommit.Interval = time.Second * 1
conf.Producer.Retry.Max = 1
conf.Producer.RequiredAcks = sarama.WaitForAll
conf.Producer.Return.Successes = true
conf.Metadata.Full = true
conf.Version = sarama.V0_10_2_0
// conf.ClientID = "sasl_scram_client"
conf.Metadata.Full = true
// conf.Net.SASL.Enable = true
// conf.Net.SASL.User = *userName
// conf.Net.SASL.Password = *passwd
// conf.Net.SASL.Handshake = true
// 使用同步producer,异步模式下有更高的性能,但是处理更复杂,这里建议先从简单的入手
producer, err := sarama.NewSyncProducer(brokers, conf)
if err != nil {
panic(err.Error())
}
return producer
}
func InitConsumerGroup(brokers []string, groupId string) sarama.ConsumerGroup {
conf := sarama.NewConfig()
conf.Consumer.Offsets.AutoCommit.Enable = true
conf.Consumer.Offsets.AutoCommit.Interval = time.Second * 1
conf.Producer.Retry.Max = 1
conf.Producer.RequiredAcks = sarama.WaitForAll
conf.Producer.Return.Successes = true
conf.Metadata.Full = true
conf.Version = sarama.V0_10_2_0
// conf.ClientID = "sasl_scram_client"
conf.Metadata.Full = true
// conf.Net.SASL.Enable = true
// conf.Net.SASL.User = *userName
// conf.Net.SASL.Password = *passwd
// conf.Net.SASL.Handshake = true
cGroup, err := sarama.NewConsumerGroup(brokers, groupId, conf)
if err != nil {
panic(err)
}
return cGroup
}