用于从strem创建GlobalKTable的Spring云函数

Question

是否有一个示例，该示例如何使用Spring Cloud流和使用Functional方法创建GlobalKTable来保持KStream的计数？

Answer 1

实现处理器接口是否正确？

    @Bean
    public Consumer<KStream<String, Long>> processorsample() {
        return input -> input.process(() -> new Processor<String, Long>() {

            @Override
            public void init(ProcessorContext context) {
                if (state == null) {
                    state = (KeyValueStore<String, Long>) context.getStateStore("mystate");
                } 
            }

            @Override
            public void process(String key, Long value) {
                if (state != null) {
                    if (key != null) {
                        Long currentCount = state.get(key);
                        if (currentCount == null) {
                            state.put(key, value);
                        } else {
                            state.put(key, currentCount + value);
                        }

                    }

                }
            }

            @Override
            public void close() {
                if (state != null) {
                    state.close();
                }
            }
        }, "mystate");

    }