Redux应用多人协做的思路和实现
先上Demo动图，效果以下： javascript
基本思路
因为redux更改数据是dispatch(action)，因此很天然而然想到以action做为基本单位在服务端和客户端进行传送，在客户端和服务端用数组来存放action，那么只要当客户端和服务端的action队列的顺序保持同样，reducer是纯函数的特性能够知道计算获得的state是同样的。前端
一些约定
本文中C1，C2...Cn表示客户端，S表示服务端，a1，a2，a3...an表示aciton，服务端是使用koa + socket.io来编写的（做为一个前端，服务端的知识几乎为0勿喷）。java
总体思路
当客户端发起一个action的时候，服务端接收到这个action，服务端作了3件事：git
- 把action推动栈中
- 把该客户端a1以前的action发送该客户端（相似git push以前有个git pull的过程）
- 将a1发送给其余的客户端
不过上面的思路是比较笼统的，细想会发现许多问题：算法
- 若是a1到达S端的时候，C二、C3还有action在派发怎么处理？
- 若是a1到达S端的时候，C1正在接收其余客户端发送的action怎么处理？
- 若是a1发送以前，C1正在发送前一个action怎么处理？ 后面咱们一一解决。
服务端派发action机制
服务端设立了2个概念：target、member指编辑对象（多是报告、流程图等）和编辑用户，当你要发送两个action：a一、a2的时候，由于网络发送前后的不肯定性，因此应该是先发送a1，而后等待客户端接收到，再发送a2，这样才能在客户端中保证a1和a2的顺序。所以，每一个member会有变量pending表示在不在发送，index表示发送的最新的action在action队列中的索引。redux
当服务端接收到客户端的aciton的时候
this.socket.on('client send action', (action) => {
//目标将action放入队列中，并返回该action的索引
let index = this.target.addAction(action);
if (this.pending) {
this.queue.push({
method: 'sendBeforeActions',
args: [index]
})
} else {
this.sendBeforeActions(index);
}
this.target.receiveAction({
member: this,
index
})
})
复制代码
这就是上面讲的当服务端收到a1的时候作的3件事情。只是这里会去判断该member是否是正在执行发送任务，若是是，那么就将发送a1前面的aciton这个动做存入到一个动做队列中，并告知target，我这个member发送了一个action。数组
sendBeforeActions
sendBeforeActions(refIndex) {
let {actions} = this.getCurrent(refIndex);
actions = actions.slice(0, -1);
this.pending = true;
this.socket.emit('server send before actions', { actions }, () => {
this.pending = false;
this.target.setIndex(this, refIndex);
this.sendProcess();
});
}
复制代码
这个函数接收一个索引，这个索引在上面的代码中是这个member接收到的action在队列中的索引，因此getCurrent(refIndex)指到refIndex为止，还没发送给这个member的全部的action（可能为空），因此要剔除自己后actions.slice(0, -1)发送给客户端。 回调中终止发送状态，设置member最新的action的index，而后执行sendProcess函数去看看，在本身自己发送的过程当中，是否是有后续的动做存入到发送队列中了浏览器
sendProcess() {
if (this.queue.length > 0 && !this.pending) {
let current = this.queue.shift();
let method = this[current.method];
method.apply(this, current.args);
}
}
复制代码
若是你注意到刚才的：缓存
if (this.pending) {
this.queue.push({
method: 'sendBeforeActions',
args: [index]
})
}
复制代码
你就会发现，若是刚才想发送before action的时候这个member在发送其余action，那么会等待这个action发送完后才触发sendProcess去执行这个发送。bash
还要将这个action发送给其余用户
在刚才的代码中
//this指某个member对象
this.target.receiveAction({
member: this,
index
})
复制代码
就是这个触发了其余客户端的发送
//this指某个target对象
receiveAction({member, index}) {
this.members.forEach(m => {
if (m.id !== member.id) {
m.queue.push({
method: 'sendActions',
args: [index]
});
m.sendProcess();
}
})
}
复制代码
若是members中存在发送方的member，那么会将发送动做存入member的发送队列中，执行sendProcess
sendActions
sendActions(refIndex) {
let {actions} = this.getCurrent(refIndex);
if (actions.length) {
this.pending = true;
this.socket.emit('server send actions', {actions}, () => {
this.pending = false;
this.target.setIndex(this, refIndex);
this.sendProcess();
})
}
}
复制代码
这个函数和sendBeforeActions几乎同样，只差要不要剔除最新的action，这样，就保证了服务端的发送action顺序
客户端IO中间件
在客户端中，将io有关的操做都封装在一个中间件中
module.exports = store => next => action => {
if (action.type === 'connection') {
//链接初始化一些事件
return initIo(action.payload)
}
if (action.type === 'disconnection') {
return socket.disconnect(action.payload)
}
if (['@replace/state'].indexOf(action.type.toLowerCase()) === -1 && !action.escapeServer && !action.temporary) {
//将action给定userId、targetId
action = actionCreator(action);
//获得新的action队列，并计算actions，而后更新到state上
let newCacheActions = [...cacheActions, action];
mapActionsToState(newCacheActions);
//发送给服务端
return delieverAction(action);
}
//这样就只容许replace state 的action进入到store里面，这个是我这个思路在实现undo、redo的一个要求，后面会讲到
next();
}
复制代码
一些全局变量
具体做用后面会用到
let cacheActions = []; //action队列，这个和服务端的action队列保持一致
let currentActions = []; //根据cacheActions计算的action
let redoActions = {}; //缓存每一个用户的undo后拿掉的action
let pending = false; //是否在发送请求
let actionsToPend = []; //缓存发送队列
let beforeActions = []; //缓存pull下来的actions
let currentAction = null;//当前发送的action
let user, tid; //用户名和targetId
let initialState; //初始的state
let timeline = []; //缓存state
复制代码
客户端总体思路图
主要讲两个地方：
（1）在computeActions的时候，碰到undo拿掉该用户的最后一个action，并把倒数第二个action提高到最后的缘由是由于假如在该用户倒数第二个action以后还有其余用户的action发生，那么可能其余用户会覆盖掉这个用户action的设定值，那么这个用户undo的时候就没法回到以前的状态了，这时候提高至关因而undo后作了新的action，这个action就是前一次的action。这个算法是有bug的，当一个用户undo的时候，因为咱们会提高他倒数第二的action，这样会致使与这个action冲突的action的修改被覆盖。这个解决冲突的策略有点问题。若是没有提高，那么若是该用户undo的时候，若是他上一个action被其余用户的action覆盖了，那么他就没法undo回去了。这个是个痛点，我还在持续探索中，欢迎大神指教。
（2）在用户pending的时候收到了actions，这个时候至关因而before actions。 下面贴几个主要函数的代码
initIo
function initIo(payload, dispatch) {
user = payload.user;
tid = parseInt(payload.tid, 10);
//初始化socket
let socket = cache.socket = io(location.protocol + '//' + location.host, {
query: {
user: JSON.stringify(user),
tid
}
});
//获取初始数据
socket.on('deliver initial data', (params) => {
...获取初始的state，actions
})
//发送action会等待pull以前的actions
socket.on('server send before actions', (payload, callback) => {
pending = false;
callback && callback();
let {actions} = payload;
actions = [...actions, ...beforeActions, currentAction];
cacheActions = [...cacheActions, ...actions];
if (actions.length > 1) {
//证实有前面的action，须要根据actions从新计算state
mapActionsToState();
}
if (actionsToPend.length) {
let action = actionsToPend.shift();
sendAction(action);
}
})
//接收actions
socket.on('server send actions', (payload, callback) => {
let {actions} = payload;
callback && callback();
if (pending) {
beforeActions = [...beforeActions, ...actions];
} else {
cacheActions = [...cacheActions, ...actions];
mapActionsToState();
}
})
}
复制代码
mapActionsToState
function mapActionsToState(actions) {
actions = actions || cacheActions;
if (actions.length === 0) {
return replaceState(dispatch)(initialState);
}
let {newCurrentActions, newRedoActions} = computeActions(actions);
let {same} = diff(newCurrentActions);
let state = initialState;
if (timeline[same]) {
state = timeline[same];
timeline = timeline.slice(0, same + 1);
}
if (same === -1) {
timeline = [];
}
let differentActions = newCurrentActions.slice(same + 1);
differentActions.forEach(action => {
state = store.reducer(state, action);
timeline.push(state);
});
currentActions = newCurrentActions;
redoActions = newRedoActions;
store.canUndo = () => currentActions.some(action => action.userId === user.id);
store.canRedo = () => !!(redoActions[user.id] || []).length;
return replaceState(dispatch)(state);
}
复制代码
computeActions
function computeActions(actions) {
let newCurrentActions = [];
let newRedoActions = {};
actions.forEach(action => {
let type = action.type.toLowerCase();
newRedoActions[action.userId] = newRedoActions[action.userId] || [];
if (type !== 'redo' && type !== 'undo') {
newCurrentActions.push(action);
newRedoActions[action.userId] = [];
}
if (type === 'undo') {
let indexes = [];
for (let i = newCurrentActions.length - 1; i >= 0; i--) {
if (newCurrentActions[i].userId === action.userId) {
indexes.push(i);
}
if (indexes.length === 2) {
break;
}
}
if (indexes.length > 0) {
let redo = newCurrentActions.splice(indexes[0], 1)[0];
newRedoActions[action.userId].push(redo);
}
if (indexes.length > 1) {
let temp = newCurrentActions.splice(indexes[1], 1);
newCurrentActions.push(temp[0]);
}
}
if (type === 'redo') {
let redo = newRedoActions[action.userId].pop();
newCurrentActions.push(redo);
}
});
return {
newCurrentActions,
newRedoActions
}
}
复制代码
diff
function diff(newCurrentActions) {
let same = -1;
newCurrentActions.some((action, index) => {
let currentAction = currentActions[index];
if (currentAction && action.id === currentAction.id) {
same = index;
return false;
}
return true;
});
return {
same
}
}
复制代码
结束语
讲了一堆，不知道有没有将本身的思路讲清楚，本身的demo也运行了起来，测试只用了两个浏览器来模拟测试，总感受一些并发延时出现还会有bug，后面会持续优化这个想法，添加一些自动化测试来验证，另外，对于服务端的存储也还没考虑，先在只在内存中跑，会思考保存方案。但愿对这方面有兴趣的大神能够指导一下
- 1. FreeSWITCH实现多人来电思路
- 2. React-Redux的用法思路
- 3. Android github 快速实现多人协做
- 4. axure实现多人协做画图
- 5. Redux和React-Redux的实现（一）：Redux的实现和context
- 6. redux: redux以及中间件redux-saga思路的代码实现模板
- 7. gitlab实现多人协同
- 8. 应用换肤实现思路总结
- 9. Redux和React-Redux的实现（二）：Provider组件和connect的实现
- 10. github 多人协做
- 更多相关文章...
- • 现实生活中的 XML - XML 教程
- • 使用Redis和Lua的原子性实现抢红包功能 - 红包项目实战
- • TiDB 在摩拜单车在线数据业务的应用和实践
- • 适用于PHP初学者的学习线路和建议
-
每一个你不满意的现在，都有一个你没有努力的曾经。