关于golang:手撸golang-GO与微服务-ESCQRS模式之2

手撸 golang GO 与微服务 ES-CQRS 模式之 2

最近浏览 [Go 微服务实战] (刘金亮, 2021.1)
本系列笔记拟采纳 golang 练习之
gitee:

https://gitee.com/ioly/learning.gooop

ES(Event Sourcing) 事件溯源十分好了解，指的是将每次的事件都记录下来，而不是去记录对象的状态。比方新建、批改等都会作为事件记录下来，当须要最新的状态时，通过事件的重叠来计算最新的状态。依照事件溯源的模式进行架构设计，就是事件驱动架构（Event DrivenArchitecture, EDA）。命令查问职责拆散（CQRS）最早来自 Betrand Meyer 写的
Object-OrientedSoftware Construction 一书，指的是命令查问拆散（Command Query Separation,CQS）。其根本思维是任何一个对象的办法都能够分为以下两大类:
▪ 命令（Command）：不返回任何后果（void），但会扭转对象的状态。▪ 查问（Query）：返回后果，然而不会扭转对象的状态，对系统没有副作用。CQRS 的外围出发点就是把零碎分为读和写两局部，从而不便别离进行优化。

依据 ES-CQRS 模式, 大幅重构 Day 1 的设计, 并进行单元测试

TodoDTO: 待办事宜数值对象
OperationTag: todo 写入事件的类型标记
TodoEvent: todo 写入事件
ClassTag: json 序列化的类型标记
tJSONData: json 序列化的数据容器
IEventBus: 事件总线接口
iTodoEventSerializer: 事件序列化到 JSON 数据的接口
iTodoReader: todo 读取接口
iTodoWriter: todo 写入接口
iJSONStore: json 文件读写接口
ITodoService: todo 待办事宜服务接口
tEventBus: 事件总线的实现
tTodoEventSerializer: 事件序列化到 JSON 的实现
tTodoWriter: 事件写入器的实现, 监听 write 指令, 并长久化到 json 存储
tMockJSONStore: 虚构的 JSON 文件读写实现
tTodoReader: 待办事宜读取器, 监听 write 和 load 指令, 并计算 todo 列表的以后状态
tMockTodoService: 待办事宜服务的实现

todo_app_test.go

package es_cqrs

import (
    td "learning/gooop/es_cqrs/todo_app"
    "testing"
)

func fnAssertTrue (t *testing.T, b bool, msg string) {
    if !b {t.Fatal(msg)
    }
}

func Test_TodoApp(t *testing.T) {t1 := &td.TodoDTO{ 1, "title-1", "content-1"}
    td.MockTodoService.Create(t1)

    all := td.MockTodoService.GetAll()
    fnAssertTrue(t, len(all) == 1, "expecting 1 item")
    fnAssertTrue(t, all[0].Title == t1.Title, "expecting" + t1.Title)
    t.Log("pass creating")

    t1.Content = "content-1 updated"
    t1.Title = "title-1 updated"
    td.MockTodoService.Update(t1)
    all = td.MockTodoService.GetAll()
    fnAssertTrue(t, len(all) == 1, "expecting 1 item")
    fnAssertTrue(t, all[0].Content == t1.Content, "expecting" + t1.Content)
    t.Log("pass updating")

    td.MockTodoService.Delete(t1)
    all = td.MockTodoService.GetAll()
    fnAssertTrue(t, len(all) == 0, "expecting 0 items")
    t.Log("pass deleting")
}

$ go test -v todo_app_test.go 
=== RUN   Test_TodoApp
22:38:08.180382833 eventbus.Pub, event=todo.write.cmd, handler=tTodoWriter.1
22:38:08.180533659 tMockJSONStore.items: {"Tag":1,"Data":{"NO":1,"Title":"title-1","Content":"content-1"}}
22:38:08.180539669 eventbus.Pub, event=todo.write.cmd, handler=tTodoReader.2
22:38:08.180552255 tTodoReader.items: [&{1 title-1 content-1}]
22:38:08.180557245 eventbus.Pub, event=todo.read.cmd, handler=tTodoReader.2
22:38:08.180560995 eventbus.Pub, event=todo.read.ret, handler=tMockTodoService
    todo_app_test.go:21: pass creating
