前邊初探富文本之CRDT協(xié)同算法一文中我們探討了為什么需要協(xié)同、分布式的最終一致性理論、偏序集與半格的概念、為什么需要有偏序關(guān)系、如何通過數(shù)據(jù)結(jié)構(gòu)避免沖突、分布式系統(tǒng)如何進(jìn)行同步調(diào)度等等，這些屬于完成協(xié)同所需要了解的基礎(chǔ)知識(shí)，實(shí)際上當(dāng)前有很多成熟的協(xié)同實(shí)現(xiàn)，例如automerge、yjs等等，本文就是關(guān)注于以yjs為CRDT協(xié)同框架來(lái)實(shí)現(xiàn)協(xié)同的實(shí)例。

描述#

接入?yún)f(xié)同框架實(shí)際上并不是一件簡(jiǎn)單的事情，當(dāng)然相對(duì)于接入OT協(xié)同而言接入CRDT協(xié)同已經(jīng)是比較簡(jiǎn)單的了，因?yàn)槲覀冎恍枰劢褂跀?shù)據(jù)結(jié)構(gòu)的使用就好，而不需要對(duì)變換有過多的關(guān)注。當(dāng)前我們更加關(guān)注的是Op-based CRDT，本文所說的CRDT也是特指的Op-based CRDT，畢竟State-baed CRDT需要將全量數(shù)據(jù)進(jìn)行傳輸，每次都要完整傳輸狀態(tài)來(lái)完成同步讓它比較難變成通用的解決方案。因此與OT算法一樣，我們依然需要Operation，在富文本領(lǐng)域，最經(jīng)典的Operation有quill的delta模型，通過retain、insert、delete三個(gè)操作完成整篇文檔的描述與操作，還有slate的JSON模型，通過insert_text、split_node、remove_text等等操作來(lái)完成整篇文檔的描述與操作。假如此時(shí)是OT的話，接下來(lái)我們就要聊到變換Transformation了，但是使用CRDT算法的情況下，我們的關(guān)注點(diǎn)變了，我們需要做的是關(guān)注于如何將我們現(xiàn)在的數(shù)據(jù)結(jié)構(gòu)轉(zhuǎn)換為CRDT框架的數(shù)據(jù)結(jié)構(gòu)，比如通過框架提供的Array、Map、Text等類型構(gòu)建我們自己的JSON數(shù)據(jù)，并且我們的Op也需要映射到對(duì)框架提供的數(shù)據(jù)結(jié)構(gòu)進(jìn)行的操作，這樣框架便可以幫我們進(jìn)行協(xié)同，當(dāng)框架完成協(xié)同之后把框架的數(shù)據(jù)結(jié)構(gòu)的改變返回，此時(shí)我們需要再將這部分改變映射到我們自己的Op，然后我們只需要在本地應(yīng)用這些遠(yuǎn)程同步并在本地轉(zhuǎn)換的Op，就可以做到協(xié)同了。

上邊這個(gè)數(shù)據(jù)轉(zhuǎn)換聽起來(lái)是不是有點(diǎn)耳熟，在前邊初探富文本之OT協(xié)同實(shí)例中，我們介紹了json0，我們也提到了一個(gè)可行的操作，我們讓變換Transformation這部分讓json0去做，我們需要關(guān)注的是從我們自己定義的數(shù)據(jù)結(jié)構(gòu)轉(zhuǎn)換到json0，在json0進(jìn)行變換操作之后我們同樣地將Op轉(zhuǎn)換后應(yīng)用到我們本地的數(shù)據(jù)就好。雖然原理是完全不同的，但是我們?cè)谝延谐墒炜蚣艿那闆r下似乎并不需要關(guān)注這點(diǎn)，我們更側(cè)重于使用，實(shí)際上在使用起來(lái)是很像的。此時(shí)假設(shè)我們有一個(gè)自研的思維導(dǎo)圖功能需要實(shí)現(xiàn)協(xié)同，而保存的數(shù)據(jù)結(jié)構(gòu)都是自定義的，沒有直接可以調(diào)用的實(shí)現(xiàn)方案，我們就需要進(jìn)行轉(zhuǎn)換適配，那么如果使用OT的話，并且借助json0做變換，那么我們需要做的是把Op轉(zhuǎn)換為json0的Op，發(fā)送的數(shù)據(jù)也將會(huì)是這個(gè)json0的Op，那么如果直接使用CRDT的話，我們更像是通過框架定義的數(shù)據(jù)結(jié)構(gòu)將Op應(yīng)用到數(shù)據(jù)結(jié)構(gòu)上，發(fā)送的數(shù)據(jù)是框架定義的數(shù)據(jù)，類似于將Op應(yīng)用到數(shù)據(jù)結(jié)構(gòu)上了，其他的操作都由框架給予完整的支持了。實(shí)際上通過框架提供的例子后，接入CRDT協(xié)同就主要是理解并且實(shí)現(xiàn)的問題了，這樣就有一個(gè)大體的實(shí)現(xiàn)方向了，而不是毫無(wú)頭緒不知道應(yīng)該從哪里開始做協(xié)同。另外還是那個(gè)宗旨，合適的才是最好的，要考慮到實(shí)現(xiàn)的成本問題，沒有必要硬套數(shù)據(jù)結(jié)構(gòu)的實(shí)現(xiàn)，OT有OT的優(yōu)點(diǎn)，CRDT有CRDT的優(yōu)點(diǎn)，CRDT這類方法相比OT還比較年輕，還是在不斷發(fā)展過程中的，實(shí)際上有些問題例如內(nèi)存占用、速度等問題最近幾年才被比較好的解決，ShareDB作者在關(guān)注CRDT不斷發(fā)展的過程中也說了CRDTs are the future。此外從技術(shù)上講，CRDT類型是OT類型的子集，也就是說，CRDT實(shí)際上是不需要轉(zhuǎn)換函數(shù)的OT類型，因此任何可以處理這些OT類型的東西也應(yīng)該能夠使用CRDT。

