寫個(gè)demo 練練手 椿争， 結(jié)論: 思維必須成流式 柳击，不要以數(shù)據(jù)庫(kù)的方式去看待流式聚合 橄杨， 流式的聚合七芭，在time window 中 也會(huì)產(chǎn)生很多事件 . 最后一點(diǎn) 素挽，kafka 提供著數(shù)據(jù)庫(kù)存儲(chǔ)能力的ktable。
也就是說(shuō) 狸驳，你可以 發(fā)請(qǐng)求給instance 预明，獲取ktable 的聚合數(shù)據(jù) 。
而不是 自己寫個(gè)服務(wù)作為消費(fèi)者然后去實(shí)現(xiàn)ktable的聚合 耙箍，這點(diǎn)有點(diǎn)別扭
建議大家看看 https://github.com/confluentinc/kafka-streams-examples.git

網(wǎng)狀網(wǎng)絡(luò)拓?fù)浣Y(jié)構(gòu) (2).png

10s 超過(guò)5次 評(píng)論 代碼

        SpecificAvroSerde<Content> contentSpecificAvroSerde = new SpecificAvroSerde<>();
        SpecificAvroSerde<ContentbyUserId> userContentSpecificAvroSerde = new SpecificAvroSerde<>();
        contentSpecificAvroSerde.configure(Collections.singletonMap(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081"), false);
        userContentSpecificAvroSerde.configure(Collections.singletonMap(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081"), false);
        Properties properties = config("contentId1", "localhost:9092", "/tmp/filter1");
        // key 為 contentId  value content
        final StreamsBuilder builder = new StreamsBuilder();
        // key userId , value content
        KStream<String, Content> kStream = builder.stream("content2", Consumed.with(Serdes.String(), contentSpecificAvroSerde))
                .selectKey((k, v) -> v.getUserId());
        KGroupedStream<String, Content> stringContentKGroupedStream = kStream.groupByKey();
        KStream<Windowed<String>, ContentbyUserId> k1ResultStrem = k1Result
                .toStream()
                .filter((k, v) -> {
            return null != v && v.getCount() > 5;
        });
        k1ResultStrem.print(Printed.toSysOut());

10s 內(nèi) 輸入事件間隔小于1s的事件數(shù) >5

  // key 為 contentId  value content
        final StreamsBuilder builder = new StreamsBuilder();
        // key userId , value content
        KStream<String, Content> kStream = builder.stream("content2", Consumed.with(Serdes.String(), contentSpecificAvroSerde))
                .selectKey((k, v) -> v.getUserId());
        KStream<Windowed<String>,Long> kStreamResult2 = stringContentKGroupedStream.windowedBy(SessionWindows.with(1 * 1000))
                .count(Materialized.<String, Long, SessionStore<Bytes, byte[]>>as("result2")
                        .withKeySerde(Serdes.String())
                        .withValueSerde(Serdes.Long()))
                .toStream()
                .filter((e, v) -> {
                            // session 窗口時(shí)間 大于 10 s 且 數(shù)量大于5
                            if (v != null && e.window().end() - e.window().start() > 10 * 1000 && v.longValue() > 5) {
                                return true;
                            }
                            return false;

                        }
                );

        kStreamResult2.print(Printed.toSysOut());

基礎(chǔ)配置

    public static Properties config(String appliactionId, String bootstrapServers, String stateDir) {
        final Properties streamsConfiguration = new Properties();
        // Give the Streams application a unique name.  The name must be unique in the Kafka cluster
        // against which the application is run.
        streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, appliactionId);
        // Where to find Kafka broker(s).
        streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
        streamsConfiguration.put("schema.registry.url", "http://localhost:8081");


        // Provide the details of our embedded http service that we'll use to connect to this streams
        // instance and discover locations of stores.
//        streamsConfiguration.put(StreamsConfig.APPLICATION_SERVER_CONFIG, host + ":" + applicationServerPort);
        streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, stateDir);
        // Set to earliest so we don't miss any data that arrived in the topics before the process
        // started
        streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // Set the commit interval to 500ms so that any changes are flushed frequently and the top five
        // charts are updated with low latency.
        streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 500);
        // Allow the user to fine-tune the `metadata.max.age.ms` via Java system properties from the CLI.
        // Lowering this parameter from its default of 5 minutes to a few seconds is helpful in
        // situations where the input topic was not pre-created before running the application because
        // the application will discover a newly created topic faster.  In production, you would
        // typically not change this parameter from its default.
        String metadataMaxAgeMs = System.getProperty(ConsumerConfig.METADATA_MAX_AGE_CONFIG);
        if (metadataMaxAgeMs != null) {
            try {
                int value = Integer.parseInt(metadataMaxAgeMs);
                streamsConfiguration.put(ConsumerConfig.METADATA_MAX_AGE_CONFIG, value);
                System.out.println("Set consumer configuration " + ConsumerConfig.METADATA_MAX_AGE_CONFIG +
                        " to " + value);
            } catch (NumberFormatException ignored) {
            }
        }
        return streamsConfiguration;
    }