spring_boot_kafka_producer

Documentation

Index

• type KafkaAsyncProducerStarter
  • func (k *KafkaAsyncProducerStarter) GetKafkaSendMsgChan() chan<- *sarama.ProducerMessage
  • func (k *KafkaAsyncProducerStarter) GetUserTopic() string
  • func (k *KafkaAsyncProducerStarter) OnStartApplication(ctx SpringBoot.ApplicationContext)
  • func (k *KafkaAsyncProducerStarter) OnStopApplication(ctx SpringBoot.ApplicationContext)
  • func (k *KafkaAsyncProducerStarter) SetErrorHandle(handle func(*sarama.ProducerError))
• type KafkaProducerConfig

Types

type KafkaAsyncProducerStarter

type KafkaAsyncProducerStarter struct {
	KafkaAsyncProducerConfig *KafkaProducerConfig `autowire:""`
	// contains filtered or unexported fields
}

func (*KafkaAsyncProducerStarter) GetKafkaSendMsgChan

func (k *KafkaAsyncProducerStarter) GetKafkaSendMsgChan() chan<- *sarama.ProducerMessage

func (*KafkaAsyncProducerStarter) GetUserTopic

func (k *KafkaAsyncProducerStarter) GetUserTopic() string

func (*KafkaAsyncProducerStarter) OnStartApplication

func (k *KafkaAsyncProducerStarter) OnStartApplication(ctx SpringBoot.ApplicationContext)

func (*KafkaAsyncProducerStarter) OnStopApplication

func (k *KafkaAsyncProducerStarter) OnStopApplication(ctx SpringBoot.ApplicationContext)

func (*KafkaAsyncProducerStarter) SetErrorHandle

func (k *KafkaAsyncProducerStarter) SetErrorHandle(handle func(*sarama.ProducerError))

type KafkaProducerConfig

type KafkaProducerConfig struct {
	ConnectTimeOut    int      `value:"${kafka.socket_connect_timeout:=30}"`
	ReadTimeOut       int      `value:"${kafka.socket_read_timeout:=30}"`
	WriteTimeOut      int      `value:"${kafka.socket_write_timeout:=30}"`
	Brokers           []string `value:"${kafka.brokers:=127.0.0.1:9092}"`
	Version           string   `value:"${kafka.version:=2.4.0}"`
	ChannelBufferSize int      `value:"${kafka.message_cache:=256}"`
	ClientId          string   `value:"${kafka.client_id:=kafka_client}"`
	IsSync            bool     `value:"${kafka.producer.is_sync:=false}"`
	AcksLevel         string   `value:"${kafka.producer.acks_level:=wait_for_local}"`
	FlushMessages     uint32   `value:"${kafka.producer.flush_messages:=10}"`
	FlushFrequency    uint32   `value:"${kafka.producer.flush_frequency_ms:=1000}"`
	FlushMaxMessages  uint32   `value:"${kafka.producer.flush_max_messages:=100}"`
	UserTopic         string   `value:"${kafka.producer.user_topic:=}"`
}