或許上邊的一些概念可能一時(shí)間讓人難以理解，所以下面的Counter與Quill兩個(gè)實(shí)例就是介紹了如何使用yjs實(shí)現(xiàn)協(xié)同，究竟如何通過數(shù)據(jù)結(jié)構(gòu)完成協(xié)同的接入工作，當(dāng)然具體的API調(diào)用還是還是需要看yjs的文檔，本文只涉及到最基本的協(xié)同操作，所有的代碼都在https://github.com/WindrunnerMax/Collab中，注意這是個(gè)pnpm的workspace monorepo項(xiàng)目，要注意使用pnpm安裝依賴。

Counter#

首先我們運(yùn)行一個(gè)基礎(chǔ)的協(xié)同實(shí)例Counter，實(shí)現(xiàn)的主要功能是在多個(gè)客戶端可以+1的情況下我們可以維護(hù)同一份計(jì)數(shù)器總數(shù)，該實(shí)例的地址是https://github.com/WindrunnerMax/Collab/tree/master/packages/crdt-counter，首先簡(jiǎn)單看一下目錄結(jié)構(gòu)(tree --dirsfirst -I node_modules):

Copy

crdt-counter ├── public │ ? ├── favicon.ico │ ? └── index.html ├── server │ ? └── index.ts ├── src │ ? ├── client.ts │ ? ├── counter.tsx │ ? └── index.tsx ├── babel.config.js├── package.json├── rollup.config.js├── rollup.server.js└── tsconfig.json

先簡(jiǎn)略說明下各個(gè)文件夾和文件的作用，public存儲(chǔ)了靜態(tài)資源文件，在客戶端打包時(shí)將會(huì)把內(nèi)容移動(dòng)到build文件夾，server文件夾中存儲(chǔ)了CRDT服務(wù)端的實(shí)現(xiàn)，在運(yùn)行時(shí)同樣會(huì)編譯為js文件放置于build文件夾下，src文件夾是客戶端的代碼，主要是視圖與CRDT客戶端的實(shí)現(xiàn)，babel.config.js是babel的配置信息，rollup.config.js是打包客戶端的配置文件，rollup.server.js是打包服務(wù)端的配置文件，package.json與tsconfig.json大家都懂，就不贅述了。

在前邊CRDT協(xié)同算法實(shí)現(xiàn)一文中，我們經(jīng)常提到的就是無(wú)需中央服務(wù)器的分布式協(xié)同，那么在這個(gè)例子中我們就來(lái)實(shí)現(xiàn)一個(gè)peer-to-peer的實(shí)例。yjs提供了一個(gè)y-webrtc的信令服務(wù)器，甚至還有公共的信令服務(wù)器可以用，當(dāng)然可能因?yàn)榫W(wǎng)絡(luò)的關(guān)系這個(gè)公共的信令服務(wù)器在國(guó)內(nèi)不是很適用。在繼續(xù)完成協(xié)同之前，我們還需要了解一下WebRTC以及信令的相關(guān)概念。

WebRTC是一種實(shí)時(shí)通信技術(shù)，重點(diǎn)在于可以點(diǎn)對(duì)點(diǎn)即P2P通信，其允許瀏覽器和應(yīng)用程序直接在互聯(lián)網(wǎng)上傳輸音頻、視頻和數(shù)據(jù)流，無(wú)需使用中間服務(wù)器進(jìn)行中轉(zhuǎn)。WebRTC利用瀏覽器內(nèi)置的標(biāo)準(zhǔn)API和協(xié)議來(lái)提供這些功能，并且支持多種編解碼器和平臺(tái)，WebRTC可以用于開發(fā)各種實(shí)時(shí)通信應(yīng)用，例如在線會(huì)議、遠(yuǎn)程協(xié)作、實(shí)時(shí)廣播、在線游戲和IoT應(yīng)用等。但是在多級(jí)NAT網(wǎng)絡(luò)環(huán)境下，P2P連接可能會(huì)受到限制，簡(jiǎn)單來(lái)說就是一臺(tái)設(shè)備無(wú)法直接發(fā)現(xiàn)另一臺(tái)設(shè)備，自然也就沒有辦法進(jìn)行P2P通信，這時(shí)需要使用特殊的技術(shù)來(lái)繞過NAT并建立P2P連接。

NAT Network Address Translation網(wǎng)絡(luò)地址轉(zhuǎn)換是一種在IP網(wǎng)絡(luò)中廣泛使用的技術(shù)，主要是將一個(gè)IP地址轉(zhuǎn)換為另一個(gè)IP地址，具體來(lái)說其工作原理是將一個(gè)私有IP地址(如在家庭網(wǎng)絡(luò)或企業(yè)內(nèi)部網(wǎng)絡(luò)中使用的地址)映射到一個(gè)公共IP地址(如互聯(lián)網(wǎng)上的IP地址)。當(dāng)一個(gè)設(shè)備從私有網(wǎng)絡(luò)向公共網(wǎng)絡(luò)發(fā)送數(shù)據(jù)包時(shí)，NAT設(shè)備會(huì)將源IP地址從私有地址轉(zhuǎn)換為公共地址，并且在返回?cái)?shù)據(jù)包時(shí)將目標(biāo)IP地址從公共地址轉(zhuǎn)換為私有地址。NAT可以通過多種方式實(shí)現(xiàn)，例如靜態(tài)NAT、動(dòng)態(tài)NAT和端口地址轉(zhuǎn)換PAT等，靜態(tài)NAT將一個(gè)私有IP地址映射到一個(gè)公共IP地址，而動(dòng)態(tài)NAT則動(dòng)態(tài)地為每個(gè)私有地址分配一個(gè)公共地址，PAT是一種特殊的動(dòng)態(tài)NAT，在將私有IP地址轉(zhuǎn)換為公共IP地址時(shí)，還會(huì)將源端口號(hào)或目標(biāo)端口號(hào)轉(zhuǎn)換為不同的端口號(hào)，以支持多個(gè)設(shè)備使用同一個(gè)公共IP地址。NAT最初是為了解決IPv4地址空間的短缺而設(shè)計(jì)的，后來(lái)也為提高網(wǎng)絡(luò)安全性并簡(jiǎn)化網(wǎng)絡(luò)管理提供了基礎(chǔ)。

