如何确保异构系统间数据的一致性交流？
- 内容介绍
- 文章标签
- 相关推荐
本文共计3823个文字，预计阅读时间需要16分钟。
之前忙碌于一个多方合作的项目，异构系统间维护数据一致性有时是个头疼的问题。今天，我从背景、问题、解决方案、技术方案等几个方面浅析一下。
之前忙活过一个多方合作的项目，异构系统间维护数据一致性有时候是个头疼但必须解决的问题，今天我从背景、问题、解决思路、技术方案等几个方面浅谈一下，抛砖引玉。
背景 异构系统近两年我司承接了某个持续性的会议项目，即每季度或月不定期举行会议。本项目目前有三个主要供应方（面向用户的A方，数据中间B方，会议数据同步C方）。 为了方便演示问题，以下流程和职责都做了裁剪。
简化流程如下： 简化职责如下：
A方职责： 用户通过官网/小程序进行报名，A方调用B方的标准接口，不存储数据 B方职责：作为ISP,提供标准查询、新增、修改等相关接口，几乎不提供定制。基于表单和表单数据，完成数据存储与流转。 C方职责：提供导入/更新/审核/注销等入口，新数据会通知到B方，B方数据新增/更新也会通知到C方。
从图例来看，B方/C方数据存储方面是冗余的。但B方只存储了核心数据，提供不了太多业务行为，C方具有业务需要的全套流程，但在此项目中作为后方支持及后续现场支持，三方形成了一种生态和谐。本篇博客主旨在讨论多方异构系统之间如何保证数据的一致性。
产品/项目从标准Sass系统来讲，这样的多方交互，不利于系统稳定性，有诸多不可控因素。但从项目角度，这是各方考虑/斗争/谈判/费用等综合因素下友好协商的结果。 当然这是一个私有部署项目，所以会有很多坚持和妥协。
大领导提到一个说法：项目是要交付的，功能完美是产品考虑的。在功能不完善的情况下，如何去交付？
最后的兜底哎，一言难尽。是通宵了几次核对/修复数据的，这是最后的办法了。为了苦逼不再重现，今年要对整个线动一动手术。（说好的.net 不996呢？）（拿着白菜价操着卖白粉的心）。
问题 请求无序C方 需要所有子会报名前，主会必须报名。 B方 各会之间的报名数据是无序到达的。
B方 任意报名数据更新或新增都会推送到C方，C方收到更新也会更新B方。这里有一些措施进行了拦截中断，但仍会频繁循环更新问题。这是目前现状（为什么会出现？太赶工？）
无开发环境，需盲写代码，发到测试环境进行联调测试。 调用链太长，日志过多，排错时需要根据调用各服务接口来判断走到了哪步，出现了一个问题。调用链能查到一些问题，但不容批量定位问题。单个查太难。
高并发下，redis组件出现各种问题（timeout等） token问题 数据丢失 更新失效 数据重复 队列积压 接口请求时间超长 其他问题...
大部分数据能对上，偶尔几十个或断断续续产生新问题的数据需要及时人工修复。功能有缺陷，人工也是一种交付办法，但不可持续，太他妈的累了。数据不一致，也是导致通宵核对/修复数据的一大原因。如果数据全一致，就不会那么辛苦了。
解决思路 管理层明确项目是要继续做的 目标产品化/更方便维护方向发展。一团队养一项目。 有改进想法提出来，拉会推进 缺人,招人（遥遥无期...）
针对请求无序问题，引入延时队列，先处理主会、子会延迟几秒钟在处理。 针对循环更新问题，记录B方数据来源，非必要情况下，不回更B方。必须终止掉。 针对排错困难问题，引入mysql记录新增报名的请求以及处理结果，可以更快查询处理结果。 针对bug,测试根据各测试场景进行复测，按10/100/1000/3000/万级规模压测。提前发下问题。 推进客户方一起做必要去重逻辑。
无论是标准产品还是交付项目，做任何改动都要评估。
多沟通，大家都是站在一条线的。有利于事情解决的方案认同度会更高。 预估花多少时间，有多少资源。 能挤出来的空窗期有多久，客户方/产品方对于需求的急迫性有多强。 基于场景测试，把缺陷优先级先列出来，根据空窗期先修复紧急缺陷。
把紧急且影响范围广的问题解决了，风险就小了很多了。80%的问题是由20%的因素造成的。 这也正符合程序优化中的时间/空间局部性。
“技术方案 mysql实现延迟队列进程运行时，在一段时间里，程序的执行往往呈现高度的局部性， 包括时间局部性和空间局部性。
”
时间局部性是一旦一个指令被执行了， 则在不久的将来，它可能再被执行。
空间局部性是一旦一个指令一个存储单元被访问，那么它附近的单元也将很快被访问.
优先处理主会，子会延时处理 由于隐私问题，这里只列部分字段
数据库轮询获取未处理数据 这里如何提高消费速度，可以参考《计算机系统结构》中标量处理机的流水线的一些知识。
首先要无相关，即按AccountId分组，分组内的数据是无冲突/相关的，可以分批进行。记录各任务状态，最后统一提交数据库状态，然后1s后继续轮询。这种类似静态流水线。动态流水线较为复杂，这里暂不做实现。
do
{
vargroupTemps=groupDatas.Skip((pageIndex-1)*pageSize).Take(pageSize).ToList();
varcurrentRecords=newList<QidianNotifydelayData>();
foreach(varitemingroupTemps)
{
currentRecords.AddRange(item.ToList());
}
vartemp=taskFunc(currentRecords);
taskList.Add(temp);
pageIndex++;
}
while((pageIndex-1)*pageSize<=groupCount);
//等待全部执行
awaitTask.WhenAll(taskList.ToArray());
await_dbContext.CommitAsync();
Thread.Sleep(1);
如果1s轮询觉得太浪费，后续可以根据请求发送标记位（下次轮询时间），有数据时，可以快速轮询，无数据时放宽时间。极端处理方式，当主会请求过来处理完成后，直接发起子会处理，但要考虑数据库是否能承受的住这种并发压力。
如果考虑请求会重复执行，可以在执行内加redis锁。慎用for update,并发一大就over.
///<summary>
///锁定执行。
///</summary>
///<paramname="key"></param>
///<paramname="func"></param>
///<paramname="timeSpan"></param>
///<returns></returns>
publicasyncTask<BizResult<T>>LockExcute<T>(stringkey,Func<Task<BizResult<T>>>func,inttimeSpan)
{
vardb=(this._cacheClientasRedisClient).Db;
varmutexKey=string.Format("mutex:",key);
if(awaitdb.StringSetAsync(mutexKey,"1",TimeSpan.FromSeconds(timeSpan),When.NotExists))
{
try
{
varitem=awaitfunc.Invoke();
returnitem;
}
catch(Exceptionex)
{
_logger.LogError("LockExcute:Exception:"+ex.Message);
returnBizResult.BusinessFailed<T>(-1,$"执行失败,Message:{ex.Message}");
}
finally
{
awaitdb.KeyDeleteAsync(mutexKey);
}
}
else
{
_logger.LogWarning($"LockExcute:Key:{key},正在处理中，请稍候");
returnBizResult.BusinessFailed<T>(-1,"正在处理中，请稍候");
}
}
由于业务中一个Account同时只能处理一个主会，如果在处理子会的时候，主会请求突然过来了，就会有问题，这里就需要加锁主会。引入了Redis延迟队列 基于Redis ZSet有序集合实现。 思路：当前时间戳和延时时间相加，也就是到期时间，存入Redis中，然后不断轮询，找到到期的，拿到再删除即可。 目前实现缺点：不利于监控，未发起www.cnblogs.com/fancunwei/p/16125202.html，转载请注明地址，谢谢
本文共计3823个文字，预计阅读时间需要16分钟。
之前忙碌于一个多方合作的项目，异构系统间维护数据一致性有时是个头疼的问题。今天，我从背景、问题、解决方案、技术方案等几个方面浅析一下。
之前忙活过一个多方合作的项目，异构系统间维护数据一致性有时候是个头疼但必须解决的问题，今天我从背景、问题、解决思路、技术方案等几个方面浅谈一下，抛砖引玉。
背景 异构系统近两年我司承接了某个持续性的会议项目，即每季度或月不定期举行会议。本项目目前有三个主要供应方（面向用户的A方，数据中间B方，会议数据同步C方）。 为了方便演示问题，以下流程和职责都做了裁剪。
简化流程如下： 简化职责如下：
A方职责： 用户通过官网/小程序进行报名，A方调用B方的标准接口，不存储数据 B方职责：作为ISP,提供标准查询、新增、修改等相关接口，几乎不提供定制。基于表单和表单数据，完成数据存储与流转。 C方职责：提供导入/更新/审核/注销等入口，新数据会通知到B方，B方数据新增/更新也会通知到C方。
从图例来看，B方/C方数据存储方面是冗余的。但B方只存储了核心数据，提供不了太多业务行为，C方具有业务需要的全套流程，但在此项目中作为后方支持及后续现场支持，三方形成了一种生态和谐。本篇博客主旨在讨论多方异构系统之间如何保证数据的一致性。
产品/项目从标准Sass系统来讲，这样的多方交互，不利于系统稳定性，有诸多不可控因素。但从项目角度，这是各方考虑/斗争/谈判/费用等综合因素下友好协商的结果。 当然这是一个私有部署项目，所以会有很多坚持和妥协。
大领导提到一个说法：项目是要交付的，功能完美是产品考虑的。在功能不完善的情况下，如何去交付？
最后的兜底哎，一言难尽。是通宵了几次核对/修复数据的，这是最后的办法了。为了苦逼不再重现，今年要对整个线动一动手术。（说好的.net 不996呢？）（拿着白菜价操着卖白粉的心）。
问题 请求无序C方 需要所有子会报名前，主会必须报名。 B方 各会之间的报名数据是无序到达的。
B方 任意报名数据更新或新增都会推送到C方，C方收到更新也会更新B方。这里有一些措施进行了拦截中断，但仍会频繁循环更新问题。这是目前现状（为什么会出现？太赶工？）
无开发环境，需盲写代码，发到测试环境进行联调测试。 调用链太长，日志过多，排错时需要根据调用各服务接口来判断走到了哪步，出现了一个问题。调用链能查到一些问题，但不容批量定位问题。单个查太难。
高并发下，redis组件出现各种问题（timeout等） token问题 数据丢失 更新失效 数据重复 队列积压 接口请求时间超长 其他问题...
大部分数据能对上，偶尔几十个或断断续续产生新问题的数据需要及时人工修复。功能有缺陷，人工也是一种交付办法，但不可持续，太他妈的累了。数据不一致，也是导致通宵核对/修复数据的一大原因。如果数据全一致，就不会那么辛苦了。
解决思路 管理层明确项目是要继续做的 目标产品化/更方便维护方向发展。一团队养一项目。 有改进想法提出来，拉会推进 缺人,招人（遥遥无期...）
针对请求无序问题，引入延时队列，先处理主会、子会延迟几秒钟在处理。 针对循环更新问题，记录B方数据来源，非必要情况下，不回更B方。必须终止掉。 针对排错困难问题，引入mysql记录新增报名的请求以及处理结果，可以更快查询处理结果。 针对bug,测试根据各测试场景进行复测，按10/100/1000/3000/万级规模压测。提前发下问题。 推进客户方一起做必要去重逻辑。
无论是标准产品还是交付项目，做任何改动都要评估。
多沟通，大家都是站在一条线的。有利于事情解决的方案认同度会更高。 预估花多少时间，有多少资源。 能挤出来的空窗期有多久，客户方/产品方对于需求的急迫性有多强。 基于场景测试，把缺陷优先级先列出来，根据空窗期先修复紧急缺陷。
把紧急且影响范围广的问题解决了，风险就小了很多了。80%的问题是由20%的因素造成的。 这也正符合程序优化中的时间/空间局部性。
“技术方案 mysql实现延迟队列进程运行时，在一段时间里，程序的执行往往呈现高度的局部性， 包括时间局部性和空间局部性。
”
时间局部性是一旦一个指令被执行了， 则在不久的将来，它可能再被执行。
空间局部性是一旦一个指令一个存储单元被访问，那么它附近的单元也将很快被访问.
优先处理主会，子会延时处理 由于隐私问题，这里只列部分字段
数据库轮询获取未处理数据 这里如何提高消费速度，可以参考《计算机系统结构》中标量处理机的流水线的一些知识。
首先要无相关，即按AccountId分组，分组内的数据是无冲突/相关的，可以分批进行。记录各任务状态，最后统一提交数据库状态，然后1s后继续轮询。这种类似静态流水线。动态流水线较为复杂，这里暂不做实现。
do
{
vargroupTemps=groupDatas.Skip((pageIndex-1)*pageSize).Take(pageSize).ToList();
varcurrentRecords=newList<QidianNotifydelayData>();
foreach(varitemingroupTemps)
{
currentRecords.AddRange(item.ToList());
}
vartemp=taskFunc(currentRecords);
taskList.Add(temp);
pageIndex++;
}
while((pageIndex-1)*pageSize<=groupCount);
//等待全部执行
awaitTask.WhenAll(taskList.ToArray());
await_dbContext.CommitAsync();
Thread.Sleep(1);
如果1s轮询觉得太浪费，后续可以根据请求发送标记位（下次轮询时间），有数据时，可以快速轮询，无数据时放宽时间。极端处理方式，当主会请求过来处理完成后，直接发起子会处理，但要考虑数据库是否能承受的住这种并发压力。
如果考虑请求会重复执行，可以在执行内加redis锁。慎用for update,并发一大就over.
///<summary>
///锁定执行。
///</summary>
///<paramname="key"></param>
///<paramname="func"></param>
///<paramname="timeSpan"></param>
///<returns></returns>
publicasyncTask<BizResult<T>>LockExcute<T>(stringkey,Func<Task<BizResult<T>>>func,inttimeSpan)
{
vardb=(this._cacheClientasRedisClient).Db;
varmutexKey=string.Format("mutex:",key);
if(awaitdb.StringSetAsync(mutexKey,"1",TimeSpan.FromSeconds(timeSpan),When.NotExists))
{
try
{
varitem=awaitfunc.Invoke();
returnitem;
}
catch(Exceptionex)
{
_logger.LogError("LockExcute:Exception:"+ex.Message);
returnBizResult.BusinessFailed<T>(-1,$"执行失败,Message:{ex.Message}");
}
finally
{
awaitdb.KeyDeleteAsync(mutexKey);
}
}
else
{
_logger.LogWarning($"LockExcute:Key:{key},正在处理中，请稍候");
returnBizResult.BusinessFailed<T>(-1,"正在处理中，请稍候");
}
}
由于业务中一个Account同时只能处理一个主会，如果在处理子会的时候，主会请求突然过来了，就会有问题，这里就需要加锁主会。引入了Redis延迟队列 基于Redis ZSet有序集合实现。 思路：当前时间戳和延时时间相加，也就是到期时间，存入Redis中，然后不断轮询，找到到期的，拿到再删除即可。 目前实现缺点：不利于监控，未发起www.cnblogs.com/fancunwei/p/16125202.html，转载请注明地址，谢谢