Kafka 实现实时读

[baisui1981]

Kafka 实现实时读，需要支持以下两种执行方式：

1. Flink SQL
2. Stream API

已经实现了一个Kafka Sink ：#179

[baisui1981]

kafka 中 一个topic 中可以 存多个表的同步信息吗？

在Apache Kafka中，一个topic主要用于存储一类特定的消息。对于数据库表的同步信息来说，理论上你可以在一个topic中存储多个表的变更数据，但这不是最佳实践。

通常情况下，为了保持消息的清晰性和系统的可维护性，每个数据库表的变更数据应该使用独立的Kafka topic。这样做有几个好处：

1. 分区和并行处理：不同的表可能有不同的吞吐量需求，独立的topic可以更容易地调整分区数量以优化并行处理。
2. 消费逻辑分离：不同的表可能由不同的消费者组或者应用来消费，独立的topic使得这些消费者可以独立工作而不互相干扰。
3. 简化错误处理和恢复：如果某个表的数据出现问题，只需要重置或重新处理该表对应的topic，而不会影响其他表的数据流。
4. 资源管理：可以更精细地控制每个topic的资源分配，例如保留策略、压缩设置等。
5. 语义清晰：每个topic只包含一种类型的事件，这使得系统更易于理解和维护。

然而，如果你确实需要将多个表的变更合并到一个topic中，你可以通过消息格式设计来实现这一点，比如在每条消息中包含一个字段来标识是哪个表的变更。但是，这种做法可能会使你的系统更加复杂，并且难以管理和扩展。

总之，虽然技术上可行，但最好为每个表创建单独的Kafka topic用于数据同步。

一个topic 只存放一个表的增量同步内容，每个topic 的消费速度不同，数据库中一次事务执行会涉及到多个表的更新，下游监听kafka增量消息会，收到增量消息可能与实务执行的前后顺序错乱

您提到的问题是分布式系统中常见的挑战之一，尤其是在处理跨多个Kafka topic的事务性保证时。当一个事务涉及到多个表（因此是多个Kafka topic）的更新时，下游消费者可能会接收到顺序错乱的消息，这是因为每个topic的消费速度不同，以及消息在网络中的传输延迟可能不同。

为了解决这个问题，可以考虑以下几种策略：

1. 使用Kafka事务API：
   Kafka从0.11版本开始支持生产者端的事务API，这使得可以在多个topic和partition之间提供恰好一次的语义保证。通过使用事务API，你可以确保一组消息要么全部提交，要么全部不提交，从而保持了事务的一致性。

2. 引入全局事务ID：
   在每条消息中加入一个全局唯一的事务ID，并且包含事务的时间戳或序列号。这样，即使消息到达的顺序与事务执行的顺序不同，消费者也可以根据事务ID对消息进行排序，以重建原始的事务顺序。

3. 使用单个复合topic：
   如果可以接受的话，可以将所有涉及的表的增量同步信息合并到一个topic中，并在消息格式中包含足够的元数据来标识是哪个表的变更。这种方式简化了消费者逻辑，因为只需要监听一个topic，但会失去一些之前提到的好处，如并行处理等。

4. 两阶段提交协议：
   为了确保跨多个topic的消息一致性，可以在应用层实现类似于两阶段提交的协议。首先，准备阶段所有的参与者同意提交；其次，提交阶段正式提交更改。不过，在分布式环境中实现这一点可能是复杂且有挑战性的。

5. 事件溯源模式：
   使用事件溯源模式，其中每个业务操作都被记录为一系列不可变的事件。这些事件可以按照发生的顺序被持久化，并且可以通过快照机制来加速读取。虽然这不是直接解决Kafka问题的方法，但它可以帮助确保数据的一致性和可追溯性。

6. CDC (Change Data Capture) 工具的支持：
   某些CDC工具，如Debezium，提供了内置的功能来帮助维持跨多个表的操作顺序。它们通常会在消息中包含必要的元数据，以便消费者能够正确地重新构建事务顺序。

选择哪种方法取决于您的具体需求、现有架构以及愿意接受的复杂度。如果可能的话，利用Kafka自带的事务特性通常是较为推荐的做法，因为它能减少额外开发工作的需要，并且更加可靠。同时，确保您的CDC工具配置正确，以尽量减少顺序混乱的风险。

[baisui1981]

kafka 文本传输format支持种类：

1. 纯json
2. canal json
3. debezium json

以上三种协议都是基于json文本传输的，在配置时有几个共通配置项：

1. ignoreParseErrors
2. timestampFormat：以SQL(yyyy-MM-dd HH:mm:ss.s{precision})或者 ISO-8601( yyyy-MM-ddTHH:mm:ss.s{precision}) 格式解析日期
3. encodeDecimal：是否将decaimal数值使用科学计数法来序列化，JsonFormatOptions.ENCODE_DECIMAL_AS_PLAIN_NUMBER
4. nullKeyMode：字段内容为map类型，当key为null时，的处理方式，literal：用一个常量字符替换、drop：直接丢弃、fail：抛异常
5. nullKeyLiteral：当nullKeyMode为literal时，使用的常量替换内容，对应的代码在 org.apache.flink.formats.json.RowDataToJsonConverters.#createMapConverter() 方法中