在互聯(lián)網(wǎng)上大多數(shù)設(shè)備都是通過路由器或防火墻連接到網(wǎng)絡(luò)的，這些設(shè)備通常使用網(wǎng)絡(luò)地址轉(zhuǎn)換NAT將內(nèi)部IP地址映射到一個(gè)公共的IP地址上，這個(gè)公共IP地址可以被其他設(shè)備用來(lái)訪問，但是這些設(shè)備內(nèi)部的IP地址是隱藏的，其他的設(shè)備不能直接通過它們的內(nèi)部IP地址建立P2P連接。因此，直接進(jìn)行P2P連接可能會(huì)受到網(wǎng)絡(luò)地址轉(zhuǎn)換NAT的限制，導(dǎo)致連接無(wú)法建立。為了解決這個(gè)問題，需要使用一些技術(shù)來(lái)繞過NAT并建立P2P連接。另外，P2P連接也需要一些控制和協(xié)調(diào)機(jī)制，以確保連接的可靠性和安全性。

信令可以用來(lái)解決多級(jí)NAT環(huán)境下的P2P連接問題，當(dāng)兩個(gè)設(shè)備嘗試建立P2P連接時(shí)，可以使用信令服務(wù)器來(lái)交換網(wǎng)絡(luò)信息，例如IP地址、端口和協(xié)議類型等，以便設(shè)備之間可以彼此發(fā)現(xiàn)并建立連接。當(dāng)然信令服務(wù)器并不是繞過NAT的唯一解決方案，STUN、TURN和ICE等技術(shù)也可以幫助解決這個(gè)問題。信令服務(wù)器的主要作用是協(xié)調(diào)不同設(shè)備之間的連接，以確保設(shè)備可以正確地發(fā)現(xiàn)和通信。在實(shí)際應(yīng)用中，通常需要同時(shí)使用多種技術(shù)和工具來(lái)解決多級(jí)NAT環(huán)境下的P2P連接問題。

那么回到WebRTC，我們即使是使用了P2P的技術(shù)，但是不可避免的需要一個(gè)信令服務(wù)器來(lái)交換WebRTC會(huì)話描述和控制信息。當(dāng)然這些信息不包括實(shí)際通信的數(shù)據(jù)流本身，而是用于描述和控制這些流的方式和參數(shù)，這些數(shù)據(jù)流本身是通過對(duì)等連接在兩個(gè)瀏覽器之間直接傳輸?shù)?。主要?shù)據(jù)流的通信不經(jīng)過中央服務(wù)器，這就使得WebRTC有著低延遲和高帶寬等優(yōu)點(diǎn)，但是同樣的因?yàn)槊總€(gè)對(duì)等點(diǎn)相互連接，不適合單個(gè)文檔上的大量協(xié)作者。

接下來(lái)我們要進(jìn)行數(shù)據(jù)結(jié)構(gòu)的設(shè)計(jì)，目前在yjs中是沒有Y.Number這個(gè)數(shù)據(jù)結(jié)構(gòu)的，也就是說yjs沒有自增自減的操作，這點(diǎn)就與前邊OT實(shí)例不一樣了，所以在這里我們需要設(shè)計(jì)數(shù)據(jù)結(jié)構(gòu)。網(wǎng)絡(luò)是不可靠的，我們不能夠在本地模擬+1的操作，就是說本地先取得值，然后進(jìn)行+1操作之后再把值推到其他的客戶端上，這樣的設(shè)計(jì)雖然在本地測(cè)試應(yīng)該是可行的，但是由于網(wǎng)絡(luò)不可靠，我們不能保證本地取值的時(shí)候獲得的是最新的值，所以這個(gè)方案是不可靠的。

那么我們思考幾種方案來(lái)實(shí)現(xiàn)這一點(diǎn)，有一種可行的方案是類似于我們之前介紹的CRDT數(shù)據(jù)結(jié)構(gòu)，我們可以構(gòu)造一個(gè)集合Y.Array，當(dāng)我們點(diǎn)+1的時(shí)候，就向集合中push一個(gè)新的值，這樣再取和的時(shí)候直接取集合長(zhǎng)度即可。

Copy

Y.Array: [] => +1 => [1] => +1 => [1, 1] => ... Counter: [1, 1].size = N

另一種方案是使用Y.Map來(lái)完成，當(dāng)用戶加入我們的P2P組的時(shí)候，我們通過其身份信息為其分配一個(gè)id，然后這個(gè)id只記錄與自增自己的值，也就是說當(dāng)某個(gè)客戶端點(diǎn)擊+1的時(shí)候，操作的只有其id對(duì)應(yīng)的數(shù)，而不能影響組網(wǎng)內(nèi)其他的用戶的值。

Copy

Y.Map: {} => +1 => {"id": 1} => +1 => {"id": 2} => ... Counter: Object.values({"id": 2}).reduce((a, b) => a + b) = N

