NiFi -更新处理器中的Luwak (Lucene)索引

Question

我正在尝试使用卢瓦克 Lucene创建一个自定义处理器，这样我就可以对传入的流文件运行查询。我想找出的是更新Luwak监视器中存在的查询索引的最佳方法(下面的示例代码)。

编辑-更多使用上下文

所谓update，我的意思是允许外部用户添加/更新/删除针对传入流文件运行的查询。我们将从一组固定的查询开始，但随后希望允许一个或多个用户更改针对传入消息执行的查询。挑战就在这里，改变正在执行的查询。

还有其他选择我应该考虑吗？如果有10k查询，更新这些查询似乎需要20多个时间。这很可能是罕见的，但重新加载/启动时间是我正在考虑的事情。

我考虑过的选项：

1. 使用UpdateAttribute并对每个流文件进行更新。不理想，特别是如果有一大堆查询要索引的话。
2. 使用http、AWS、SQS等来发送一个高优先级的流文件进行更新(比任何其他源都高)。不可怕，但还是不对。
3. 使用NiFi API在更新时启动/停止处理器。似乎不是一种非常有效的方法来执行更新，特别是如果它们经常发生。

实例化监视器：

Monitor monitor = new Monitor(new LuceneQueryParser("field"), new TermFilteredPresearcher());

添加查询--我试图优化的内容：

        //Add queries to the monitor
        for (Map.Entry<String, String> entry : bucketList.entrySet()) {
            MonitorQuery q = new MonitorQuery(entry.getKey(),entry.getValue());
            monitor.update(q);
        }

Answer 1

当处理器启动时，您可以启动一个后台计时器线程，该线程定期构建一个新监视器，然后替换处理器所使用的监视器。

您可能希望在处理器中设置一个成员变量，如下所示：

AtomicReference<Monitor> monitorHolder = new AtomicReference<Monitor>();

然后，在@OnScheduled中，您可以构建初始监视器并将其设置在持卡器中。

然后，在onTrigger中，您总是首先获得监视器：

Monitor localMonitor = monitorHolder.get();

然后，在后台线程中，您可以调用monitorHolder.set(newMonitor)，它不会影响当前处理器的执行，但将在下次调用onTrigger时生效。