Golang分布式同步操作转换(OT)插件库DOT的使用

DOT项目结合了操作转换(OT)、CmRDT、持久化/不可变数据结构和响应式流处理技术，实现了分布式数据同步和冲突自由合并的丰富自定义数据结构。

主要特性

1. 小型、经过良好测试的变更和不可变持久化值
2. 支持丰富的用户自定义类型，而不仅仅是协作文本
3. 流和类似Git的分支、合并支持
4. 简单的网络支持(Gob序列化)和存储支持
5. 自动随变更更新的强引用支持
6. 任何类型和变更的丰富内置撤销支持
7. 折叠(在未提交变更之上的已提交变更)
8. 支持CmRDT类型

完整示例Demo

服务器端实现

func Server() {
    // 使用基于本地文件的bolt DB后端
    http.Handle("/dot/", dot.BoltServer("file.bolt"))
    http.ListenAndServe(":8080", nil)
}

定义TODO类型

// Todo跟踪单个待办事项
type Todo struct {
    Complete bool
    Description string
}

// TodoList跟踪待办事项集合
type TodoList []Todo

类型注册

func init() {
    nw.Register(Todo{})
    nw.Register(TodoList{})
}

切换完成状态

func Toggle(t *TodoListStream, index int) {
    todoStream := t.Item(index) // 获取索引位置的流
    completeStream := todoStream.Complete() // 获取Complete字段的流
    completeStream.Update(!completeStream.Value) // 切换状态
}

修改描述

func SpliceDescription(t *TodoListStream, index, offset, count int, replacement string) {
    todoStream := t.Item(index)
    descStream := todoStream.Description()
    descStream.Splice(offset, count, replacement) // 执行拼接操作
}

添加待办事项

func AddTodo(t *TodoListStream, todo Todo) {
    t.Splice(len(t.Value), 0, todo) // 在末尾添加
}

客户端连接

func Client(stop chan struct{}, render func(*TodoListStream)) {
    url := "http://localhost:8080/dot/"
    session, todos := SavedSession()
    s, store := session.NonBlockingStream(url, nil)
    defer store.Close()

    todosStream := &TodoListStream{Stream: s, Value: todos}
    
    ticker := time.NewTicker(500*time.Millisecond)
    changed := true
    for {
        if changed {
            render(todosStream)
        }
        select {
        case <- stop:
            return
        case <- ticker.C:
        }

        Lock.Lock()
        s.Push()
        s.Pull()
        next := todosStream.Latest()
        changed = next != todosStream
        todosStream, s = next, next.Stream
        Lock.Unlock()
    }

    SaveSession(session, todosStream.Value)
}

运行Demo

1. 生成代码文件:

$ go get github.com/tvastar/test/cmd/testmd
$ testmd -pkg example -o examples/todo.go README.md
$ testmd -pkg main codegen.md > examples/generated.go

1. 启动服务器:

$ go run server.go

1. 启动客户端:

$ go run client.go

工作原理

DOT的核心由值、变更和流组成:

1. 值实现Value接口，集合类型还实现Collection接口
2. 变更表示可以合并的值变更
3. 流表示值的变更序列，具有收敛性

基本变更示例

initial := types.S8("hello")
append := changes.Splice{
    Offset: len("hello"),
    Before: types.S8(""),
    After: types.S8(" world"),
}
updated := initial.Apply(nil, append)
fmt.Println(updated) // 输出: hello world

流变更示例

initial := &streams.S8{Stream: streams.New(), Value: "hello"}
updated := initial.Splice(5, 0, " world")
fmt.Println(updated.Value) // 输出: hello world

合并变更

initial := types.S8("hello")
insert := changes.Splice{Offset: 5, Before: types.S8(""), After: types.S8(" world")}
remove := changes.Splice{Offset: 3, Before: types.S8("lo"), After: types.S8("")}

inserted := initial.Apply(nil, insert)
removed := initial.Apply(nil, remove)

removex, insertx := insert.Merge(remove)

final1 := inserted.Apply(nil, removex)
final2 := removed.Apply(nil, insertx)

fmt.Println(final1, final1 == final2) // 输出: hel world true

撤销支持

master := &streams.S16{Stream: streams.New(), Value: "hello"}
s := undo.New(master.Stream)
undoableChild := &streams.S16{Stream: s, Value: master.Value}

undoableChild = undoableChild.Splice(0, len("h"), "H")
fmt.Println(undoableChild.Value) // 输出: Hello

s.Undo()
undoableChild = undoableChild.Latest()
fmt.Println(undoableChild.Value) // 输出: hello

DOT提供了强大的分布式同步功能，通过操作转换和流处理实现了数据的最终一致性，特别适合需要实时协作的应用场景。