kafka

README

[outputs] plugin kafka

{
    "kafka": {
        "addrs": [
            "192.168.0.1:9092",
            "192.168.0.2:9092",
            "192.168.0.3:9092"
        ],
        "topic": "test",
        "max_procs": 100000
    }
}

Documentation

Index

• Constants
• type KafkaOutput
  • func NewKafkaOutput() *KafkaOutput
  • func (out *KafkaOutput) ChanInfo() string
  • func (out *KafkaOutput) Close() error
  • func (out *KafkaOutput) Init(config transport.Configer) error
  • func (out *KafkaOutput) Start() error
  • func (out *KafkaOutput) Version() string
  • func (out *KafkaOutput) Write(msg []byte) (int, error)
  • func (out *KafkaOutput) WriteToTopic() error

Constants

const (
	VERSION = "0.0.1"
)

Types

type KafkaOutput

type KafkaOutput struct {
	Addrs    []string `json:"addrs"`
	Topic    string   `json:"topic"`
	MaxProcs int      `json:"max_procs"` //最大并发写协程

	Message chan []byte //将数据写入这个管道中
	// contains filtered or unexported fields
}

func NewKafkaOutput

func NewKafkaOutput() *KafkaOutput

func (*KafkaOutput) ChanInfo

func (out *KafkaOutput) ChanInfo() string

func (*KafkaOutput) Close

func (out *KafkaOutput) Close() error

func (*KafkaOutput) Init

func (out *KafkaOutput) Init(config transport.Configer) error

func (*KafkaOutput) Start

func (out *KafkaOutput) Start() error

func (*KafkaOutput) Version

func (out *KafkaOutput) Version() string

func (*KafkaOutput) Write

func (out *KafkaOutput) Write(msg []byte) (int, error)

func (*KafkaOutput) WriteToTopic

func (out *KafkaOutput) WriteToTopic() error