我有下面的应用程序(我对这个框架很陌生),我希望看到缓存大小在从队列中读取消息时(增加),但它一直保持为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);
}
有人能说出缺了什么/错了什么吗?
谢谢!
2条答案
按热度按时间9gm1akwq1#
由于性能原因,默认情况下禁用度量。您可以使用
CacheConfiguration.setStatisticsEnabled(true)
或者statisticsEnabled
属性。huus2vyu2#
可能您没有消耗足够的条目来填充
IgniteDataStreamer
缓冲器。尝试设置刷新超时: