job

Documentation

Overview

Package job includes mainly 2 interface: Processor & Collector.

Processor is supposed to receive a single string (refered to as a "msg")
as input, then it outputs one or more semi-finished results(refered to
as a "item") via channels to Collector.
Collector receive the items and flush them out when appropriate or do
anything you like.

Index

• Constants
• func BKDRHash(str string) int
• func FilePutContentAppend(file string, content string) error
• func GetLeastBusyChannel(channels [](chan Item)) int
• func GetShortHostName() (string, error)
• type Collector
  • func NewCollector(name string) Collector
• type DocIdInfo
• type Item
• type LogInfo
• type LogtoHdfsCollector
  • func (p *LogtoHdfsCollector) Collect(item Item) error
  • func (p *LogtoHdfsCollector) Destory() error
  • func (p *LogtoHdfsCollector) HandleZip(path string) (string, error)
  • func (p *LogtoHdfsCollector) Init(ctx *sj.Json, id int) error
  • func (p *LogtoHdfsCollector) Tick() error
• type LogtoHdfsProcessor
  • func (p *LogtoHdfsProcessor) Destory() error
  • func (p *LogtoHdfsProcessor) Init(ctx *sj.Json, id int, itemChans [](chan Item)) error
  • func (p *LogtoHdfsProcessor) Process(msg string) error
  • func (p *LogtoHdfsProcessor) Tick() error
• type MessageQueue
  • func NewMessageQueue(c redis.Conn) *MessageQueue
  • func (mq *MessageQueue) AddQueue(name, value string) error
  • func (mq *MessageQueue) GetQueue(name string) (string, error)
• type Processor
  • func NewProcessor(name string) Processor
• type SLMap
• type StrSlice
• type TestCollector
  • func (p *TestCollector) Collect(item Item) error
  • func (p *TestCollector) Destory() error
  • func (p *TestCollector) Init(ctx *sj.Json, id int) error
  • func (p *TestCollector) Tick() error
• type TestProcessor
  • func (p *TestProcessor) Destory() error
  • func (p *TestProcessor) Init(ctx *sj.Json, id int, itemChans [](chan Item)) error
  • func (p *TestProcessor) Process(msg string) error
  • func (p *TestProcessor) Tick() error
• type TimeSlice
  • func (p TimeSlice) Len() int
  • func (p TimeSlice) Less(i, j int) bool
  • func (p TimeSlice) Swap(i, j int)

Constants

const TimeFormatHour = "2006-01-02-15"

const TimeFormatMinute = "2006-01-02-15-04"

Functions

func BKDRHash

func BKDRHash(str string) int

func FilePutContentAppend

func FilePutContentAppend(file string, content string) error

func GetLeastBusyChannel

func GetLeastBusyChannel(channels [](chan Item)) int

得到len()最小的channel，做负载均衡使用

func GetShortHostName

func GetShortHostName() (string, error)

Types

type Collector

type Collector interface {
	Init(ctx *sj.Json, id int) error
	Collect(item Item) error
	Tick() error
	Destory() error
}

func NewCollector

func NewCollector(name string) Collector

type DocIdInfo

type DocIdInfo struct {
	Count      int    `json:"count"`
	Errmsg     string `json:"errmsg"`
	Errno      int    `json:"errno"`
	StartIndex int    `json:"start_index"`
	Time       int    `json:"time"`
}

type Item

type Item struct {
	Category string
	Id       string
	Content  string
	RawMsg   string
}

type LogInfo

type LogInfo struct {
	// contains filtered or unexported fields
}

type LogtoHdfsCollector

type LogtoHdfsCollector struct {
	// contains filtered or unexported fields
}

func (*LogtoHdfsCollector) Collect

func (p *LogtoHdfsCollector) Collect(item Item) error

func (*LogtoHdfsCollector) Destory

func (p *LogtoHdfsCollector) Destory() error

func (*LogtoHdfsCollector) HandleZip

func (p *LogtoHdfsCollector) HandleZip(path string) (string, error)

func (*LogtoHdfsCollector) Init

func (p *LogtoHdfsCollector) Init(ctx *sj.Json, id int) error

func (*LogtoHdfsCollector) Tick

func (p *LogtoHdfsCollector) Tick() error

type LogtoHdfsProcessor

type LogtoHdfsProcessor struct {
	// contains filtered or unexported fields
}

receive local paths, and logto hadoop

func (*LogtoHdfsProcessor) Destory

func (p *LogtoHdfsProcessor) Destory() error

func (*LogtoHdfsProcessor) Init

func (p *LogtoHdfsProcessor) Init(ctx *sj.Json, id int, itemChans [](chan Item)) error

func (*LogtoHdfsProcessor) Process

func (p *LogtoHdfsProcessor) Process(msg string) error

func (*LogtoHdfsProcessor) Tick

func (p *LogtoHdfsProcessor) Tick() error

type MessageQueue

type MessageQueue struct {
	// contains filtered or unexported fields
}

message queue

func NewMessageQueue

func NewMessageQueue(c redis.Conn) *MessageQueue

@todo init redis

func (*MessageQueue) AddQueue

func (mq *MessageQueue) AddQueue(name, value string) error

add queue

func (*MessageQueue) GetQueue

func (mq *MessageQueue) GetQueue(name string) (string, error)

get queue

type Processor

type Processor interface {
	Init(ctx *sj.Json, id int, itemChans [](chan Item)) error
	Process(msg string) error
	Tick() error
	Destory() error
}

func NewProcessor

func NewProcessor(name string) Processor

type SLMap

type SLMap map[string]LogInfo

type StrSlice

type StrSlice []string

type TestCollector

type TestCollector struct {
	// contains filtered or unexported fields
}

func (*TestCollector) Collect

func (p *TestCollector) Collect(item Item) error

func (*TestCollector) Destory

func (p *TestCollector) Destory() error

func (*TestCollector) Init

func (p *TestCollector) Init(ctx *sj.Json, id int) error

func (*TestCollector) Tick

func (p *TestCollector) Tick() error

type TestProcessor

type TestProcessor struct {
	// contains filtered or unexported fields
}

func (*TestProcessor) Destory

func (p *TestProcessor) Destory() error

func (*TestProcessor) Init

func (p *TestProcessor) Init(ctx *sj.Json, id int, itemChans [](chan Item)) error

func (*TestProcessor) Process

func (p *TestProcessor) Process(msg string) error

func (*TestProcessor) Tick

func (p *TestProcessor) Tick() error

type TimeSlice

type TimeSlice []time.Time

func (TimeSlice) Len

func (p TimeSlice) Len() int

func (TimeSlice) Less

func (p TimeSlice) Less(i, j int) bool

func (TimeSlice) Swap

func (p TimeSlice) Swap(i, j int)