之前忙活过一个多方合作的项目，异构系统间维护数据一致性有时候是个头疼但必须解决的问题，今天我从背景、问题、解决思路、技术方案等几个方面浅谈一下，抛砖引玉。
近两年我司承接了某个持续性的会议项目，即每季度或月不定期举行会议。本项目目前有三个主要供应方（面向用户的A方，数据中间B方，会议数据同步C方【我司】）。为了方便演示问题，以下流程和职责都做了裁剪。
简化流程如下：
简化职责如下：
从图例来看，B方/C方数据存储方面是冗余的。但B方只存储了核心数据，提供不了太多业务行为，C方具有业务需要的全套流程，但在此项目中作为后方支持及后续现场支持，三方形成了一种生态和谐。本篇博客主旨在讨论多方异构系统之间如何保证数据的一致性。
从标准Sass系统来讲，这样的多方交互，不利于系统稳定性，有诸多不可控因素。但从项目角度，这是各方考虑/斗争/谈判/费用等综合因素下友好协商的结果。当然这是一个私有部署项目，所以会有很多坚持和妥协。
大领导提到一个说法：项目是要交付的，功能完美是产品考虑的。在功能不完善的情况下，如何去交付？
哎，一言难尽。是通宵了几次核对/修复数据的，这是最后的办法了。为了苦逼不再重现，今年要对整个线动一动手术。（说好的.net 不996呢？）（拿着白菜价操着卖白粉的心）。
大部分数据能对上，偶尔几十个或断断续续产生新问题的数据需要及时人工修复。功能有缺陷，人工也是一种交付办法，但不可持续，太他妈的累了。数据不一致，也是导致通宵核对/修复数据的一大原因。如果数据全一致，就不会那么辛苦了。
无论是标准产品还是交付项目，做任何改动都要评估。
把紧急且影响范围广的问题解决了，风险就小了很多了。80%的问题是由20%的因素造成的。 这也正符合程序优化中的时间/空间局部性。
“进程运行时，在一段时间里，程序的执行往往呈现高度的局部性，包括时间局部性和空间局部性。
”
时间局部性是一旦一个指令被执行了， 则在不久的将来，它可能再被执行。
空间局部性是一旦一个指令一个存储单元被访问，那么它附近的单元也将很快被访问.
优先处理主会，子会延时处理
由于隐私问题，这里只列部分字段
数据库轮询获取未处理数据
这里如何提高消费速度，可以参考《计算机系统结构》中标量处理机的流水线的一些知识。
首先要无相关，即按AccountId分组，分组内的数据是无冲突/相关的，可以分批进行。记录各任务状态，最后统一提交数据库状态，然后1s后继续轮询。这种类似静态流水线。动态流水线较为复杂，这里暂不做实现。
do
{
var groupTemps = groupDatas.Skip((pageIndex - 1) * pageSize).Take(pageSize).ToList();
var currentRecords = new List<QidianNotifydelayData>();
foreach (var item in groupTemps)
{
currentRecords.AddRange(item.ToList());
}
var temp = taskFunc(currentRecords);
taskList.Add(temp);
pageIndex++;
}
while ((pageIndex - 1) * pageSize <= groupCount);
//等待全部执行
await Task.WhenAll(taskList.ToArray());
await _dbContext.CommitAsync();
Thread.Sleep(1);
如果1s轮询觉得太浪费，后续可以根据请求发送标记位（下次轮询时间），有数据时，可以快速轮询，无数据时放宽时间。极端处理方式，当主会请求过来处理完成后，直接发起子会处理，但要考虑数据库是否能承受的住这种并发压力。
如果考虑请求会重复执行，可以在执行内加redis锁。慎用for update,并发一大就over.
/// <summary>
/// 锁定执行。
/// </summary>
/// <param name="key"></param>
/// <param name="func"></param>
/// <param name="timeSpan"></param>
/// <returns></returns>
public async Task<BizResult<T>> LockExcute<T>(string key, Func<Task<BizResult<T>>> func, int timeSpan)
{
var db = (this._cacheClient as RedisClient).Db;
var mutexKey = string.Format("mutex:", key);
if (await db.StringSetAsync(mutexKey, "1", TimeSpan.FromSeconds(timeSpan), When.NotExists))
{
try
{
var item = await func.Invoke();
return item;
}
catch (Exception ex)
{
_logger.LogError("LockExcute:Exception:" + ex.Message);
return BizResult.BusinessFailed<T>(-1, $"执行失败,Message:{ex.Message}");
}
finally
{
await db.KeyDeleteAsync(mutexKey);
}
}
else
{
_logger.LogWarning($"LockExcute:Key:{key},正在处理中，请稍候");
return BizResult.BusinessFailed<T>(-1, "正在处理中，请稍候");
}
}
/// <summary>
/// 3.入队列
/// </summary>
/// <param name="redisModel"></param>
/// <returns></returns>
public async Task EnqueueZset(DataToModel redisModel)
{
redisModel.UpdateTime = redisModel.UpdateTime.AddSeconds(5);// 最后更新时间 + 5秒
var redisDb = _redisConnectionService.GetRedisConnectionMultiplexer().GetDatabase(0);//默认DB0
if (redisDb != null)
{
IsoDateTimeConverter timeFormat = new IsoDateTimeConverter();
timeFormat.DateTimeFormat = "yyyy-MM-dd HH:mm:ss.fff";
await redisDb.SortedSetAddAsync(ZSet_Queue, JsonConvert.SerializeObject(redisModel, Formatting.Indented, timeFormat), redisModel.UpdateTime.ToTimeStamp());//得分 --放入redis
_logger.LogInformation($"数据排队--入队列！redisModel：{JsonConvert.SerializeObject(redisModel)}");
}
}
这里只做简单截图
“-c, --container="": 容器名
”
-f, --follow[=false]: 指定是否持续输出日志
--interactive[=true]: 如果为true，当需要时提示用户进行输入。默认为true
--limit-bytes=0: 输出日志的最大字节数。默认无限制
-p, --previous[=false]: 如果为true，输出pod中曾经运行过，但目前已终止的容器的日志
--since=0: 仅返回相对时间范围，如5s、2m或3h，之内的日志。默认返回所有日志。只能同时使用since和since-time中的一种
--since-time="": 仅返回指定时间（RFC3339格式）之后的日志。默认返回所有日志。只能同时使用since和since-time中的一种
--tail=-1: 要显示的最新的日志条数。默认为-1，显示所有的日志
--timestamps[=false]: 在日志中包含时间戳
这里只截图简单信息
此处暂不截图。
当然那是进展顺利的情况下，不顺利的情况下就变成了这样
某些时候也会听到如下言论：
就如现在的疫情封控一样，做好了精准防控一片赞歌，失控了就好好居家、共渡难关。网络和现实都会告诉你什么就是人间。
以上是关于定制化需求的一些解决方案，希望对未来类似产品或项目做个参考。本篇从问题着手，分析有利于解决/消除异构系统数据一致性办法。当然数据一致性也依赖于自身系统的高可用，这里未做过多描述，以后再说。
到此结束，谢谢观看！