batch

Documentation

Index

• Constants
• Variables
• func BufferLen(ctx context.Context, tableID int, tableName string) int
• func FailedCheck(ctx context.Context) error
• func HandleBatchFailed(ctx context.Context, name, params string, seq int32) error
• func Insert(ctx context.Context, tableInfo *obj.TblTable, subIntervals map[string]int)
• func InsertHandle(ctx context.Context) error
• func PushBatchInsert(ctx context.Context, nameSrv string, tableID int, table string, ...) (n int, err error)
• type BatchItems
  • func (s BatchItems) Len() int
  • func (s BatchItems) Less(i, j int) bool
  • func (s BatchItems) Swap(i, j int)
• type Cron
• type Filter
  • func (ft *Filter) Handle(ctx context.Context, req *filter.Request, resp *filter.Response, ...) error
• type InsertItem

Constants

const (
	FlushOpen  = 1 //开
	FlushClose = 0 //关
)

const (
	ReadPriority  = 0
	WritePriority = 1
)

const (
	PreBatchInsertBuff     = "BatchBuff"     //批量插入缓冲区
	PreBatchExchange       = "BatchEx"       //交换缓冲区
	PreBatchMutex          = "BatchMutex"    //缓冲区处理互斥
	PreBatchInsertFailBuff = "BatchFailBuff" //批量插入失败缓冲区
)

const (
	RetBatchInsert        = 101 // 批量插入异常
	RetBatchDataUnMarshal = 102 // 批量结果解压缩失败
	RetHasBatchFailed     = 103 // 有异常待处理批量插入记录
	RetBatchFailedHandle  = 104 // 批量插入异常处理失败
	RetFormatDataError    = 105 // 格式化数据 error
)

const BufferRedis = "buffer" //BufferRedis 批量插入缓冲区

const (
	InsertPopMax = 70000
)

Variables

var (
	WaitInsert       = map[int]map[string]bool{}
	WaitInsertRWLock = new(sync.RWMutex)
)

Functions

func BufferLen

func BufferLen(ctx context.Context, tableID int, tableName string) int

BufferLen 缓冲区数据个数

func FailedCheck

func FailedCheck(ctx context.Context) error

FailedCheck 检查批量插入失败数据

func HandleBatchFailed

func HandleBatchFailed(ctx context.Context, name, params string, seq int32) error

HandleBatchFailed 处理批量插入失败数据

func Insert

func Insert(ctx context.Context, tableInfo *obj.TblTable, subIntervals map[string]int)

Insert 批量插入

func InsertHandle

func InsertHandle(ctx context.Context) error

InsertHandle 批量插入处理

func PushBatchInsert

func PushBatchInsert(ctx context.Context, nameSrv string, tableID int, table string,
	data map[string]interface{}, datas []map[string]interface{}, dataType map[string]int8) (n int, err error)

PushBatchInsert 批量插入缓冲区

Types

type BatchItems

type BatchItems []*InsertItem

func (BatchItems) Len

func (s BatchItems) Len() int

func (BatchItems) Less

func (s BatchItems) Less(i, j int) bool

func (BatchItems) Swap

func (s BatchItems) Swap(i, j int)

type Cron

type Cron struct {
	Switch bool          //true-开 false-关
	Cron   *cron.Cron    //定时器
	Table  *obj.TblTable //表信息
}

type Filter

type Filter struct{}

Filter 表的唯一键生成插件

func (*Filter) Handle

func (ft *Filter) Handle(ctx context.Context, req *filter.Request, resp *filter.Response,
	dbConf obj.TblDB, tableConf obj.TblTable, config map[string]interface{}) error

type InsertItem

type InsertItem struct {
	Rand     int                      `json:"rand,omitempty"`
	Time     int64                    `json:"time,omitempty"`
	Table    string                   `json:"proto,omitempty"`
	Retry    int                      `json:"retry,omitempty"`
	Data     map[string]interface{}   `json:"data,omitempty"`
	Datas    []map[string]interface{} `json:"datas,omitempty"`
	DataType map[string]int8          `json:"data_type,omitempty"`
	Errors   map[int]string           `json:"errors,omitempty"`
}