假设我们有一个将其状态存储在数据库中的程序，我们希望其他程序在发生变化时做一些事情。例如，我们可能想在银行余额下降到某个阈值以下时发送电子邮件通知。这是应用程序使用Kafka等消息队列的一个非常常见的原因。
不幸的是，当组件发生故障时，这种实现并不工作。
我怀疑有很多真实的应用会出现这种错误。大多数时候，这些应用都能正常工作，而且变化是在多个系统中复制的。然而，当事情重新开始时，更新可能会丢失，或者出现额外的更新。在这篇文章中，我将尝试解释这种情况是如何出错的，以及一些修复的方法。
1、并行更新两者
应用程序执行以下操作。
- 将更改写入数据库。
- 同时，将消息发布到消息队列中。
这种情况下的问题是，数据库更新可能失败，但发布消息却成功了。这意味着消费流的应用程序会收到一个在数据库中不存在的 "额外 "更新。
2、更新数据库，然后发布消息
好的，让我们再试一下，确保更新数据库成功。
- 把变化写入数据库。
- 等待数据库确认写入发生。
- 将消息发布到消息队列中。
我们解决了 "额外 "的消息更新问题然而，我们仍然有一个问题。如果应用程序在写入数据库之后，但在发布消息之前崩溃，那么流就会缺少一个更新。如果消息队列不可用，这可能特别糟糕。应用程序可以重新尝试发布消息一段时间。然而，如果消息队列瘫痪的时间足够长，应用程序很可能会耗尽内存，或者被重新启动。在这种情况下，所有等待的更新都会丢失。
那么现在怎么办？我们不能按顺序进行操作，也不能并行操作。诀窍在于对工作进行排序，以便有一个单一的提交点：在这个点之前，操作没有发生，但在操作发生之后。我觉得这个概念很有帮助。如果你在一个应该提供某种形式的原子性的协议中找不到一个单一的提交点，要么你需要继续挖掘，要么协议是错误的。
在这种情况下，我们有两样东西可以存储 "正确 "的状态：消息队列和数据库。
我们需要改变它，以保证只有一个存储正确状态，保持一个真相来源。
3、数据库是真相的来源
最直接的选择可能是让数据库成为更新和要发布的消息的真实来源。
要做到这一点，应用程序需要在一个数据库事务中执行两个数据库操作：
- 更新应用程序的状态，并向 "待更新 "表追加一条新记录。
- 这个事务成为 "提交点"。
- 如果它提交了，说明对数据库的更新已经发生。
- 我们只需要保证消息被发布。
为了做到这一点，我们运行一个单独的进程来定期扫描待更新表。这个进程将消息发布到队列中，然后删除记录。这种消息发布是以 "至少一次 "的语义进行的。在失败的情况下，我们可以重试，直到它成功。我们最终可能会在队列中为一个单一的更新发布一个以上的消息，但我们永远不会错过任何一个消息。
一个更有效、有时更方便的实现方式是使用一个 "变化数据捕获CDC "系统，通过与现有的超前写入日志（如Debezium）整合，复制数据库的变化。这有效地做了同样的事情，但没有应用程序的参与。缺点是消息的内容仅限于数据库本身记录的内容，而不能是特定的应用。
在我看来，这两种解决方案中的任何一种都是最好的。为了完整起见，我们列出了其余两个解决方案，但对我来说，它们似乎没有那么好。
4、消息队列是真理的来源
另一个解决方案是依靠消息队列。
- 应用程序不直接写到数据库中，相反，它只写到消息队列中。
- 然后，消息队列被复制到数据库和其他应用程序。
- 这意味着 "提交点 "现在正在发布消息。
最大的缺点是，如果有需要执行的约束，数据库可能会拒绝一些更新，那么我们就会遇到同样的问题，即更新在消息队列中，但没有应用到数据库中。
例如，如果更新代表 "提取现金"，而银行应用强制要求账户不能透支，数据库可能需要拒绝更新。
理论上，一个应用程序可以解决这个问题。它需要实现一种两阶段的提交方式：
- 在第一阶段，应用程序直接访问数据库，检查更新是否可以应用，并在逻辑上 "锁定 "相关项目。
- 然后，更新被发送到消息队列中。
- 处理消息，然后应用更新并 "解锁 "项目。
这是很复杂的，所以你可能应该让数据库存储待定的更新。
另一个缺点是很难知道更新是什么时候应用的，因为.这可能导致应用程序保存了一个更新，但当用户重新加载状态时，他们直到一段时间后才看到更新的错误。这也可以通过让应用程序轮询数据库来解决，直到它看到更新被应用。
也许解决方案。两阶段提交？
另一个解决方案是让应用程序在数据库和队列之间使用两阶段提交。
这样做的好处是队列中不会有重复的数据。
缺点是这需要使用很少使用的业务功能，并且需要复杂的处理应用程序的重新启动。
我认为Kafka的事务支持可以用来做这个，尽管他们的文档说他们不支持。
这也有一个问题，就是当应用程序崩溃时，出现"卡住的事务"。这些 "卡住的事务 "可能可以通过让数据库成为两阶段提交的真相来源来解决，但这需要非常谨慎的操作排序。
两阶段提交的解决方案在理论上似乎很有趣，但并不实用。