22:38:08.180580644 eventbus.Pub, event=todo.write.cmd, handler=tTodoWriter.1
22:38:08.180604465 tMockJSONStore.items: {"Tag":1,"Data":{"NO":1,"Title":"title-1","Content":"content-1"}}, {"Tag":2,"Data":{"NO":1,"Title":"title-1 updated","Content":"content-1 updated"}}
22:38:08.180612665 eventbus.Pub, event=todo.write.cmd, handler=tTodoReader.2
22:38:08.180618512 tTodoReader.items: [&{1 title-1 updated content-1 updated}]
22:38:08.18062244 eventbus.Pub, event=todo.read.cmd, handler=tTodoReader.2
22:38:08.180626445 eventbus.Pub, event=todo.read.ret, handler=tMockTodoService
    todo_app_test.go:29: pass updating
22:38:08.180642172 eventbus.Pub, event=todo.write.cmd, handler=tTodoWriter.1
22:38:08.180656612 tMockJSONStore.items: {"Tag":1,"Data":{"NO":1,"Title":"title-1","Content":"content-1"}}, {"Tag":2,"Data":{"NO":1,"Title":"title-1 updated","Content":"content-1 updated"}}, {"Tag":3,"Data":{"NO":1,"Title":"title-1 updated","Content":"content-1 updated"}}
22:38:08.180669129 eventbus.Pub, event=todo.write.cmd, handler=tTodoReader.2
22:38:08.180672774 tTodoReader.items: []
22:38:08.180675952 eventbus.Pub, event=todo.read.cmd, handler=tTodoReader.2
22:38:08.180679309 eventbus.Pub, event=todo.read.ret, handler=tMockTodoService
    todo_app_test.go:34: pass deleting
--- PASS: Test_TodoApp (0.00s)
PASS
ok      command-line-arguments  0.002s

待办事宜数值对象

package todo_app

type TodoDTO struct {
    NO      int
    Title   string
    Content string
}

func (me *TodoDTO) Clone() *TodoDTO {
    return &TodoDTO{me.NO, me.Title, me.Content,}
}

todo 写入事件的类型标记

package todo_app


type OperationTag int

const OPCreated OperationTag = 1
const OPUpdated OperationTag = 2
const OPDeleted OperationTag = 3

todo 写入事件

package todo_app

type TodoEvent struct {
    Tag OperationTag
    Data *TodoDTO
}

json 序列化的类型标记

package todo_app

type ClassTag int

const TodoEventClass ClassTag = 1

json 序列化的数据容器

package todo_app

import "encoding/json"

type tJSONData struct {
    Tag     ClassTag
    Content []byte}



func (me *tJSONData) Set(tag ClassTag, it interface{}) error {
    me.Tag = tag

    j, e := json.Marshal(it)
    if e != nil {return e}
    me.Content = j
    return nil
}

func (me *tJSONData) Get(it interface{}) error {return json.Unmarshal(me.Content, it)
}

事件总线接口

package todo_app

type EventHandleFunc func(e string, args interface{})
type EventHandler struct {
    ID      string
    Handler EventHandleFunc
}

type IEventBus interface {Pub(e string, args interface{})
    Sub(e string, id string, handleFunc EventHandleFunc)
    Unsub(e string, id string)
}

const EventWriteTodoCmd = "todo.write.cmd"
const EventReadTodoCmd = "todo.read.cmd"
const EventReadTodoRet = "todo.read.ret"
const EventLoadTodoCmd = "todo.load.cmd"

事件序列化到 JSON 数据的接口

package todo_app

type iTodoEventSerializer interface {Serialize(it *TodoEvent) *tJSONData
}

todo 读取接口

package todo_app

type iTodoReader interface {All() []*TodoDTO
    HandleTodoEvent(e *TodoEvent)
}

todo 写入接口

package todo_app

type iTodoWriter interface {HandleTodoEvent(e *TodoEvent)
}

json 文件读写接口

package todo_app

type iJSONStore interface {Load()
    Append(it *tJSONData)
}

todo 待办事宜服务接口

package todo_app

type ITodoService interface {Create(it *TodoDTO)
    Update(it *TodoDTO)
    Delete(it *TodoDTO)

    GetAll() []*TodoDTO
}

事件总线的实现

package todo_app

import (
    "learning/gooop/saga/mqs/logger"
    "sync"
)

type tEventBus struct {
    rwmutex *sync.RWMutex
    items   map[string][]*EventHandler}

func newEventHandler(id string, handleFunc EventHandleFunc) *EventHandler {
    return &EventHandler{id, handleFunc,}
}

func newEventBus() IEventBus {it := new(tEventBus)
    it.init()
    return it
}

func (me *tEventBus) init() {me.rwmutex = new(sync.RWMutex)
    me.items = make(map[string][]*EventHandler)
}

