What is Stream Processing
Java developers should be very impressed with Stream API in Java, which greatly improves the ability to handle data collections.
int sum = widgets.stream()
.filter(w -> w.getColor() == RED)
.mapToInt(w -> w.getWeight())
.sum();
The idea of Stream is to abstract the data processing into a data stream and return a new stream for use after each process.
Stream Function Definition
The most important step is to think through the requirements before writing the code, so let's try to put ourselves in the author's shoes and think about the flow of the component. First of all, let's put the underlying implementation logic aside and try to define the stream function from scratch.
Stream's workflow is actually part of the production-consumer model, and the whole process is very similar to the production process in a factory.
- creation phase/data acquisition (raw material)
- processing phase/intermediate processing (pipeline processing)
- aggregation stage/final operation (final product)
The API is defined around the three life cycles of a stream.
Creation Phase
In order to create the abstract object stream, it can be understood as a constructor.
We support three ways of constructing streams: slicing conversion, channel conversion, and functional conversion.
Note that the methods in this phase are normal public methods and are not bound to the stream object.
func Just(items ... .interface{}) Stream
func Range(source <-chan interface{}) Stream
func From(generate GenerateFunc) Stream
func Concat(s Stream, others . . Stream) Stream
Processing phase
The operations required in the processing phase often correspond to our business logic, such as conversion, filtering, de-duplication, sorting, and so on.
The API for this phase is a method that needs to be bound to a Stream object.
The following definition is combined with common business scenarios.
Distinct(keyFunc KeyFunc) Stream
Filter(filterFunc FilterFunc, opts ... . Option) Stream
Group(fn KeyFunc) Stream
Head(n int64) Stream
Tail(n int64) Stream
Map(fn MapFunc, opts . . Option) Stream
Merge() Stream
Reverse() Stream
Sort(fn LessFunc) Stream
Walk(fn WalkFunc, opts ... . Option) Stream
Concat(streams ... . Stream) Stream
The processing logic of the processing phase returns a new Stream object, and there is a basic implementation paradigm here.
Aggregation stage
The aggregation phase is actually the result of the processing we want, e.g. whether it matches, count the number, traverse, etc.
AllMatch(fn PredicateFunc) bool
AnyMatch(fn PredicateFunc) bool
NoneMatch(fn PredicateFunc) bool
Count() int
Done()
ForAll(fn ForAllFunc)
ForEach(fn ForEachFunc)
After sorting out the requirements boundaries of the component, we have a clearer idea of what we are going to implement with Stream. In my perception, a real architect's grasp of requirements and their subsequent evolution can be very precise, and this can only be achieved by thinking deeply about the requirements and penetrating the essence behind them. By replacing the author's perspective to simulate the entire project build process, learning the author's thinking methodology is the greatest value of our learning open source projects.
Well, let's try to define the complete Stream interface and functions.
The role of the interface is not just a template, but also to use its abstraction capabilities to build the overall framework of the project without getting bogged down in the details at the beginning, to quickly express our thinking process through the interface concisely, to learn to develop a top-down thinking approach to observe the whole system from a global perspective, it is easy to get bogged down in the details at the beginning.
rxOptions struct {
unlimitedWorkers bool
workers int
}
Option func(opts *rxOptions)
KeyFunc func(item interface{}) interface{}
FilterFunc func(item interface{}) bool
MapFunc func(intem interface{}) interface{}
LessFunc func(a, b interface{}) bool
WalkFunc func(item interface{}, pip chan<- interface{})
PredicateFunc func(item interface{}) bool
ForAllFunc func(pip <-chan interface{})
ForEachFunc func(item interface{})
ParallelFunc func(item interface{})
ReduceFunc func(pip <-chan interface{}) (interface{}, error)
GenerateFunc func(source <-chan interface{})
Stream interface {
Distinct(keyFunc KeyFunc) Stream
Filter(filterFunc FilterFunc, opts . . Option) Stream
Group(fn KeyFunc) Stream
Head(n int64) Stream
Tail(n int64) Stream
First() interface{}
Last() interface{}
Map(fn MapFunc, opts . . Option) Stream
Merge() Stream
Reverse() Stream
Sort(fn LessFunc) Stream
Walk(fn WalkFunc, opts ... . Option) Stream
Concat(streams ... . Stream) Stream
AllMatch(fn PredicateFunc) bool
AnyMatch(fn PredicateFunc) bool
NoneMatch(fn PredicateFunc) bool
Count() int
Done()
ForAll(fn ForAllFunc)
ForEach(fn ForEachFunc)
}
The channel() method is used to get the Stream pipeline properties, since we are dealing with the interface object in the implementation, we expose a private method to read out.
channel() chan interface{}
Implementation ideas
With the functional definition sorted out, next consider a few engineering implementations.
How to implement chain calls
Chain calls, the builder pattern used to create objects can achieve the chain call effect. In fact, Stream implements a similar chain effect on the same principle, creating a new Stream to return in each call.
Distinct(keyFunc KeyFunc) Stream
Filter(filterFunc FilterFunc, opts . . Option) Stream
How to achieve the effect of pipeline processing
The pipeline can be understood as a storage container for data in Stream. In go we can use channel as a pipeline for data to achieve the effect of asynchronous non-blocking when Stream chain calls perform multiple operations.
How to support parallel processing
Data processing is essentially processing the data in the channel, so to achieve parallel processing is simply to consume the channel in parallel, using the goroutine and WaitGroup can be very convenient to achieve parallel processing.
go-zero implementation
core/fx/stream.go
The implementation of Stream in go-zero does not define an interface, but the logic is the same when it comes to the underlying implementation.
To implement the Stream interface we define an internal implementation class, where source is of type channel, to emulate the pipeline functionality.
Stream struct {
source <-chan interface{}
}
Create API
channel Creation of Range
Create stream via channel
func Range(source <-chan interface{}) Stream {
return Stream{
source: source,
}
}
Variable Parameter Pattern Creation of Just
It's a good habit to create streams in variable parameter mode and close the channel when you're done writing.
func Just(items ... .interface{}) Stream {
source := make(chan interface{}, len(items))
for _, item := range items {
source <- item
}
close(source)
return Range(source)
}
function to create From
Stream creation by function
func From(generate GenerateFunc) Stream {
source := make(chan interface{})
threading.GoSafe(func() {
defer close(source)
generate(source)
})
return Range(source)
}
Because it involves external calls to function parameters, the execution process is not available so you need to catch runtime exceptions to prevent panic errors from being transmitted to the upper layers and crashing the application.
func Recover(cleanups ... . func()) {
for _, cleanup := range cleanups {
cleanup()
}
if r := recover(); r ! = nil {
logx.ErrorStack(r)
}
}
func RunSafe(fn func()) {
defer rescue.Recover()
fn()
}
func GoSafe(fn func()) {
go Runsage(fn)
}
Splicing Concat
Splice other Streams to create a new Stream, calling the internal Concat method method, the source code implementation of Concat will be analyzed later.
func Concat(s Stream, others . . Stream) Stream {
return s.Concat(others...)
}
Processing API
de-duplication Distinct
Because the function parameter KeyFunc func(item interface{}) interface{} is passed in, it means that it also supports custom distincting according to business scenarios, essentially using the results returned by KeyFunc to achieve distincting based on a map.
The function arguments are very powerful and provide a great deal of flexibility.
func (s Stream) Distinct(keyFunc KeyFunc) Stream {
source := make(chan interface{})
threading.GoSafe(func() {
defer close(source)
keys := make(map[interface{}]lang.PlaceholderType)
for item := range s.source {
key := keyFunc(item)
if _, ok := keys[key]; !ok {
source <- item
keys[key] = lang.
Placeholder }
}
})
return Range(source)
}
Use case.
Just(1, 2, 3, 3, 4, 5, 5).Distinct(func(item interface{}) interface{} {
return item
}).ForEach(func(item interface{}) {
t.Log(item)
})
Just(1, 2, 3, 3, 4, 5, 5).Distinct(func(item interface{}) interface{} {
uid := item.(int)
if uid > 3 {
return 4
}
return item
}).ForEach(func(item interface{}) {
t.Log(item)
})
Filter Filter
The actual filtering logic is delegated to the Walk method by abstracting the filtering logic into a FilterFunc and then acting on the item separately to decide whether to write back to a new channel based on the Boolean value returned by the FilterFunc.
The Option parameter contains two options.
- unlimitedWorkers No limit on the number of concurrent processes
- workers Limit the number of concurrent processes
FilterFunc func(item interface{}) bool
func (s Stream) Filter(filterFunc FilterFunc, opts . . Option) Stream {
return s.Walk(func(item interface{}, pip chan<- interface{}) {
if filterFunc(item) {
pip <- item
}
}, opts...)
}
Example usage.
func TestInternalStream_Filter(t *testing.T) {
channel := Just(1, 2, 3, 4, 5).Filter(func(item interface{}) bool {
return item.(int)%2 == 0
}).channel()
for item := range channel {
t.Log(item)
}
}
Iterate through the Walk
walk means walk, here it means to perform a WalkFunc operation on each item and write the result to a new Stream.
Note here that the order of the data in the channel of the new Stream is random because the internal concurrent mechanism is used to read and write data asynchronously.
WalkFunc func(item interface{}, pipe chan<- interface{})
func (s Stream) Walk(fn WalkFunc, opts . .Option) Stream {
option := buildOptions(opts...)
if option.unlimitedWorkers {
return s.walkUnLimited(fn, option)
}
return s.walkLimited(fn, option)
}
func (s Stream) walkUnLimited(fn WalkFunc, option *rxOptions) Stream {
pipe := make(chan interface{}, defaultWorkers)
go func() {
var wg sync.WaitGroup
for item := range s.source {
val := item
wg.Add(1)
threading.GoSafe(func() {
defer wg.Done()
fn(item, pipe)
})
}
wg.Wait()
close(pipe)
}()
return Range(pipe)
}
func (s Stream) walkLimited(fn WalkFunc, option *rxOptions) Stream {
pipe := make(chan interface{}, option.workers)
go func() {
var wg sync.WaitGroup
pool := make(chan lang.PlaceholderType, option.workers)
for item := range s.source {
val := item
pool <- lang.
wg.Add(1)
threading.GoSafe(func() {
defer func() {
wg.Done()
<-pool
}()
fn(item, pipe)
})
}
wg.Wait()
close(pipe)
}()
return Range(pipe)
}
Use case.
The order of returns is randomized.
func Test_Stream_Walk(t *testing.T) {
Just(1, 2, 3).Walk(func(item interface{}, pip chan<- interface{}) {
pip <- item.(int) * 100
}, WithWorkers(3)).ForEach(func(item interface{}) {
t.Log(item)
})
}
Grouping Groups
Put in map by matching item.
KeyFunc func(item interface{}) interface{}
func (s Stream) Group(fn KeyFunc) Stream {
groups := make(map[interface{}][]interface{})
for item := range s.source {
key := fn(item)
groups[key] = append(groups[key], item)
}
source := make(chan interface{})
go func() {
for _, group := range groups {
source <- group
}
close(source)
}()
return Range(source)
}
gets the first n elements of Head
n is greater than the actual dataset length, all elements will be returned
func (s Stream) Head(n int64) Stream {
if n < 1 {
panic("n must be greather than 1")
}
source := make(chan interface{})
go func() {
for item := range s.source {
n--
if n >= 0 {
source <- item
}
if n == 0 {
close(source)
break
}
}
if n > 0 {
close(source)
}
}()
return Range(source)
}
Example usage.
func TestInternalStream_Head(t *testing.T) {
channel := Just(1, 2, 3, 4, 5).Head(2).channel()
for item := range channel {
t.Log(item)
}
}
Get the last n elements of Tail
It is interesting to understand the implementation of the Ring in order to ensure that the last n elements are obtained using the Ring data structure.
type Ring struct {
elements []interface{}
index int
lock sync.Mutex
}
func NewRing(n int) *Ring {
if n < 1 {
panic("n should be greather than 0")
}
return &Ring{
elements: make([]interface{}, n),
}
}
func (r *Ring) Add(v interface{}) {
r.lock.Lock()
defer r.lock.Unlock()
r.elements[r.index%len(r.elements)] = v
r.index++
}
func (r *Ring) Take() []interface{} {
r.lock.Lock()
defer r.lock.Unlock()
var size int
var start int
if r.index > len(r.elements) {
size = len(r.elements)
start = r.index % len(r.elements)
} else {
size = r.index
}
elements := make([]interface{}, size)
for i := 0; i < size; i++ {
elements[i] = r.elements[(start+i)%len(r.elements)]
}
return elements
}
To summarize the advantages of ring slicing.
- Supports automatic scrolling updates
- Memory saving
Ring slicing enables old data to be overwritten by new data when the fixed capacity is full, and can be used to read n elements after the channel due to this feature.
func (s Stream) Tail(n int64) Stream {
if n < 1 {
panic("n must be greather than 1")
}
source := make(chan interface{})
go func() {
ring := collection.NewRing(int(n))
for item := range s.source {
ring.Add(item)
}
for _, item := range ring.Take() {
source <- item
}
close(source)
}()
return Range(source)
}
So why not just use a len(source) length slice?
The answer is to save memory. Any data structure that involves a ring type has the advantage of saving memory and allocating resources on demand.
Example usage.
func TestInternalStream_Tail(t *testing.T) {
channel := Just(1, 2, 3, 4, 5).Tail(2).channel()
for item := range channel {
t.Log(item)
}
channel2 := Just(1, 2, 3, 4, 5).Tail(6).channel()
for item := range channel2 {
t.Log(item)
}
}
element conversion Map
Element conversion, internally done by a concurrent process to complete the conversion operation, note that the output channel is not guaranteed to be output in the original order.
MapFunc func(intem interface{}) interface{}
func (s Stream) Map(fn MapFunc, opts . . Option) Stream {
return s.Walk(func(item interface{}, pip chan<- interface{}) {
pip <- fn(item)
}, opts...)
}
Example usage.
func TestInternalStream_Map(t *testing.T) {
channel := Just(1, 2, 3, 4, 5, 2, 2, 2, 2, 2, 2, 2).Map(func(item interface{}) interface{} {
return item.(int) * 10
}).channel()
for item := range channel {
t.Log(item)
}
}
Merge Merge
The implementation is relatively simple, and I've thought long and hard about what scenarios would be suitable for this method.
func (s Stream) Merge() Stream {
var items []interface{}
for item := range s.source {
items = append(items, item)
}
source := make(chan interface{}, 1)
source <- items
return Range(source)
}
Reverse
Reverses the elements of the channel. The flow of the reversal algorithm is
- Find the middle node
- The two sides of the node start swapping two by two
Notice why slices are used to receive s.source when it is fetched? Slices are automatically expanded, wouldn't it be better to use arrays?
In fact, you can't use arrays here, because you don't know that Stream writing to source is often done asynchronously in a concurrent process, and the channels in each Stream may change dynamically.
func (s Stream) Reverse() Stream {
var items []interface{}
for item := range s.source {
items = append(items, item)
}
for i := len(items)/2 - 1; i >= 0; i-- {
opp := len(items) - 1 - i
items[i], items[opp] = items[opp], items[i]
}
return Just(items...)
}
Example usage.
func TestInternalStream_Reverse(t *testing.T) {
channel := Just(1, 2, 3, 4, 5).Reverse().channel()
for item := range channel {
t.Log(item)
}
}
Sort
The intranet calls the official slice package sorting scheme, just pass in the comparison function to implement the comparison logic.
func (s Stream) Sort(fn LessFunc) Stream {
var items []interface{}
for item := range s.source {
items = append(items, item)
}
sort.Slice(items, func(i, j int) bool {
return fn(i, j)
})
return Just(items...)
}
Example usage.
func TestInternalStream_Sort(t *testing.T) {
channel := Just(1, 2, 3, 4, 5).Sort(func(a, b interface{}) bool {
return a.(int) > b.(int)
}).channel()
for item := range channel {
t.Log(item)
}
}
Splicing Concat
func (s Stream) Concat(steams . .Stream) Stream {
source := make(chan interface{})
go func() {
NewRoutineGroup()
group.Run(func() {
for item := range s.source {
source <- item
}
})
for _, stream := range steams {
group.Run(func() {
for item := range stream.channel() {
source <- item
}
})
}
group.Wait()
close(source)
}()
return Range(source)
}
Aggregate API
Match All AllMatch
func (s Stream) AllMatch(fn PredicateFunc) bool {
for item := range s.source {
if !fn(item) {
go drain(s.source)
return false
}
}
return true
}
Arbitrary Match AnyMatch
func (s Stream) AnyMatch(fn PredicateFunc) bool {
for item := range s.source {
if fn(item) {
go drain(s.source)
return true
}
}
return false
}
NoneMatch
func (s Stream) NoneMatch(fn func(item interface{}) bool) bool {
for item := range s.source {
if fn(item) {
go drain(s.source)
return false
}
}
return true
}
Quantity count Count
func (s Stream) Count() int {
var count int
for range s.source {
count++
}
return count
}
Clear Done
func (s Stream) Done() {
drain(s.source)
}
Iterate over all elements ForAll
func (s Stream) ForAll(fn ForAllFunc) {
fn(s.source)
}
Iterate over each element ForEach
func (s Stream) ForAll(fn ForAllFunc) {
fn(s.source)
}
Summary
The core logic is to use the channel as a pipe and the data as a stream, and to continuously receive/write data to the channel using a concurrent process to achieve an asynchronous non-blocking effect.
Going back to the problem mentioned at the beginning, it seems very difficult to implement a stream beforehand, and it is hard to imagine that such a powerful component can be implemented in 300+ lines of code in go.
The basis for this efficiency comes from three language features.
- channel
- concurrency
- functional programming
Reference
pipeline pattern
slice-reversal algorithm
Project address
https://github.com/zeromicro/go-zero
Welcome to use go-zero and star to support us!