mline

Documentation

Index

• type AsyncCtx
  • func (m *AsyncCtx) R() (interface{}, error)
  • func (m *AsyncCtx) SetR(r interface{}, err error)
• type AsyncR
• type CallCtx
  • func NewCallCtx(hashIndex int, call CallFn, param interface{}) *CallCtx
• type CallFn
• type MultiLine
  • func NewMultiLine(opts ...pipe.Option) *MultiLine
  • func (c *MultiLine) AsyncCall(ctx context.Context, callCtx *CallCtx) (interface{}, error)
  • func (c *MultiLine) IndexOf(i int) int
  • func (c *MultiLine) QSize() int
  • func (c *MultiLine) Run()
  • func (c *MultiLine) SlotSize() int
  • func (c *MultiLine) Stop()
  • func (c *MultiLine) WaitStop(ctx context.Context) error

Types

type AsyncCtx

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

AsyncCtx : async call context

func (*AsyncCtx) R

func (m *AsyncCtx) R() (interface{}, error)

R : get response with wait

func (*AsyncCtx) SetR

func (m *AsyncCtx) SetR(r interface{}, err error)

SetR : set return

type AsyncR

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

AsyncR : async call result.

type CallCtx

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

CallCtx : call context, function and param

func NewCallCtx

func NewCallCtx(hashIndex int, call CallFn, param interface{}) *CallCtx

NewCallCtx ： new call context hashIndex : 一个与请求相关的散列值，例如如果用user_id作为散列分组，可以直接传user_id，如果是一个字符串，可以将此字符串CRC32散列一下。 传入的hashIndex应该是稳定的散列方式，例如在请求A中使用了CRC32作为散列方式，那所以相关的请求都应该用CRC32， 断不可在别的请求中使用类似xxhash这样别的散列方式。 call : 回调函数 param : 函数参数

type CallFn

type CallFn func(ctx context.Context, sIndex int, req interface{}) (rsp interface{}, err error)

CallFn : call function，回调函数 Input: ctx -- context sIndex -- 表示在处理的go routine数组中对应的index，如果每个go routine有独立的缓存/内存，那此index可以用于定位相关的内存/缓存。 param -- call param Output: r -- 回调函数调用后的返回值 err - 回调函数调用失败后返回error

type MultiLine

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

MultiLine : multi-queue handler

func NewMultiLine

func NewMultiLine(opts ...pipe.Option) *MultiLine

NewMultiLine : new multi-queue group

func (*MultiLine) AsyncCall

func (c *MultiLine) AsyncCall(ctx context.Context, callCtx *CallCtx) (interface{}, error)

AsyncCall : wrap call ctx -- context.Context callCtx -- call context

func (*MultiLine) IndexOf

func (c *MultiLine) IndexOf(i int) int

IndexOf : get slot index. 将散列值映射成处理数组的index，举例来说，如果以user id作为散列值，则整个处理逻辑会用user id的绝对值对处理数组长度取模，取模后的值就是 其在数组中的位置。 Input : i -- a slot key number. 此参数就是分片使用的hash值。 Output : index of slot.返回此hash值在处理数组中对应的位置。

func (*MultiLine) QSize

func (c *MultiLine) QSize() int

QSize : get queue size in each slot

func (*MultiLine) Run

func (c *MultiLine) Run()

Run : run all queue msg handler

func (*MultiLine) SlotSize

func (c *MultiLine) SlotSize() int

SlotSize : get slot size

func (*MultiLine) Stop

func (c *MultiLine) Stop()

Stop : stop

func (*MultiLine) WaitStop

func (c *MultiLine) WaitStop(ctx context.Context) error

WaitStop : wait stop