kafka

Documentation

Index

• Constants
• func RegisterKafkaConfig(config *KafkaConfig)
• type KafkaConfig
• type KafkaConsumer
  • func GetConsumer(key string) *KafkaConsumer
• type KafkaProducer
  • func GetProducer(key string) *KafkaProducer
• type SaslConfig
• type TlsConfig

Constants

const (
	SaslTypePlain = "PLAIN"
	SaslTypeScram = "SCRAM"

	SaslScramAlgorithmSha256 = "SCRAM-SHA-256"
	SaslScramAlgorithmSha512 = "SCRAM-SHA-512"
)

Functions

func RegisterKafkaConfig

func RegisterKafkaConfig(config *KafkaConfig)

Types

type KafkaConfig

type KafkaConfig struct {
	Producer map[string]*KafkaProducer
	Consumer map[string]*KafkaConsumer
}

type KafkaConsumer

type KafkaConsumer struct {
	Key       string   `json:"key"`
	Address   []string `json:"address"` // kafka地址
	Group     string   `json:"group"`   // groupId
	Topic     string   `json:"topic"`
	Partition int      `json:"partition"`
	Offset    int64    `json:"offset"`

	SaslConfig *SaslConfig
	TlsConfig  *TlsConfig
	// contains filtered or unexported fields
}

func GetConsumer

func GetConsumer(key string) *KafkaConsumer

type KafkaProducer

type KafkaProducer struct {
	Address       []string
	Topic         string
	Async         bool
	Ack           int
	CompressCodec string

	SaslConfig *SaslConfig
	TlsConfig  *TlsConfig
	// contains filtered or unexported fields
}

func GetProducer

func GetProducer(key string) *KafkaProducer

type SaslConfig

type SaslConfig struct {
	SaslType      string `yaml:"sasl_type"`
	SaslUsername  string `yaml:"sasl_username"`
	SaslPassword  string `yaml:"sasl_password"`
	SaslScramAlgo string `yaml:"sasl_scram_algo"`
}

type TlsConfig

type TlsConfig struct {
	TlsEnable          bool   `yaml:"tls_enable"`
	CaFile             string `yaml:"ca_file"`
	CertFile           string `yaml:"cert_file"`
	KeyFile            string `yaml:"key_file"`
	InsecureSkipVerify bool   `yaml:"insecure_skip_verify"`
}