这个问题在这里已经有了答案：
在kafka streams作业中执行同步数据库查询或restful调用是一种好的做法吗(1个答案）
两年前关门了。
有一个kafka流组件可以从主题中获取json数据。现在我要做的是：
解析输入的json数据并获取某个id（identifier）属性的值
对oracle数据库中的特定表进行查找
使用查找表中的数据丰富输入json
将丰富的json数据发布到另一个主题
实现第二步的最佳设计方法是什么？我对如何做其他步骤有一个合理的想法。非常感谢您的帮助。
这个问题在这里已经有了答案：
在kafka streams作业中执行同步数据库查询或restful调用是一种好的做法吗(1个答案）
两年前关门了。
有一个kafka流组件可以从主题中获取json数据。现在我要做的是：
解析输入的json数据并获取某个id（identifier）属性的值
对oracle数据库中的特定表进行查找
使用查找表中的数据丰富输入json
将丰富的json数据发布到另一个主题
实现第二步的最佳设计方法是什么？我对如何做其他步骤有一个合理的想法。非常感谢您的帮助。
1条答案
按热度按时间yks3o0rb1#
根据您所讨论的数据集的大小和流的容量，我会尽可能多地缓存数据库（假设它不会经常更改）。在延迟和性能方面，通过查询每个记录上的数据库来扩充数据是非常昂贵的。
我以前做过的方法是示例化一个线程，该线程的唯一任务是维护一个新的本地缓存（通常是一个concurrenthashmap），并将其提供给需要它的进程。在这种情况下，您可能需要创建一个处理器，将对上述concurrenthashmap的引用提供给它，当kafka记录出现时，使用键查找数据，扩充记录，并将其发送到sink处理器或另一个streams处理器，这取决于您对它的处理方式。
在查找失败的情况下，您可以回退以按需对数据库进行实际查询，但您可能希望对此进行测试和分析，因为在100%缓存未命中的最坏情况下，您将大量查询数据库。