Kafka Streams 有一个接口,Processor
,其实现是有状态的。在开发者指南中给出了一个示例实现:
public class WordCountProcessor implements Processor<String, String> {
private ProcessorContext context;
private KeyValueStore<String, Long> kvStore;
@Override
@SuppressWarnings("unchecked")
public void init(ProcessorContext context) {
// keep the processor context locally because we need it in punctuate() and commit()
this.context = context;
// call this processor's punctuate() method every 1000 time units.
this.context.schedule(1000);
// retrieve the key-value store named "Counts"
kvStore = (KeyValueStore) context.getStateStore("Counts");
}
@Override
public void process(String dummy, String line) {
String[] words = line.toLowerCase().split(" ");
for (String word : words) {
Long oldValue = kvStore.get(word);
if (oldValue == null) {
kvStore.put(word, 1L);
} else {
kvStore.put(word, oldValue + 1L);
}
}
}
@Override
public void punctuate(long timestamp) {
KeyValueIterator<String, Long> iter = this.kvStore.all();
while (iter.hasNext()) {
KeyValue<String, Long> entry = iter.next();
context.forward(entry.key, entry.value.toString());
}
iter.close();
// commit the current processing progress
context.commit();
}
@Override
public void close() {
// close the key-value store
kvStore.close();
}
}
init
方法初始化了WordCountProcessor
的内部状态,例如检索键值存储。其他方法,如process
和close
,利用了这个状态。我不清楚如何在Clojure中“实现”这样的接口。我们如何将
init
检索到的状态传递给process
、close
等方法?
使用闭包?
我想到的一个想法是使用闭包:(let [ctx (atom nil)]
(reify Processor
(close [this]
;; Do something w/ ctx
)
(init [this context]
(reset! ctx context))
(process [this k v]
;; Do something w/ ctx
)
(punctuate [this timestamp]
;; Do something w/ ctx
)))
令人烦恼的是,每次都需要从
ProcessorContext
对象开始,因此键值存储代码将在所有需要键值存储的方法中重复出现。我看不到(一般)避免这种情况的方法,但我们可以根据具体情况用更具体的状态替换
ctx
原子。有更好的方法吗?