flink中自定义CheckpointedFunction，以实现将读取kafka时的偏移量存储到redis

好的，下面是一个简单的示例，展示了如何在 Flink 中实现一个支持状态检查点的 Kafka 消费者，并将消费到的偏移量保存到 Redis 中：

public class KafkaConsumerFunction implements SourceFunction<String>, CheckpointedFunction {

    private transient KafkaConsumer<String, String> kafkaConsumer;

    private transient Jedis jedis;

    private final String kafkaTopic;

    private final String kafkaBootstrapServers;

    private final String redisHost;

    private final int redisPort;

    private long lastCommittedOffset = 0L;

    private transient ListState<Long> checkpointedOffset;

    public KafkaConsumerFunction(String kafkaTopic, String kafkaBootstrapServers, String redisHost, int redisPort) {
        this.kafkaTopic = kafkaTopic;
        this.kafkaBootstrapServers = kafkaBootstrapServers;
        this.redisHost = redisHost;
        this.redisPort = redisPort;
    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        kafkaConsumer = new KafkaConsumer<>(createConsumerProperties());
        kafkaConsumer.subscribe(Collections.singletonList(kafkaTopic));
        jedis = new Jedis(redisHost, redisPort);

        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                ctx.collect(record.value());
                lastCommittedOffset = record.offset();
            }
            commitOffsetToRedis();
        }
    }

    private Properties createConsumerProperties() {
        Properties props = new Properties();
        props.put("bootstrap.servers", kafkaBootstrapServers);
        props.put("group.id", "flink-kafka-consumer");
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        return props;
    }

    private void commitOffsetToRedis() {
        jedis.set("kafka-offset", String.valueOf(lastCommittedOffset));
    }

    @Override
    public void cancel() {
        kafkaConsumer.close();
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        checkpointedOffset.clear();
        checkpointedOffset.add(lastCommittedOffset);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        ListStateDescriptor<Long> descriptor = new ListStateDescriptor<>("kafka-offset", Long.class);
        checkpointedOffset = context.getOperatorStateStore().getListState(descriptor);

        if (context.isRestored()) {
            for (Long offset : checkpointedOffset.get()) {
                lastCommittedOffset = offset;
            }
        }
    }

}

在这个示例中，我们实现了一个 Kafka 消费者，每次从 Kafka 中消费一条消息时，将消息的偏移量保存到 `lastCommittedOffset` 变量中，并将 `lastCommittedOffset` 变量的值保存到 Redis 中。在 `snapshotState()` 方法中，我们将 `lastCommittedOffset` 的值保存到 `ListState` 中，这个 `ListState` 会被 Flink 自动保存到检查点中。在 `initializeState()` 方法中，我们从 `ListState` 中恢复之前保存的 `lastCommittedOffset` 的值，如果当前任务是从检查点中恢复的，则将 `lastCommittedOffset` 的值设置为之前保存的值。这样，在任务重新启动时，我们就可以从上一次保存的偏移量的位置开始消费 Kafka 中的消息了。需要注意的是，这个示例中的 Redis 客户端使用的是 Jedis，如果你使用的是 Redisson 或者其他的 Redis 客户端，需要根据具体的客户端文档来实现 Redis 的操作。