processflow基于私有云存储的多人实时协作尝试

news2024/12/24 8:48:16

前言

继上一篇基于oneDrive实现多人协作之后，发现Onedrive存在诸多限制之处，比如国内网络环境很差，多人协作还需要processflow用户自己注册oneDrive账号，然后oneDrive系统内进行文件分享，才能拿到分享链接进行多人协作。过程可谓是相当之繁琐。后面便萌生了基于一些开源的分布式文件多版本存储系统代替oneDrive的功能，经历了一个多月的尝试，最终得到了我想要的结果。实现了私有云多版本文件存储服务（这里大家可以参考一下seafile和nextcloud，我这里选则的是seafile，因为是用C写的，性能更好，而且从技术架构角度考虑，前后端分离更适合二次开发，换皮啥的；nextcloud就是太丑了，基于php宇宙第一开发语言写的，接口数据格式相当恶心，全部都是xml，层层封装，和他交互起来太繁琐）。在这基础之上，便开始了processflow的多人协作功能的进化。

技术介绍

主要使用了websocket技术+多版本管理的文件存储服务，websocket主要传递用户的鼠标移动轨迹和事件行为传递作用，使用了广播模式对用户事件进行发布。当用户修改了流程图文件，会将变动的内容通过websocket广播到各个web客户端。内容的存储和变动内容合并则是通过文件存储服务来实现。

websocket通信原理：

多人协作组会话管理：

利用协作文件的uid或hash值作为channeld，每个channel中会管理一个map，map中维护每个协作者的clientId和sessionId。拿到sessionId再去获取内存中的session来与客户端进行通信。

seafile分布式架构设计：

Seafile 的文件版本管理基于 Git 版本控制系统，并结合了一些自定义的特性来满足文件同步和共享的需求。以下是 Seafile 文件版本管理的基本原理：

Git 版本控制系统：Seafile 使用 Git 作为底层的版本控制系统。Git 是一种分布式版本控制系统，它可以跟踪文件的每个修改，并记录这些修改的历史。
文件快照：在 Seafile 中，每当文件发生变化时，会创建一个文件快照。文件快照是文件在特定时间点的副本，用于记录文件的状态和内容。
增量存储：为了节省存储空间和提高效率，Seafile 使用增量存储机制。当文件发生变化时，只有发生变化的部分会被保存为新的快照，而不是完整地复制整个文件。
版本标记：每个文件快照都有一个唯一的版本标记，用于标识该快照的位置和顺序。通过版本标记，可以轻松地查找和恢复特定版本的文件。
差异比较和合并：Seafile 可以比较不同版本的文件快照之间的差异，并根据需要进行合并。这使得多个用户可以同时编辑同一个文件，并将其更改合并到最新版本中。
回滚和恢复：通过版本管理功能，Seafile 允许用户回滚到先前的文件版本或恢复已删除的文件。这对于修复错误、还原误操作或检索旧版本的文件非常有用。
冲突解决：当多个用户同时编辑同一个文件时，可能会发生冲突。Seafile 提供了冲突解决机制，允许用户手动解决冲突或自动合并冲突。

在processflow的版本管理中，主要用到了ETag（实体标签）和时间戳两种保障机制机制，用于标识和比较资源的不同版本。它们的作用如下：
ETag：ETag 是一个由服务器生成的唯一标识符，用于表示资源的特定版本。当客户端请求某个资源时，服务器会将该资源的 ETag 值返回给客户端。客户端可以在后续请求中将该 ETag 值发送给服务器，以便服务器判断资源是否发生了变化。如果资源的 ETag 值与客户端发送的值匹配，则表示资源未发生变化，服务器可以返回一个 304 Not Modified 状态码，告知客户端使用缓存的版本。如果 ETag 值不匹配，则表示资源已经发生了变化，服务器会返回新的资源内容。
时间戳：时间戳是指资源的最后修改时间。当客户端请求某个资源时，服务器会将该资源的最后修改时间返回给客户端。客户端可以在后续请求中将该时间戳发送给服务器，以便服务器判断资源是否发生了变化。如果资源的最后修改时间与客户端发送的时间戳相同，则表示资源未发生变化，服务器可以返回一个 304 Not Modified 状态码，告知客户端使用缓存的版本。如果时间戳不同，则表示资源已经发生了变化，服务器会返回新的资源内容。
这两种机制都可以用于减少网络传输和服务器负载，提高资源访问效率。选择使用哪种机制取决于具体的应用场景和需求。ETag 更加精确，可以识别出资源内容的任何变化，但需要服务器计算和存储额外的标识符。时间戳则更加简单，只需要记录资源的最后修改时间，但无法捕捉到内容没有发生变化但是重新生成了相同内容的情况。

import requests

url = "https://example.com/resource"
headers = {}

# 发送第一次请求，获取资源和相关信息
response = requests.get(url)
etag = response.headers.get("ETag")
last_modified = response.headers.get("Last-Modified")

# 在后续请求中发送 ETag 或时间戳
if etag:
    headers["If-None-Match"] = etag
if last_modified:
    headers["If-Modified-Since"] = last_modified

# 发送带有 ETag 或时间戳的请求
response = requests.get(url, headers=headers)

# 根据服务器返回的状态码处理结果
if response.status_code == 304:
    # 资源未发生变化，使用缓存的版本
    print("Using cached version.")
else:
    # 资源已经发生变化，使用新的版本
    print("Using updated version.")
    print(response.text)