在這里我們使用的是Y.Map的方案，畢竟如果是Y.Array的話占用資源會(huì)是比較大的，當(dāng)然因?yàn)閷?shí)例中并沒有身份信息，每次進(jìn)入的時(shí)候都是會(huì)隨機(jī)分配id的，當(dāng)然這不會(huì)影響到我們的Counter。此外還有比較重要的一點(diǎn)是，因?yàn)槲覀兪侵苯舆M(jìn)行P2P通信的，當(dāng)所有的設(shè)備都離線的時(shí)候，由于沒有設(shè)計(jì)實(shí)際的數(shù)據(jù)存儲(chǔ)機(jī)制，所以數(shù)據(jù)會(huì)丟失，這點(diǎn)也是需要注意的。

接下來(lái)我們看看代碼的實(shí)現(xiàn)，首先我們來(lái)看看服務(wù)端，這里主要實(shí)現(xiàn)是調(diào)用了一下y-webrtc-signaling來(lái)啟動(dòng)一個(gè)信令服務(wù)器，這是y-webrtc給予的開箱即用的功能，也可以基于這些內(nèi)容進(jìn)行改寫，不過因?yàn)槭切帕罘?wù)器，除非有著很高的穩(wěn)定性、定制化等要求，否則直接當(dāng)作開箱即用的信令服務(wù)器就好。后邊主要是使用了express啟動(dòng)了一個(gè)靜態(tài)資源服務(wù)器，因?yàn)橹苯釉跒g覽器打開文件的file協(xié)議有很多的安全限制，所以需要一個(gè)HTTP Server。

Copy

import { exec } from "child_process";import express from "express";// https://github.com/yjs/y-webrtc/blob/master/bin/server.jsexec("PORT=3001 npx y-webrtc-signaling", (err, stdout, stderr) => { // 調(diào)用`y-webrtc-signaling` ?console.log(stdout, stderr); });const app = express(); // 實(shí)例化`express`app.use(express.static("build")); // 客戶端打包過后的靜態(tài)資源路徑app.listen(3000);console.log("Listening on http://localhost:3000");

在客戶端方面主要是定義了一個(gè)定義了一個(gè)共用的鏈接，通過id來(lái)加入我們的P2P組，并且還有密碼的保護(hù)，這里需要鏈接的信令服務(wù)器也就是上邊啟動(dòng)的y-webrtc的3001端口的信令服務(wù)。之后我們通過observe定義的Y.Map數(shù)據(jù)結(jié)構(gòu)的變化來(lái)執(zhí)行回調(diào)，在這里實(shí)際上就是將回調(diào)過后的整個(gè)Map數(shù)據(jù)傳回回調(diào)函數(shù)，然后在視圖層進(jìn)行Counter的計(jì)算，這里還有一個(gè)transaction.origin判斷是為了防止我們本地的調(diào)用觸發(fā)回調(diào)。最后我們定義了一個(gè)increase函數(shù)，在這里我們通過transact作為事務(wù)來(lái)執(zhí)行set操作，因?yàn)槲覀冎暗脑O(shè)計(jì)只會(huì)處理我們當(dāng)前客戶端對(duì)應(yīng)的id的那個(gè)值，本地的值是可信的，直接自增即可，transact最后一個(gè)參數(shù)也就是上邊提到了的transaction.origin，可以用來(lái)判斷事件的來(lái)源。

Copy

import { Doc, Map as YMap } from "yjs";import { WebrtcProvider } from "y-webrtc";const getRandomId = () => Math.floor(Math.random() * 10000).toString();export type ClientCallback = (record: Record<string, number>) => void;class Connection { ?private doc: Doc; ?private map: YMap<number>; ?public id: string = getRandomId(); // 當(dāng)前客戶端生成的唯一`id` ?public counter = 0; // 當(dāng)前客戶端的初始值 ?constructor() { ? ?const doc = new Doc(); ? ?new WebrtcProvider("crdt-example", doc, { // `P2P`組名稱 // `Y.Doc`實(shí)例 ? ? ?password: "room-password", // `P2P`組密碼 ? ? ?signaling: ["ws://localhost:3001"], // 信令服務(wù)器 ? ?}); ? ?const yMapDoc = doc.getMap<number>("counter"); // 獲取數(shù)據(jù)結(jié)構(gòu) ? ?this.doc = doc; ? ?this.map = yMapDoc; ?} ?bind(cb: ClientCallback) { ? ?this.map.observe(event => { // 監(jiān)聽數(shù)據(jù)結(jié)構(gòu)變化 // 如果是多層嵌套需要`observeDeep` ? ? ?if (event.transaction.origin !== this) { // 防止本地修改時(shí)觸發(fā) ? ? ? ?const record = [...this.map.entries()].reduce( // 獲取`Y.Map`定義中的所有數(shù)據(jù) ? ? ? ? ?(cur, [key, value]) => ({ ...cur, [key]: value }), ? ? ? ? ?{} as Record<string, number> ? ? ? ?); ? ? ? ?cb(record); // 執(zhí)行回調(diào) ? ? ?} ? ?}); ?} ?public increase() { ? ?this.doc.transact(() => { // 事務(wù) ? ? ?this.map.set(this.id, ++this.counter); // 自增本地`id`對(duì)應(yīng)的值 ? ?}, this); // 來(lái)源 ?} }export default new Connection();

Quill#

在運(yùn)行富文本的實(shí)例Quill之前，我們不妨先來(lái)簡(jiǎn)單討論一下是如何在富文本上應(yīng)用的CRDT，在前文CRDT協(xié)同算法中主要討論的是分布式與CRDT的原理，并沒有涉及具體的富文本該如何設(shè)計(jì)數(shù)據(jù)結(jié)構(gòu)，那么在這里我們簡(jiǎn)單討論下yjs在富文本上應(yīng)用CRDT的設(shè)計(jì)。看之前描述那一節(jié)的時(shí)候我們可能會(huì)產(chǎn)生一些有趣的想法，或許我們可以這么來(lái)做，可以通過底層來(lái)實(shí)現(xiàn)OT，之后在上層封裝一層數(shù)據(jù)結(jié)構(gòu)供外部使用的方式，從而對(duì)外看起來(lái)像是CRDT。當(dāng)然原理上是不會(huì)這么做的，因?yàn)檫@樣失去了擁抱CRDT的意義，可能會(huì)有部分借鑒實(shí)現(xiàn)的思路，但是不會(huì)直接這么做的。