func (me *tEventBus) Pub(e string, args interface{}) {me.rwmutex.RLock()
    defer me.rwmutex.RUnlock()

    handlers, ok := me.items[e]
    if ok {
        for _, it := range handlers {logger.Logf("eventbus.Pub, event=%s, handler=%s", e, it.ID)
            it.Handler(e, args)
        }
    }
}

func (me *tEventBus) Sub(e string, id string, handleFunc EventHandleFunc) {me.rwmutex.Lock()
    defer me.rwmutex.Unlock()

    handler := newEventHandler(id, handleFunc)
    handlers, ok := me.items[e]

    if ok {me.items[e] = append(handlers, handler)
    } else {me.items[e] = []*EventHandler{handler}
    }
}

func (me *tEventBus) Unsub(e string, id string) {me.rwmutex.Lock()
    defer me.rwmutex.Unlock()

    handlers, ok := me.items[e]
    if ok {
        for i, it := range handlers {
            if it.ID == id {lastI := len(handlers) - 1
                if i != lastI {handlers[i], handlers[lastI] = handlers[lastI], handlers[i]
                }
                me.items[e] = handlers[:lastI]
            }
        }
    }
}

var GlobalEventBus = newEventBus()

事件序列化到 JSON 的实现

package todo_app

type tTodoEventSerializer struct {
}

func newEventSeiralizer() iTodoEventSerializer {it := new(tTodoEventSerializer)
    return it
}

func (me *tTodoEventSerializer) Serialize(e *TodoEvent) *tJSONData {it := new(tJSONData)
    err := it.Set(TodoEventClass, e)
    if err != nil {return nil}
    return it
}

var gDefaultEventSerializer = newEventSeiralizer()

事件写入器的实现, 监听 write 指令, 并长久化到 json 存储

package todo_app

import (
    "fmt"
    "sync/atomic"
)

type tTodoWriter struct {id string}

func newTodoWriter() iTodoWriter {it := new(tTodoWriter)
    it.init()
    return it
}

func (me *tTodoWriter) init() {me.id = fmt.Sprintf("tTodoWriter.%d", atomic.AddInt32(&gWriterCounter, 1))

    GlobalEventBus.Sub(EventWriteTodoCmd, me.id, me.handleWriteTodoCmd)
}

func (me *tTodoWriter) handleWriteTodoCmd(e string, args interface{}) {
    switch e {
    case EventWriteTodoCmd:
        if it, ok := args.(*TodoEvent); ok {me.HandleTodoEvent(it)
        }
        break
    }
}

func (me *tTodoWriter) HandleTodoEvent(e *TodoEvent) {j := gDefaultEventSerializer.Serialize(e)
    if j != nil {MockJSONStore.Append(j)
    }
}

var gWriterCounter int32 = 0

虚构的 JSON 文件读写实现

package todo_app

import (
    "fmt"
    "learning/gooop/saga/mqs/logger"
    "strings"
    "sync"
)

type tMockJSONStore struct {
    rwmutex *sync.RWMutex
    once    sync.Once
    items   []*tJSONData}

func newMockJSONStore() iJSONStore {it := new(tMockJSONStore)
    it.init()
    return it
}

func (me *tMockJSONStore) init() {me.rwmutex = new(sync.RWMutex)
    me.items = []*tJSONData{}
}

func (me *tMockJSONStore) Load() {me.once.Do(func() {me.rwmutex.RLock()
        defer me.rwmutex.RUnlock()

        for _, it := range me.items {
            switch it.Tag {
            case TodoEventClass:
                v := new(TodoEvent)
                e := it.Get(v)
                if e == nil {GlobalEventBus.Pub(EventLoadTodoCmd, e)
                }
                break
            }
        }
    })
}

func (me *tMockJSONStore) Append(it *tJSONData) {me.rwmutex.Lock()
    defer me.rwmutex.Unlock()

    me.items = append(me.items, it)

    lines := []string{}
    for _,it := range me.items {lines = append(lines, fmt.Sprintf("%s", string(it.Content)))
    }
    logger.Logf("tMockJSONStore.items: %s", strings.Join(lines, ","))
}

var MockJSONStore = newMockJSONStore()

待办事宜读取器, 监听 write 和 load 指令, 并计算 todo 列表的以后状态

package todo_app

import (
    "fmt"
    "learning/gooop/saga/mqs/logger"
    "strings"
    "sync"
    "sync/atomic"
)

type tTodoReader struct {
    id string
    rwmutex *sync.RWMutex
    items []*TodoDTO}

