实现pipe-filter framework
image
Pipe-Filter 模式：
- ⾮常适合与数据处理及数据分析系统
- Filter封装数据处理的功能
- Pipe⽤于连接Filter传递数据或者在异步处理过程中缓冲数据流
- 进程内同步调⽤时，pipe演变为数据在⽅法调⽤间传递
- 松耦合：Filter只跟数据（格式）耦合
Filter和组合模式：
image
image
示例：
image
简单示例代码：
filter.go
// Package pipefilter is to define the interfaces and the structures for pipe-filter style implementation
package pipefilter
// Request is the input of the filter
type Request interface{}
// Response is the output of the filter
type Response interface{}
// Filter interface is the definition of the data processing components
// Pipe-Filter structure
type Filter interface {
Process(data Request) (Response, error)
}
split_filter.go
package pipefilter
import (
"errors"
"strings"
)
var SplitFilterWrongFormatError = errors.New("input data should be string")
type SplitFilter struct {
delimiter string
}
func NewSplitFilter(delimiter string) *SplitFilter {
return &SplitFilter{delimiter}
}
func (sf *SplitFilter) Process(data Request) (Response, error) {
str, ok := data.(string) //检查数据格式/类型，是否可以处理
if !ok {
return nil, SplitFilterWrongFormatError
}
parts := strings.Split(str, sf.delimiter)
return parts, nil
}
split_filter_test.go
package pipefilter
import (
"reflect"
"testing"
)
func TestStringSplit(t *testing.T) {
sf := NewSplitFilter(",")
resp, err := sf.Process("1,2,3")
if err != nil {
t.Fatal(err)
}
parts, ok := resp.([]string)
if !ok {
t.Fatalf("Repsonse type is %T, but the expected type is string", parts)
}
if !reflect.DeepEqual(parts, []string{"1", "2", "3"}) {
t.Errorf("Expected value is {\"1\",\"2\",\"3\"}, but actual is %v", parts)
}
}
func TestWrongInput(t *testing.T) {
sf := NewSplitFilter(",")
_, err := sf.Process(123)
if err == nil {
t.Fatal("An error is expected.")
}
}
实现micro-kernel framework
image
-
特点
- 易于扩展
- 错误隔离
- 保持架构⼀致性
-
要点
-
内核包含公共流程或通⽤逻辑
- 将可变或可扩展部分规划为扩展点
-
抽象扩展点⾏为，定义接⼝
- 利⽤插件进⾏扩展
示例：
image
简单示例代码：
agent.go
package microkernel
import (
"context"
"errors"
"fmt"
"strings"
"sync"
)
const (
Waiting = iota
Running
)
var WrongStateError = errors.New("can not take the operation in the current state")
type CollectorsError struct {
CollectorErrors []error
}
func (ce CollectorsError) Error() string {
var strs []string
for _, err := range ce.CollectorErrors {
strs = append(strs, err.Error())
}
return strings.Join(strs, ";")
}
type Event struct {
Source string
Content string
}
type EventReceiver interface {
OnEvent(evt Event)
}
type Collector interface {
Init(evtReceiver EventReceiver) error
Start(agtCtx context.Context) error
Stop() error
Destory() error
}
type Agent struct {
collectors map[string]Collector
evtBuf chan Event
cancel context.CancelFunc
ctx context.Context
state int
}
func (agt *Agent) EventProcessGroutine() {
var evtSeg [10]Event
for {
for i := 0; i < 10; i++ {
select {
case evtSeg[i] = <-agt.evtBuf:
case <-agt.ctx.Done():
return
}
}
fmt.Println(evtSeg)
}
}
func NewAgent(sizeEvtBuf int) *Agent {
agt := Agent{
collectors: map[string]Collector{},
evtBuf: make(chan Event, sizeEvtBuf),
state: Waiting,
}
return &agt
}
func (agt *Agent) RegisterCollector(name string, collector Collector) error {
if agt.state != Waiting {
return WrongStateError
}
agt.collectors[name] = collector
return collector.Init(agt)
}
func (agt *Agent) startCollectors() error {
var err error
var errs CollectorsError
var mutex sync.Mutex
for name, collector := range agt.collectors {
go func(name string, collector Collector, ctx context.Context) {
defer func() {
mutex.Unlock()
}()
err = collector.Start(ctx)
mutex.Lock()
if err != nil {
errs.CollectorErrors = append(errs.CollectorErrors,
errors.New(name+":"+err.Error()))
}
}(name, collector, agt.ctx)
}
if len(errs.CollectorErrors) == 0 {
return nil
}
return errs
}
func (agt *Agent) stopCollectors() error {
var err error
var errs CollectorsError
for name, collector := range agt.collectors {
if err = collector.Stop(); err != nil {
errs.CollectorErrors = append(errs.CollectorErrors,
errors.New(name+":"+err.Error()))
}
}
if len(errs.CollectorErrors) == 0 {
return nil
}
return errs
}
func (agt *Agent) destoryCollectors() error {
var err error
var errs CollectorsError
for name, collector := range agt.collectors {
if err = collector.Destory(); err != nil {
errs.CollectorErrors = append(errs.CollectorErrors,
errors.New(name+":"+err.Error()))
}
}
if len(errs.CollectorErrors) == 0 {
return nil
}
return errs
}
func (agt *Agent) Start() error {
if agt.state != Waiting {
return WrongStateError
}
agt.state = Running
agt.ctx, agt.cancel = context.WithCancel(context.Background())
go agt.EventProcessGroutine()
return agt.startCollectors()
}
func (agt *Agent) Stop() error {
if agt.state != Running {
return WrongStateError
}
agt.state = Waiting
agt.cancel()
return agt.stopCollectors()
}
func (agt *Agent) Destory() error {
if agt.state != Waiting {
return WrongStateError
}
return agt.destoryCollectors()
}
func (agt *Agent) OnEvent(evt Event) {
agt.evtBuf <- evt
}
agent_test.go
package microkernel
import (
"context"
"errors"
"fmt"
"testing"
"time"
)
type DemoCollector struct {
evtReceiver EventReceiver
agtCtx context.Context
stopChan chan struct{}
name string
content string
}
func NewCollect(name string, content string) *DemoCollector {
return &DemoCollector{
stopChan: make(chan struct{}),
name: name,
content: content,
}
}
func (c *DemoCollector) Init(evtReceiver EventReceiver) error {
fmt.Println("initialize collector", c.name)
c.evtReceiver = evtReceiver
return nil
}
func (c *DemoCollector) Start(agtCtx context.Context) error {
fmt.Println("start collector", c.name)
for {
select {
case <-agtCtx.Done():
c.stopChan <- struct{}{}
break
default:
time.Sleep(time.Millisecond * 50)
c.evtReceiver.OnEvent(Event{c.name, c.content})
}
}
}
func (c *DemoCollector) Stop() error {
fmt.Println("stop collector", c.name)
select {
case <-c.stopChan:
return nil
case <-time.After(time.Second * 1):
return errors.New("failed to stop for timeout")
}
}
func (c *DemoCollector) Destory() error {
fmt.Println(c.name, "released resources.")
return nil
}
func TestAgent(t *testing.T) {
agt := NewAgent(100)
c1 := NewCollect("c1", "1")
c2 := NewCollect("c2", "2")
agt.RegisterCollector("c1", c1)
agt.RegisterCollector("c2", c2)
if err := agt.Start(); err != nil {
fmt.Printf("start error %v\n", err)
}
fmt.Println(agt.Start())
time.Sleep(time.Second * 1)
agt.Stop()
agt.Destory()
}
网友评论