首先我們可以回憶一下CRDT在集合這個(gè)數(shù)據(jù)結(jié)構(gòu)上的設(shè)計(jì)，我們主要考慮到了集合的添加和刪除如何完整的保證交換律、結(jié)合律、冪等律，那么現(xiàn)在在富文本的實(shí)現(xiàn)上，我們不僅需要考慮到插入和刪除，需要考慮到順序的問題，并且我們還需要保證CCI，即最終一致性、因果一致性、意圖一致性，當(dāng)然還需要考慮到Undo/Redo、光標(biāo)同步等相關(guān)的問題。

那么我們首先來(lái)看看如何保證插入數(shù)據(jù)的順序，對(duì)于OT而言是通過索引得知用戶要操作的位置，并且通過變換來(lái)確保最終一致性，那么CRDT是不需要這么做的，上邊也提到過完全靠OT的話可能就失去了擁抱CRDT的意義，那么如何確保要插入的位置正確呢，CRDT不靠索引的話就需要靠數(shù)據(jù)結(jié)構(gòu)來(lái)完成這點(diǎn)，我們可以通過相對(duì)位置來(lái)完成，例如我們目前有AB字符串，此時(shí)在中間插入了C字符，那么這個(gè)字符就需要被標(biāo)記為在A之后，在B之前，那么很顯然，我們需要為每個(gè)字符都分配唯一的id，否則我們是無(wú)法做到這一點(diǎn)的，當(dāng)然這塊實(shí)際上還有優(yōu)化空間，在這里就先不談這點(diǎn)，那么由此我們通過相對(duì)位置保證了插入的順序。

接下來(lái)我們?cè)倏纯磩h除的問題，在前文的Observed-Remove Set集合數(shù)據(jù)結(jié)構(gòu)中我們是可以真正的進(jìn)行刪除操作的，而在這里由于我們是通過相對(duì)位置來(lái)實(shí)現(xiàn)完整的順序，所以實(shí)際上我們是不能夠真正地將我們標(biāo)記的Item進(jìn)行刪除的，Item可以理解為插入的字符，也就是所謂的軟刪除。舉個(gè)例子，目前我們有AB字符串，其中一個(gè)客戶端刪除了B，另一個(gè)客戶端同時(shí)在A與B之間增加了C，那么此時(shí)這兩個(gè)Op同步到了第三個(gè)客戶端，那么假如增加了C這個(gè)操作先到并且執(zhí)行了，再刪除了B，那么沒有問題，可是假設(shè)我們先刪除了B，再增加了C，那么這個(gè)C我們就不能夠找到他要插入的位置，因?yàn)?code>B已經(jīng)被刪除了，我們是要在A與B之間去插入C的，那么這樣這個(gè)操作就無(wú)法執(zhí)行下去了，由此這樣其實(shí)就導(dǎo)致了操作不滿足交換律，那么這就不能真的作為CRDT的數(shù)據(jù)結(jié)構(gòu)設(shè)計(jì)了。其實(shí)我們可能會(huì)想，為什么需要兩個(gè)位置來(lái)保證插入的字符位置，完全可以用B的左側(cè)或者A的右側(cè)來(lái)完成，實(shí)際上思考一下這是同樣的問題，多個(gè)客戶端來(lái)操作的話假如一個(gè)刪除了A另一個(gè)刪除了B，那么便無(wú)論如何也找不到插入的位置了，這是不滿足交換律和結(jié)合律的操作，就不能作為CRDT的實(shí)現(xiàn)了。因此為了沖突的解決yjs并沒有真正的刪除Item，而是采用了標(biāo)記的形式，即刪除的Item會(huì)被加入一個(gè)deleted標(biāo)記，那么不刪除會(huì)造成一個(gè)明顯的問題，空間的占用會(huì)無(wú)限增長(zhǎng)，因此yjs引入了墓碑機(jī)制，當(dāng)確認(rèn)了內(nèi)容不會(huì)再被干涉之后，將對(duì)象的內(nèi)容替換為空的墓碑對(duì)象。

上邊也提到了沖突的問題，很明顯在設(shè)計(jì)上是存在沖突的問題的，因?yàn)?code>CRDT實(shí)際上并不是完全為了協(xié)同編輯的場(chǎng)景而專門設(shè)計(jì)的，其主要是為了解決分布式場(chǎng)景中的一致性問題，所以在應(yīng)用到協(xié)同編輯的場(chǎng)景中，不可避免地會(huì)出現(xiàn)沖突的問題，實(shí)際上這個(gè)沖突主要是為了集合順序的引入而導(dǎo)致的，要是不關(guān)心順序，那么自然就不會(huì)出現(xiàn)沖突問題了。那么為了使數(shù)據(jù)能夠滿足三律，在前文我們引入了一個(gè)偏序的概念，但是在協(xié)同編輯設(shè)計(jì)中，使用偏序不能夠保證數(shù)據(jù)同步的正確性和一致性，因?yàn)槠錈o(wú)法處理一些關(guān)鍵的沖突情況，舉一個(gè)簡(jiǎn)單的例子，假設(shè)我們此時(shí)有AB字符串，如果一個(gè)客戶端在AB中加入了C，另一個(gè)加入了D，那么究竟誰(shuí)在前呢，所以我們需要引入全序的方法，即任意兩個(gè)Item都是可以比較的。那么很明顯的，如果我們?yōu)槊總€(gè)Item附加上時(shí)間戳的元信息，便可以引入全序了，但是實(shí)際上由于不同的客戶端可能具有不同的時(shí)鐘偏差，網(wǎng)絡(luò)延遲和時(shí)鐘不同步等問題也可能導(dǎo)致時(shí)間戳不可靠。那么相比之下，邏輯時(shí)鐘或者邏輯時(shí)間戳可以使用更簡(jiǎn)單和可靠的方式來(lái)維護(hù)事件的順序:

• 每次發(fā)生本地事件時(shí)，clock = clocl + 1。

• 每次接收到遠(yuǎn)程事件時(shí)，clock = max(clock, remoteClock) + 1。

看起來(lái)依舊會(huì)有發(fā)生沖突的可能，那么我們可以再引入一個(gè)客戶端的唯一id，也就是clientID。這種機(jī)制看似簡(jiǎn)單，但實(shí)際上使我們獲得了數(shù)學(xué)上性質(zhì)良好的全序結(jié)構(gòu)，這意味著我們可以在任意兩個(gè)Item之間對(duì)比獲得邏輯上的先后關(guān)系，這對(duì)保證CRDT算法的正確性相當(dāng)重要。此外，通過這種方式我們也可以保證因果一致性，假如此時(shí)我們有兩個(gè)操作a、b如果有因果關(guān)系，那么a.clock一定大于b.clock，這樣的得到的順序一定是滿足因果關(guān)系的，當(dāng)然如果沒有因果關(guān)系，就可以取任意的順序執(zhí)行了。舉個(gè)例子，我們有三個(gè)客戶端A、B、C以及字符串SE，A在SE中間添加了a字符，此時(shí)這個(gè)操作同步到了B，B將a字符給刪除了，假設(shè)此時(shí)C先收到了B的刪除操作，因?yàn)檫@個(gè)操作依賴于A的操作，需要進(jìn)行因果依賴關(guān)系的檢查，這個(gè)操作的邏輯時(shí)鐘和位移大于C本地文檔中已經(jīng)應(yīng)用的操作的邏輯時(shí)鐘和位移，需要等待先前的操作被應(yīng)用后再應(yīng)用這個(gè)操作，當(dāng)然這并不是在yjs中的實(shí)現(xiàn)，因?yàn)?code>yjs不會(huì)存在真正的刪除操作，并且在刪除操作的時(shí)候?qū)嶋H上并不會(huì)導(dǎo)致時(shí)鐘的增加，只是增加一個(gè)標(biāo)記，上邊這個(gè)例子其實(shí)可以換個(gè)說法，兩個(gè)相同的插入操作，因?yàn)槲覀兪窍鄬?duì)位置，所以后一個(gè)插入操作是依賴前一個(gè)插入操作的，因此就需要因果檢查，其實(shí)這也是件有意思的事情，當(dāng)收到在同一個(gè)位置編輯的不同客戶端操作時(shí)候，如果時(shí)鐘相同就是沖突操作，不相同就是因果關(guān)系。

那么由此我們通過CRDT數(shù)據(jù)結(jié)構(gòu)與算法設(shè)計(jì)解決了最終一致性和因果一致性，對(duì)于意圖一致性的問題，當(dāng)不存在沖突的時(shí)候我們是能夠保證意圖的，即插入文檔的Item的順序，在沖突的時(shí)候我們實(shí)際上會(huì)比較clientID決定究竟誰(shuí)在前在后，其實(shí)實(shí)際上無(wú)論誰(shuí)在前還是在后都可以認(rèn)為是一種烏龍，我們?cè)跊_突的時(shí)候只保證最終一致性，對(duì)于意圖一致性則需要做額外的設(shè)計(jì)才可以實(shí)現(xiàn)，在這里就不做過多探討了。實(shí)際上yjs還有大量的設(shè)計(jì)與優(yōu)化操作，以及基于YATA的沖突解決算法等，比如通過雙向鏈表來(lái)保存文檔結(jié)構(gòu)順序，通過Map為每個(gè)客戶端保存的扁平的?Item數(shù)組，優(yōu)化本地插入的速度而設(shè)計(jì)的緩存機(jī)制(鏈表的查找O(N)與跟隨光標(biāo)的位置緩存)，傾向于State-based的刪除，Undo/Redo，光標(biāo)同步，壓縮數(shù)據(jù)網(wǎng)絡(luò)傳輸?shù)鹊?，還是很值得研究的。

我們?cè)倩氐礁晃谋镜膶?shí)例Quill中，實(shí)現(xiàn)的主要功能是在quill富文本編輯器中接入?yún)f(xié)同，并支持編輯光標(biāo)的同步，該實(shí)例的地址是https://github.com/WindrunnerMax/Collab/tree/master/packages/crdt-quill，首先簡(jiǎn)單看一下目錄結(jié)構(gòu)(tree --dirsfirst -I node_modules):

Copy

crdt-quill ├── public │ ? └── favicon.ico ├── server │ ? └── index.ts ├── src │ ? ├── client.ts │ ? ├── index.css │ ? ├── index.ts │ ? └── quill.ts ├── package.json├── rollup.config.js├── rollup.server.js└── tsconfig.json

依舊簡(jiǎn)略說明下各個(gè)文件夾和文件的作用，public存儲(chǔ)了靜態(tài)資源文件，在客戶端打包時(shí)將會(huì)把內(nèi)容移動(dòng)到build文件夾，server文件夾中存儲(chǔ)了CRDT服務(wù)端的實(shí)現(xiàn)，在運(yùn)行時(shí)同樣會(huì)編譯為js文件放置于build文件夾下，src文件夾是客戶端的代碼，主要是視圖與CRDT客戶端的實(shí)現(xiàn)，rollup.config.js是打包客戶端的配置文件，rollup.server.js是打包服務(wù)端的配置文件，package.json與tsconfig.json大家都懂，就不贅述了。