func newTodoReader() iTodoReader {it := new(tTodoReader)
    it.init()
    return it
}

func (me *tTodoReader) init() {id := fmt.Sprintf("tTodoReader.%d", atomic.AddInt32(&gReaderCounter, 1))
    me.id = id
    me.rwmutex = new(sync.RWMutex)

    GlobalEventBus.Sub(EventWriteTodoCmd, me.id, me.handleEvent)
    GlobalEventBus.Sub(EventLoadTodoCmd, me.id, me.handleEvent)
    GlobalEventBus.Sub(EventReadTodoCmd, me.id, me.handleEvent)
}


func (me *tTodoReader) handleEvent(e string, args interface{}) {
    switch e {
    case EventWriteTodoCmd:
        fallthrough
    case EventLoadTodoCmd:
        if v,ok := args.(*TodoEvent);ok {me.HandleTodoEvent(v)
        }
        break

    case EventReadTodoCmd:
        me.handleReadTodoList()}
}

func (me *tTodoReader) handleReadTodoList() {GlobalEventBus.Pub(EventReadTodoRet, me.All())
}

func (me *tTodoReader) All() []*TodoDTO {me.rwmutex.RLock()
    defer me.rwmutex.RUnlock()

    lst := make([]*TodoDTO, len(me.items))
    for i,it := range me.items {lst[i] = it
    }
    return lst
}

func (me *tTodoReader) HandleTodoEvent(e *TodoEvent) {me.rwmutex.Lock()
    defer me.rwmutex.Unlock()

    switch e.Tag {
    case OPCreated:
        me.items = append(me.items, e.Data.Clone())
        break

    case OPUpdated:
        for i,it := range me.items {
            if it.NO == e.Data.NO {me.items[i] = e.Data.Clone()
                break
            }
        }
        break

    case OPDeleted:
        for i,it := range me.items {
            if it.NO == e.Data.NO {lastI := len(me.items) - 1
                if i == lastI {me.items[i] = nil
                } else {me.items[i], me.items[lastI] = me.items[lastI], nil
                }

                me.items = me.items[:lastI]
                break
            }
        }
        break
    }

    lines := []string{}
    for _,it := range me.items {lines = append(lines, fmt.Sprintf("%v", it))
    }
    logger.Logf("tTodoReader.items: [%s]", strings.Join(lines, ","))
}

var gReaderCounter int32 = 1

待办事宜服务的实现, 提供 todo 项的 CRUD

package todo_app

type tMockTodoService struct {items []*TodoDTO
    writer iTodoWriter
    reader iTodoReader
}

func newMockTodoService() ITodoService {it := new(tMockTodoService)
    it.init()
    return it
}

func (me *tMockTodoService) init() {me.writer = newTodoWriter()
    me.reader = newTodoReader()

    GlobalEventBus.Sub(EventReadTodoRet, "tMockTodoService", me.handleReadTodoRet)
}

func (me *tMockTodoService) handleReadTodoRet(e string, args interface{}) {
    switch e {
    case EventReadTodoRet:
        if it,ok := args.([]*TodoDTO);ok {me.items = it}
        break
    }
}

func (me *tMockTodoService) Create(it *TodoDTO) {GlobalEventBus.Pub(EventWriteTodoCmd, &TodoEvent{ OPCreated, it.Clone() })
}

func (me *tMockTodoService) Update(it *TodoDTO) {GlobalEventBus.Pub(EventWriteTodoCmd, &TodoEvent{ OPUpdated, it.Clone() })
}

func (me *tMockTodoService) Delete(it *TodoDTO) {GlobalEventBus.Pub(EventWriteTodoCmd, &TodoEvent{ OPDeleted, it.Clone() })
}

func (me *tMockTodoService) GetAll() []*TodoDTO {
    me.items = nil
    GlobalEventBus.Pub(EventReadTodoCmd, nil)

    lst := me.items
    me.items = nil
    return lst
}

var MockTodoService = newMockTodoService()

(ES-CQRS end)

关于golang:手撸golang-GO与微服务-ESCQRS模式之2

缘起

ES-CQRS 模式

指标 (Day 2)

设计

单元测试

测试输入

TodoDTO.go

OperationTag.go

TodoEvent.go

ClassTag.go

tJSONData.go

IEventBus.go

iTodoEventSerializer.go

iTodoReader.go

iTodoWriter.go

iJSONStore.go

ITodoService.go

tEventBus.go

tTodoEventSerializer.go

tTodoWriter.go

tMockJSONStore.go

tTodoReader.go

tMockTodoService.go