mirror/awesome-patterns
1
0
Fork 0
mirror of https://github.com/crazybber/awesome-patterns.git synced 2024-11-25 14:26:04 +03:00
Code Issues Packages Projects Releases Wiki Activity

[fanout] Pipeline will maintain its broker

Browse Source
This commit is contained in:
paulaan 2019-05-01 18:14:30 +07:00
parent 26549a1e88
commit 42e2301e29
1 changed files with 11 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
messaging/fanout.go
View File

@ -14,13 +14,13 @@ const (
) )
var ( var (
running uint32 = 0
log, _ = zap.NewDevelopment() log, _ = zap.NewDevelopment()
) )
type Pipeline struct { type Pipeline struct {
workers []*worker workers []*worker
chain chan interface{} chain chan interface{}
running *uint32
} }
func (p *Pipeline) Start() { func (p *Pipeline) Start() {
@ -29,7 +29,7 @@ func (p *Pipeline) Start() {
for { for {
for _, c := range cs { for _, c := range cs {
expectationWorkers := uint32(len(ch)/(MaxQueueSize/MaxWorkers)) + 1 expectationWorkers := uint32(len(ch)/(MaxQueueSize/MaxWorkers)) + 1
runningWorker := atomic.LoadUint32(&running) runningWorker := atomic.LoadUint32(p.running)
if c.index > runningWorker+1 || c.index > expectationWorkers { if c.index > runningWorker+1 || c.index > expectationWorkers {
break break
} }
@ -42,11 +42,11 @@ func (p *Pipeline) Start() {
go writer.stream(val) go writer.stream(val)
} }
} }
}
} }
} }
}
go distributeToChannels(p.chain, p.workers) go distributeToChannels(p.chain, p.workers)
} }
func (p *Pipeline) Dispatch(msg interface{}) { func (p *Pipeline) Dispatch(msg interface{}) {
@ -58,6 +58,7 @@ type DispatcherBuilder func() Dispatcher
func NewPipeline(d DispatcherBuilder, idle uint32, debug bool) *Pipeline { func NewPipeline(d DispatcherBuilder, idle uint32, debug bool) *Pipeline {
wk := make([]*worker, 0, MaxWorkers) wk := make([]*worker, 0, MaxWorkers)
ch := make(chan interface{}, 4096) ch := make(chan interface{}, 4096)
pipe := &Pipeline{chain: ch, running: new(uint32)}
for i := 0; i < MaxWorkers; i++ { for i := 0; i < MaxWorkers; i++ {
wk = append(wk, wk = append(wk,
&worker{ &worker{
@ -67,9 +68,11 @@ func NewPipeline(d DispatcherBuilder, idle uint32, debug bool) *Pipeline {
debug: debug, debug: debug,
idle: idle, idle: idle,
Dispatcher: d(), Dispatcher: d(),
broker: pipe.running,
}) })
} }
return &Pipeline{workers: wk, chain: ch} pipe.workers = wk
return pipe
} }
type Dispatcher interface { type Dispatcher interface {
@ -85,6 +88,7 @@ type worker struct {
chain chan interface{} chain chan interface{}
debug bool debug bool
idle uint32 idle uint32
broker *uint32
Dispatcher Dispatcher
} }
@ -93,8 +97,8 @@ func (c *worker) stream(val interface{}) {
if !c.running { if !c.running {
c.mutex.Lock() c.mutex.Lock()
c.running = true c.running = true
atomic.AddUint32(&running, 1) atomic.AddUint32(c.broker, 1)
defer atomic.AddUint32(&running, ^uint32(1-1)) defer atomic.AddUint32(c.broker, ^uint32(1 - 1))
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
err := c.Before(ctx) err := c.Before(ctx)