quill的數(shù)據(jù)結(jié)構(gòu)并不是JSON而是Delta，Delta是通過retain、insert、delete三個(gè)操作完成整篇文檔的描述與操作，我們?cè)囅胍幌旅枋鲆欢巫址牟僮餍枰裁矗遣皇峭ㄟ^這三種操作就能夠完全覆蓋了，所以通過Delta來(lái)描述文本增刪改是完全可行的，而且12年quill的開源可以說是富文本發(fā)展的一個(gè)里程碑，于是yjs是直接原生支持Delta數(shù)據(jù)結(jié)構(gòu)的。

接下來(lái)我們看看來(lái)看看服務(wù)端，這里主要實(shí)現(xiàn)是調(diào)用了一下y-websocket來(lái)啟動(dòng)一個(gè)websocket服務(wù)器，這是y-websocket給予的開箱即用的功能，也可以基于這些內(nèi)容進(jìn)行改寫，yjs還提供了y-mongodb-provider等服務(wù)端服務(wù)可以使用。后邊主要是使用了express啟動(dòng)了一個(gè)靜態(tài)資源服務(wù)器，因?yàn)橹苯釉跒g覽器打開文件的file協(xié)議有很多的安全限制，所以需要一個(gè)HTTP Server。

Copy

import { exec } from "child_process";import express from "express";// https://github.com/yjs/y-websocket/blob/master/bin/server.jsexec("PORT=3001 npx y-websocket", (err, stdout, stderr) => { // 調(diào)用`y-websocket` ?console.log(stdout, stderr); });const app = express(); // 實(shí)例化`express`app.use(express.static("build")); // 客戶端打包過后的靜態(tài)資源路徑app.use(express.static("node_modules/quill/dist")); // `quill`靜態(tài)資源路徑app.listen(3000);console.log("Listening on http://localhost:3000");

在客戶端方面主要是定義了一個(gè)定義了一個(gè)共用的鏈接，通過crdt-quill作為RoomName進(jìn)入組，這里需要鏈接的websocket服務(wù)器也就是上邊啟動(dòng)的y-websocket的3001端口的服務(wù)。之后我們定義了頂層的數(shù)據(jù)結(jié)構(gòu)為YText數(shù)據(jù)結(jié)構(gòu)的變化來(lái)執(zhí)行回調(diào)，并且將一些信息暴露了出去，doc就是這需要使用的yjs實(shí)例，type是我們定義的頂層數(shù)據(jù)結(jié)構(gòu)，awareness意為感知，只要是用來(lái)完成實(shí)時(shí)數(shù)據(jù)同步，在這里是用來(lái)同步光標(biāo)選區(qū)。

Copy

import { Doc, Text as YText } from "yjs";import { WebsocketProvider } from "y-websocket";class Connection { ?public doc: Doc; // `yjs`實(shí)例 ?public type: YText; // 頂層數(shù)據(jù)結(jié)構(gòu) ?private connection: WebsocketProvider; // `WebSocket`鏈接 ?public awareness: WebsocketProvider["awareness"]; // 數(shù)據(jù)實(shí)時(shí)同步 ?constructor() { ? ?const doc = new Doc(); // 實(shí)例化 ? ?const provider = new WebsocketProvider("ws://localhost:3001", "crdt-quill", doc); // 鏈接`WebSocket`服務(wù)器 ? ?provider.on("status", (e: { status: string }) => { ? ? ?console.log("WebSocket", e.status); // 鏈接狀態(tài) ? ?}); ? ?this.doc = doc; // `yjs`實(shí)例 ? ?this.type = doc.getText("quill"); // 獲取頂層數(shù)據(jù)結(jié)構(gòu) ? ?this.connection = provider; // 鏈接 ? ?this.awareness = provider.awareness; // 數(shù)據(jù)實(shí)時(shí)同步 ?} ?reconnect() { ? ?this.connection.connect(); // 重連 ?} ?disconnect() { ? ?this.connection.disconnect(); // 斷線 ?} }export default new Connection();

在客戶端主要分為了兩部分，分別是實(shí)例化quill的實(shí)例，以及quill與yjs客戶端通信的實(shí)現(xiàn)。在quill的實(shí)現(xiàn)中主要是將quill實(shí)例化，注冊(cè)光標(biāo)的插件，隨機(jī)生成id的方法，通過id獲取隨機(jī)顏色的方法，以及光標(biāo)同步的位置轉(zhuǎn)換。在quill與yjs客戶端通信的實(shí)現(xiàn)中，主要是完成了對(duì)于quill與doc的事件監(jiān)聽，主要是遠(yuǎn)程數(shù)據(jù)變更的回調(diào)，本地?cái)?shù)據(jù)變化的回調(diào)，光標(biāo)同步事件感知的回調(diào)。

Copy

