kafka

Documentation

Overview

Package kafka Kafka 传输层实现

Kafka 使用基于偏移量的消息确认机制

Package kafka Kafka 传输层实现

Kafka 使用基于偏移量的消息确认机制

Index

• func New(_ context.Context, cfg Config) (transport.Transport[Channel], error)
• type Channel
• type Config
• type EventBuilder
  • func NewEventBuilder[T any](ce *cloudevent.CloudEvent[T], cd codec.Codec, ...) *EventBuilder[T]
  • func (b *EventBuilder[T]) Send(ctx context.Context) error
  • func (b *EventBuilder[T]) WithExtensions(ext cloudevent.Extensions) *EventBuilder[T]
  • func (b *EventBuilder[T]) WithSubject(subject string) *EventBuilder[T]
• type ReceivedEvent

Functions

func New

func New(_ context.Context, cfg Config) (transport.Transport[Channel], error)

New 创建 Kafka Transport

Types

type Channel

type Channel = eventapiv1.OperationRule_Channel

Channel 是 IDL 生成的 Channel 类型别名

type Config

type Config struct {
	// Brokers Kafka broker 地址列表
	Brokers []string

	// ClientID 客户端标识
	ClientID string

	// Timeout 连接超时时间
	Timeout time.Duration

	// Async 是否启用异步发送
	Async bool

	// BatchSize 批量发送的消息数量
	BatchSize int

	// BatchBytes 批量大小的最大字节数
	BatchBytes int

	// BatchTimeout 刷新批量消息前的等待时间
	BatchTimeout time.Duration

	// GroupID 消费者组 ID
	// 如果为空且调用 OnReceive，会返回错误
	GroupID string
}

Config Kafka 传输层配置

type EventBuilder

type EventBuilder[T any] struct {
	// contains filtered or unexported fields
}

EventBuilder Kafka 事件构建器

func NewEventBuilder

func NewEventBuilder[T any](
	ce *cloudevent.CloudEvent[T],
	cd codec.Codec,
	sendFn func(ctx context.Context, event *cloudevent.CloudEvent[[]byte]) error,
) *EventBuilder[T]

NewEventBuilder 创建事件构建器

func (*EventBuilder[T]) Send

func (b *EventBuilder[T]) Send(ctx context.Context) error

Send 发送事件

func (*EventBuilder[T]) WithExtensions

func (b *EventBuilder[T]) WithExtensions(ext cloudevent.Extensions) *EventBuilder[T]

WithExtensions 设置扩展属性

func (*EventBuilder[T]) WithSubject

func (b *EventBuilder[T]) WithSubject(subject string) *EventBuilder[T]

WithSubject 设置 CloudEvent subject

type ReceivedEvent

type ReceivedEvent[T any] struct {
	Event     *cloudevent.CloudEvent[T]   // 接收到的 CloudEvent
	Commit    func(context.Context) error // 提交偏移量
	Partition int32                       // 所在分区
	Offset    int64                       // 消息偏移量
}

ReceivedEvent Kafka 接收事件