kafka

Documentation

Index

• Constants
• type NodeConfiguration
• type ProducerNode
  • func (x *ProducerNode) Destroy()
  • func (x *ProducerNode) Init(ruleConfig types.Config, configuration types.Configuration) error
  • func (x *ProducerNode) New() types.Node
  • func (x *ProducerNode) OnMsg(ctx types.RuleContext, msg types.RuleMsg)
  • func (x *ProducerNode) Type() string

Constants

const (
	KeyPartition = "partition"
	KeOffset     = "offset"
)

Types

type NodeConfiguration

type NodeConfiguration struct {
	// kafka服务器地址列表，多个与逗号隔开
	Server string
	// Topic 发布主题，可以使用 ${metadata.key} 读取元数据中的变量或者使用 ${msg.key} 读取消息负荷中的变量进行替换
	Topic string
	// Key 分区键，可以使用 ${metadata.key} 读取元数据中的变量或者使用 ${msg.key} 读取消息负荷中的变量进行替换
	Key string
	//Partition 分区编号
	Partition int32
}

NodeConfiguration 节点配置

type ProducerNode

type ProducerNode struct {
	base.SharedNode[sarama.SyncProducer]
	Config NodeConfiguration
	// contains filtered or unexported fields
}

func (*ProducerNode) Destroy

func (x *ProducerNode) Destroy()

Destroy 销毁组件

func (*ProducerNode) Init

func (x *ProducerNode) Init(ruleConfig types.Config, configuration types.Configuration) error

Init 初始化组件

func (*ProducerNode) New

func (x *ProducerNode) New() types.Node

func (*ProducerNode) OnMsg

func (x *ProducerNode) OnMsg(ctx types.RuleContext, msg types.RuleMsg)

OnMsg 处理消息

func (*ProducerNode) Type

func (x *ProducerNode) Type() string

Type 返回组件类型