import Quill from "quill";import QuillCursors from "quill-cursors";import tinyColor from "tinycolor2";import { Awareness } from "y-protocols/awareness.js";import { ?Doc, ?Text as YText, ?createAbsolutePositionFromRelativePosition, ?createRelativePositionFromJSON, } from "yjs";export type { Sources } from "quill";Quill.register("modules/cursors", QuillCursors); // 注冊(cè)光標(biāo)插件export default new Quill("#editor", { // 實(shí)例化`quill` ?theme: "snow", ?modules: { cursors: true }, });const COLOR_MAP: Record<string, string> = {}; // `id => color`export const getRandomId = () => Math.floor(Math.random() * 10000).toString(); // 隨機(jī)生成用戶`id`export const getCursorColor = (id: string) => { // 根據(jù)`id`獲取顏色 ?COLOR_MAP[id] = COLOR_MAP[id] || tinyColor.random().toHexString(); ?return COLOR_MAP[id]; };export const updateCursor = ( ?cursor: QuillCursors, ?state: Awareness["states"] extends Map<number, infer I> ? I : never, ?clientId: number, ?doc: Doc, ?type: YText) => { ?try { ? ?// 從`Awareness`中取得狀態(tài) ? ?if (state && state.cursor && clientId !== doc.clientID) { ? ? ?const user = state.user || {}; ? ? ?const color = user.color || "#aaa"; ? ? ?const name = user.name || `User: ${clientId}`; ? ? ?// 根據(jù)`clientId`創(chuàng)建光標(biāo) ? ? ?cursor.createCursor(clientId.toString(), name, color); ? ? ?// 相對(duì)位置轉(zhuǎn)換為絕對(duì)位置 // 選區(qū)為`focus --- anchor` ? ? ?const focus = createAbsolutePositionFromRelativePosition( ? ? ? ?createRelativePositionFromJSON(state.cursor.focus), ? ? ? ?doc ? ? ?); ? ? ?const anchor = createAbsolutePositionFromRelativePosition( ? ? ? ?createRelativePositionFromJSON(state.cursor.anchor), ? ? ? ?doc ? ? ?); ? ? ?if (focus && anchor && focus.type === type) { ? ? ? ?// 移動(dòng)光標(biāo)位置 ? ? ? ?cursor.moveCursor(clientId.toString(), { ? ? ? ? ?index: focus.index, ? ? ? ? ?length: anchor.index - focus.index, ? ? ? ?}); ? ? ?} ? ?} else { ? ? ?// 根據(jù)`clientId`移除光標(biāo) ? ? ?cursor.removeCursor(clientId.toString()); ? ?} ?} catch (err) { ? ?console.error(err); ?} };

Copy

import "./index.css";import quill, { getRandomId, updateCursor, Sources, getCursorColor } from "./quill";import client from "./client";import Delta from "quill-delta";import QuillCursors from "quill-cursors";import { compareRelativePositions, createRelativePositionFromTypeIndex } from "yjs";const userId = getRandomId(); // 本地客戶端的`id` 或者使用`awareness.clientID`const doc = client.doc; // `yjs`實(shí)例const type = client.type; // 頂層類型const cursors = quill.getModule("cursors") as QuillCursors; // `quill`光標(biāo)模塊const awareness = client.awareness; // 實(shí)時(shí)通信感知模塊// 設(shè)置當(dāng)前客戶端的信息 `State`的數(shù)據(jù)結(jié)構(gòu)類似于`Record<string, unknown>`awareness.setLocalStateField("user", { ?name: "User: " + userId, ?color: getCursorColor(userId), });// 頁(yè)面顯示的用戶信息const userNode = document.getElementById("user") as HTMLInputElement; userNode && (userNode.value = "User: " + userId); type.observe(event => { ?// 來(lái)源信息 // 本地`UpdateContents`不應(yīng)該再觸發(fā)`ApplyDelta' ?if (event.transaction.origin !== userId) { ? ?const delta = event.delta; ? ?quill.updateContents(new Delta(delta), "api"); // 應(yīng)用遠(yuǎn)程數(shù)據(jù), 來(lái)源 ?} }); quill.on("editor-change", (_: string, delta: Delta, state: Delta, origin: Sources) => { ?if (delta && delta.ops) { ? ?// 來(lái)源信息 // 本地`ApplyDelta`不應(yīng)該再觸發(fā)`UpdateContents` ? ?if (origin !== "api") { ? ? ?doc.transact(() => { ? ? ? ?type.applyDelta(delta.ops); // 應(yīng)用`Ops`到`yjs` ? ? ?}, userId); // 來(lái)源 ? ?} ?} ?const sel = quill.getSelection(); // 選區(qū) ?const aw = awareness.getLocalState(); // 實(shí)時(shí)通信狀態(tài)數(shù)據(jù) ?if (sel === null) { // 失去焦點(diǎn) ? ?if (awareness.getLocalState() !== null) { ? ? ?awareness.setLocalStateField("cursor", null); // 清除選區(qū)狀態(tài) ? ?} ?} else { ? ?// 卷對(duì)位置轉(zhuǎn)換為相對(duì)位置 // 選區(qū)為`focus --- anchor` ? ?const focus = createRelativePositionFromTypeIndex(type, sel.index); ? ?const anchor = createRelativePositionFromTypeIndex(type, sel.index + sel.length); ? ?if ( ? ? ?!aw || ? ? ?!aw.cursor || ? ? ?!compareRelativePositions(focus, aw.cursor.focus) || ? ? ?!compareRelativePositions(anchor, aw.cursor.anchor) ? ?) { ? ? ?// 選區(qū)位置發(fā)生變化 設(shè)置位置信息 ? ? ?awareness.setLocalStateField("cursor", { focus, anchor }); ? ?} ?} ?// 更新所有光標(biāo)狀態(tài)到本地 ?awareness.getStates().forEach((aw, clientId) => { ? ?updateCursor(cursors, aw, clientId, doc, type); ?}); });// 初始化更新所有遠(yuǎn)程光標(biāo)狀態(tài)到本地awareness.getStates().forEach((state, clientId) => { ?updateCursor(cursors, state, clientId, doc, type); });// 監(jiān)聽遠(yuǎn)程狀態(tài)變化的回調(diào)awareness.on( ?"change", ?({ added, removed, updated }: { added: number[]; removed: number[]; updated: number[] }) => { ? ?const states = awareness.getStates(); ? ?added.forEach(id => { ? ? ?const state = states.get(id); ? ? ?state && updateCursor(cursors, state, id, doc, type); ? ?}); ? ?updated.forEach(id => { ? ? ?const state = states.get(id); ? ? ?state && updateCursor(cursors, state, id, doc, type); ? ?}); ? ?removed.forEach(id => { ? ? ?cursors.removeCursor(id.toString()); ? ?}); ?} );