queue

Documentation

Index

• Constants
• func AtomicRename(sourceFile, targetFile string) error
• type BackendQueue
  • func NewDirectQueue(name string) BackendQueue
  • func NewDiskQueue(opts NewDiskQueueOptions) BackendQueue
• type DataQueue
• type LinesQueue
• type NewDiskQueueOptions

Constants

const (
	StatusInit int32 = iota
	StatusClosed
)

const (
	FromNone = iota
	FromDisk
	FromMemory
)

Functions

func AtomicRename

func AtomicRename(sourceFile, targetFile string) error

Types

type BackendQueue

type BackendQueue interface {
	Name() string
	Put([]byte) error
	ReadChan() <-chan []byte // this is expected to be an *unbuffered* channel
	Close() error
	Delete() error
	Depth() int64
	Empty() error
}

BackendQueue represents the behavior for the secondary message storage system

func NewDirectQueue

func NewDirectQueue(name string) BackendQueue

func NewDiskQueue

func NewDiskQueue(opts NewDiskQueueOptions) BackendQueue

newDiskQueue instantiates a new instance of diskQueue, retrieving metadata from the filesystem and starting the read ahead goroutine

type DataQueue

type DataQueue interface {
	// PutDatas 用于存放一组数据
	PutDatas([]Data) error
	// ReadDatasChan 用于获取直接读取 Data 的管道
	ReadDatasChan() <-chan []Data
}

DataQueue 代表了无需编解码可直接放取 Data 的队列

type LinesQueue

type LinesQueue interface {
	// PutDatas 用于存放一组数据
	PutLines([]string) error
	// ReadLinesChan 用于获取直接读取 []line 的管道
	ReadLinesChan() <-chan []string
}

LinesQueue 代表了无需编解码可直接放取 []Line 的队列

type NewDiskQueueOptions

type NewDiskQueueOptions struct {
	Name              string
	DataPath          string
	MaxBytesPerFile   int64
	MinMsgSize        int32
	MaxMsgSize        int32
	SyncEveryWrite    int64
	SyncEveryRead     int64
	SyncTimeout       time.Duration
	WriteRateLimit    int
	EnableMemoryQueue bool
	MemoryQueueSize   int64

	// DisableDiskUsedLimit 指示是否禁用磁盘占用限制，超出限制后数据将不会再写入到文件而被直接丢弃
	DisableDiskUsedLimit bool
	MaxDiskUsedBytes     int64
}