README
¶
Kafka
Broker: Kafka
You can run examples by replace configs use your kafka instance configs.
Documentation
¶
Index ¶
- type Consumer
- func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error
- func (c *Consumer) Close() error
- func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
- func (c *Consumer) Setup(sarama.ConsumerGroupSession) error
- func (c *Consumer) Start()
- func (c *Consumer) Subscribe(handler ConsumerHandler)
- type ConsumerConfig
- type ConsumerHandler
- type ProducerConfig
- type SyncProducer
- type TraceInterceptor
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Consumer ¶
type Consumer struct {
// contains filtered or unexported fields
}
Consumer consumer struct
func NewConsumer ¶
func NewConsumer(config *ConsumerConfig) *Consumer
NewConsumer returns a consumer instance
func (*Consumer) Cleanup ¶
func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error
Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
func (*Consumer) ConsumeClaim ¶
func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
func (*Consumer) Setup ¶
func (c *Consumer) Setup(sarama.ConsumerGroupSession) error
Setup is run at the beginning of a new session, before ConsumeClaim
func (*Consumer) Subscribe ¶
func (c *Consumer) Subscribe(handler ConsumerHandler)
Subscribe subscribe to topic
type ConsumerConfig ¶
type ConsumerConfig struct {
Addr []string
Topic []string
Gid string
EnableSASLAuth bool
SASLMechanism string
SASLUser string
SASLPassword string
SASLHandshake bool
DialTimeout time.Duration
ConsumeOldest bool
EnableReturnError bool
ClientID string
}
ConsumerConfig kafka consumer config
type ConsumerHandler ¶
type ConsumerHandler func(context.Context, *sarama.ConsumerMessage) error
type ProducerConfig ¶
type ProducerConfig struct {
Addr []string
Topic []string
EnableSASLAuth bool
SASLMechanism string
SASLUser string
SASLPassword string
SASLHandshake bool
DialTimeout time.Duration
SlowSendDuration time.Duration
EnableReturnSuccess bool
ClientID string
}
ProducerConfig kafka producer config
type SyncProducer ¶
type SyncProducer struct {
// contains filtered or unexported fields
}
SyncProducer send message sync
func NewSyncProducer ¶
func NewSyncProducer(config *ProducerConfig) *SyncProducer
NewSyncProducer returns a SyncProducer instance
func (*SyncProducer) SendSyncMsg ¶
func (sp *SyncProducer) SendSyncMsg(ctx context.Context, content string) error
SendSyncMsg send message sync
type TraceInterceptor ¶
type TraceInterceptor struct {
TraceID string
}
func (*TraceInterceptor) OnSend ¶
func (ti *TraceInterceptor) OnSend(message *sarama.ProducerMessage)
Source Files
Directories
Click to show internal directories.
Click to hide internal directories.