pipe

README

pipe

用于串行化处理，基本原理是启动任务处理的go routine来处理被投递到队列中的任务。 如果需要多路go routine处理，则在任务分发时通过计算slot，将不同的请求分发到不同slot对应的go routine上执行。

mq

生产者消费者队列，生产者在往队列中投递了消息后，消费者会得到通知并进行处理。 此队列支持投递控制消息和普通消息，控制消息优先级高于普通消息；同时，对于控制消息和普通消息都支持分优先级， 可以通过调用高优先级的函数来让投递的消息优先级更高。

q

mq的简化版，mq是早期设计的生产者消费者队列，在使用过程中发现控制消息没有实际的用处，q这个包去掉了mq中控制消息 相关功能。

line

生产者/消费者模式的go routine控制，其中投递的消息是执行的函数本身，由于golang没有像C那样直接可以做各个指针类型转化的功能。 函数签名的参数使用了interface{}，使用起来不算特别方便，可以使用syncx/semap替代相关功能。

mline

在line包的基础上，对line进行了多路并发的封装，与line的问题一样， 函数签名的参数使用了interface{}，使用起来不算特别方便，可以使用syncx/semap替代相关功能。

mux

用多路go routine来封装多路生产者/消费者模式中的消费者，封装了针对数据CRUD的操作， 函数签名的参数使用了interface{}，使用起来不算特别方便，可以使用syncx/semap替代相关功能。

Documentation

Index

• Constants
• Variables
• func ConvertQueueErr(err error) error
• func GetOption(opts ...Option) (slotSize int, qSize int)
• func NormalizeSlotIndex(index int, slotSize int) int
• type Option
  • func WithQSize(qSize int) Option
  • func WithSlotSize(slotSize int) Option

Constants

const (
	//DefaultSlotSize slot size
	//素数
	DefaultSlotSize = 509
	//DefaultQSize default pipe size, total current request can be pipe is 509*1024*8 = 4169728
	DefaultQSize = 1024 * 8
)

Variables

var (
	//ErrNoHandler -- no message handler
	ErrNoHandler = status.Error(codes.Unimplemented, "no.req.handler")
	//ErrQueueFull -- msg queue is full
	ErrQueueFull = status.Error(codes.ResourceExhausted, "req.queue.full")
	//ErrQueueClosed -- msg queue closed
	ErrQueueClosed = status.Error(codes.Unavailable, "req.queue.closed")
	//ErrInvalidParam -- invalid msg param
	ErrInvalidParam = status.Error(codes.InvalidArgument, "invalid.req.param")
	//ErrInvalidRsp -- invalid result
	ErrInvalidRsp = status.Error(codes.Internal, "invalid.rsp.msg")
)

Functions

func ConvertQueueErr

func ConvertQueueErr(err error) error

ConvertQueueErr convert msg queue error

func GetOption

func GetOption(opts ...Option) (slotSize int, qSize int)

GetOption : return slot size and q size

func NormalizeSlotIndex

func NormalizeSlotIndex(index int, slotSize int) int

NormalizeSlotIndex slot index

Types

type Option

type Option func(o *_Option)

Option shunt option function

func WithQSize

func WithQSize(qSize int) Option

WithQSize setup queue size in each slot

func WithSlotSize

func WithSlotSize(slotSize int) Option

WithSlotSize setup slot size