要將Kafka數據寫入Redis,可以按照以下步驟進行:
以下是一個示例代碼,展示了如何將Kafka數據寫入Redis:
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import redis.clients.jedis.Jedis;
import java.util.Collections;
import java.util.Properties;
public class KafkaToRedisExample {
private static final String KAFKA_BOOTSTRAP_SERVERS = "localhost:9092";
private static final String KAFKA_TOPIC = "your-kafka-topic";
private static final String REDIS_HOST = "localhost";
private static final int REDIS_PORT = 6379;
public static void main(String[] args) {
// 創建Kafka消費者配置
Properties kafkaProps = new Properties();
kafkaProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BOOTSTRAP_SERVERS);
kafkaProps.put(ConsumerConfig.GROUP_ID_CONFIG, "kafka-to-redis-example-group");
kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 創建Redis客戶端
Jedis jedis = new Jedis(REDIS_HOST, REDIS_PORT);
// 創建Kafka消費者
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(kafkaProps);
consumer.subscribe(Collections.singletonList(KAFKA_TOPIC));
// 持續從Kafka消費消息并寫入Redis
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
// 解析Kafka消息
String key = record.key();
String value = record.value();
// 寫入Redis
jedis.set(key, value);
System.out.println("Wrote to Redis: " + key + " - " + value);
}
// 提交消費位移
consumer.commitAsync();
}
}
}
請根據你的實際情況修改KAFKA_BOOTSTRAP_SERVERS
、KAFKA_TOPIC
、REDIS_HOST
和REDIS_PORT
等配置。這個示例代碼使用了Kafka的Java客戶端和Jedis庫來連接Kafka和Redis。