如何正确读取ignite缓存

cpjpxq1n  于 2021-06-07  发布在  Kafka
关注(0)|答案(2)|浏览(404)

我有下面的应用程序(我对这个框架很陌生),我希望看到缓存大小在从队列中读取消息时(增加),但它一直保持为0。

KafkaStreamer<String, String, String> kafkaStreamer = new KafkaStreamer<>();

    Ignition.setClientMode(true);

    Ignite ignite = Ignition.start();

    Properties settings = new Properties();
    // Set a few key parameters
    settings.put("bootstrap.servers", "localhost:9092");
    settings.put("group.id", "test");
    settings.put("zookeeper.connect", "localhost:2181");
    settings.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    settings.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    settings.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    settings.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

    // Create an instance of StreamsConfig from the Properties instance
    kafka.consumer.ConsumerConfig config = new ConsumerConfig(settings);

    IgniteCache<String, String> cache = ignite.getOrCreateCache("myCache");

    IgniteDataStreamer<String, String> stmr = ignite.dataStreamer("myCache");
    // allow overwriting cache data
    stmr.allowOverwrite(true);

    kafkaStreamer.setIgnite(ignite);
    kafkaStreamer.setStreamer(stmr);

    // set the topic
    kafkaStreamer.setTopic("test");

    // set the number of threads to process Kafka streams
    kafkaStreamer.setThreads(1);

    // set Kafka consumer configurations
    kafkaStreamer.setConsumerConfig(config);

    // set decoders
    StringDecoder keyDecoder = new StringDecoder(null);
    StringDecoder valueDecoder = new StringDecoder(null);

    kafkaStreamer.setKeyDecoder(keyDecoder);
    kafkaStreamer.setValueDecoder(valueDecoder);

    kafkaStreamer.start();

    while (true) {

        System.out.println(cache.metrics().getSize());
        Thread.sleep(200);
    }

有人能说出缺了什么/错了什么吗?
谢谢!

9gm1akwq

9gm1akwq1#

由于性能原因,默认情况下禁用度量。您可以使用 CacheConfiguration.setStatisticsEnabled(true) 或者 statisticsEnabled 属性。

huus2vyu

huus2vyu2#

可能您没有消耗足够的条目来填充 IgniteDataStreamer 缓冲器。尝试设置刷新超时:

stmr.autoFlushFrequency(1000);

相关问题