jiacai2050 / prosumer

A producer-consumer solution for Golang

Prosumer

A producer-consumer solution for Golang.

Motivation

Go is popular for its simplicity, builtin support for concurrency, light-weight goroutine. However, there are some tricks(here and there) when to coordinate among different goroutines, especially when implements producer-consumer pattern using buffered chan.

I don't want cover details here, guys who interested can check links above. But following cannot be emphasized too much:

Close chan is a sender -> receiver communication, not the reverse. #11344

Feature

• Inner buffer queue support RejectPolicy
• Graceful close implemented

If you have any suggestions/questions, open a PR.

Usage

There is a quickstart example , view the GoDoc for details.

License

MIT License

prosumer's Issues

Allow more flexible consumer signature

The consumer uses func(ls []interface{}) error which requires that all elements be individually cast:

consumer := func(ls []interface{}) error {
	var items []int
	for _, item := range items {
		items = append(items, item.(int))
	}

	// ...
	return nil
}

Using reflect you could have a dynamic signature (which would be backwards compatible with []interface{}):

consumer := func(ls []int) error {
	fmt.Printf("get %+v \n", ls)
	wg.Add(-len(ls))
	return nil
}

How to write subscription publishing mode (observer mode)

Subscription publishing mode is a special production and consumer mode, so how to write subscription publishing mode (observer mode) using jiacai2050/prosumer

suggestions from reddit

https://www.reddit.com/r/golang/comments/cd14x2/a_producerconsumer_solution_for_golang/etr9drk?utm_source=share&utm_medium=web2x

• make logger as params, if nil, don't log anything.
• make queue private
• force close
• use a timer instead of calling time.After, as you can re-use timers saving allocs.
  Ticker isn't suitable here, as for doWork, which reset last, will be called when batchSize or batchInterval is met.
• received := make([]interface{}, 0, batchSize)
• abstract Coordinator.Start() logic into worker type.
• prosumer.Put(..) return error