Если вы не хотите обрабатывать свои данные в режиме реального времени, вы можете пересмотреть вопрос о том, является ли Kafka правильным решением для вас.Тем не менее, вы можете попробовать это:
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "your_consumer_group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("foo", "bar"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
}
// After data is consumed, you make your thread sleep until next 30 min:
Thread.sleep(30 * 60 * 1000);
}
Если вы хотите, чтобы ваша пакетная обработка в режиме реального времени проводилась на каждой 30-й или 0-й минуте каждого часа, вы можете использовать этот спящий режим вместо этого:
Thread.sleep(System.currentTimeMillis() % (30*60*1000));
Это заставит вашего потребителя проснуться в 00:00
, 00:30
, 01:00
, 01:30
и т. Д. Перейдите по этой ссылке для получения подробной информации: https://kafka.apache.org/0100/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html
СноваВы, вероятно, не хотите использовать Кафку таким образом.Возможно, лучше сделать дамп данных в какое-либо хранилище (например, файлы паркетных файлов, разбитые по дате и времени) и выполнять некоторую пакетную обработку каждые 30 минут.