java – Kafka Streams本地国营商店

我有一个简单的流应用程序将一个主题作为输入流并将KeyValues转换为另一个,如：

StoreBuilder<KeyValueStore<Long, CategoryDto>> builder =
        Stores.keyValueStoreBuilder(Stores.inMemoryKeyValueStore(CategoryTransformer.STORE_NAME),
                                    Serdes.Long(), CATEGORY_JSON_SERDE);
    streamsBuilder.addStateStore(builder)
                         .stream(categoryTopic, Consumed.with(Serdes.Long(), CATEGORY_JSON_SERDE))
                         .transform(CategoryTransformer::new, CategoryTransformer.STORE_NAME);

static class CategoryTransformer implements Transformer<Long, CategoryDto, KeyValue<Long, CategoryDto>> {

    static final String STORE_NAME = "test-store";

    private KeyValueStore<Long, CategoryDto> store;

    @Override
    public void init(ProcessorContext context) {
      store = (KeyValueStore<Long, CategoryDto>) context.getStateStore(STORE_NAME);
    }

    @Override
    public KeyValue<Long, CategoryDto> transform(Long key, CategoryDto value) {
      store.put(key, value);
      return KeyValue.pair(key, value);
    }

    @Override
    public KeyValue<Long, CategoryDto> punctuate(long timestamp) {
      return null;
    }

    @Override
    public void close() {

    }
  }

在这里我不得不使用变压器,因为我需要获取存储并更新相关值.

问题是使用本地状态存储和将值放在ForeachAction内的简单HashMap之间有什么区别？

在这种情况下使用本地国营商店有什么好处？

解决方法:

虽然它没有显示在您的代码中,但我假设您以某种方式读取并使用存储状态.

使用简单(在内存中)存储状态HashMap会使您的状态完全不持久,这意味着当下列任何一种情况发生时,您的状态将会丢失(这些都不是特别的,假设它会经常发生)：

>您的流处理器/应用程序停止,
>崩溃,或
>由于重新平衡,部分迁移到其他地方(其他JVM).

非持久状态的问题是,当发生上述任何一种情况时,kafka-streams将在最后提交的偏移处重新启动处理.因此,在崩溃/停止/重新平衡之前处理的所有记录都不会被重新处理,这意味着当处理重新开始时,HashMap的内容将为空.这当然不是你想要的.

另一方面,如果您使用其中一个提供的状态存储,kafka-streams将确保一旦处理在上面列出的任何中断之后重新启动,状态将可用,就好像处理从未停止,而不重新处理任何以前处理过的记录.

标签：apache-kafka-streams,java,apache-kafka
来源： https://codeday.me/bug/20190823/1697590.html