flink不再将偏移量保存到主题 __consumer_offsets 中，而是存储到redis，如何设置

要将 Flink 的偏移量保存到 Redis 中，您需要使用 Flink 的 RedisConnector 库。以下是一些基本步骤：

1. 在您的 maven 项目中添加 Flink 的 RedisConnector 依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-redis_2.11</artifactId>
    <version>1.10.0</version>
</dependency>

2. 在 Flink 程序中，使用 Redis 的 Jedis 连接池创建一个 RedisSink。例如：

RedisSink<Tuple2<String, Long>> redisSink = new RedisSink<>(jedisPool, new MyRedisMapper());

3. 创建一个实现 RedisMapper 接口的类，以便将消息中的键值对添加到 Redis 中。例如：

public class MyRedisMapper implements RedisMapper<Tuple2<String, Long>> {

    @Override
    public RedisCommandDescription getCommandDescription() {
        return new RedisCommandDescription(RedisCommand.SET);
    }

    @Override
    public String getKeyFromData(Tuple2<String, Long> data) {
        return data.f0;
    }

    @Override
    public String getValueFromData(Tuple2<String, Long> data) {
        return data.f1.toString();
    }
}

4. 将 RedisSink 添加到您的 Flink 程序中。例如：

DataStream<Tuple2<String, Long>> dataStream = ...;
dataStream.addSink(redisSink);

5. 在您的 Flink 程序中使用 Flink 的 Checkpointing 机制，以便在发生故障时能够恢复偏移量。例如：

env.enableCheckpointing(5000);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);
env.getCheckpointConfig().setCheckpointTimeout(60000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

6. 在您的 Flink 程序中，使用 Flink 的 Kafka Consumer，将偏移量存储到 Redis 中。例如：

FlinkKafkaConsumer010<String> kafkaConsumer = new FlinkKafkaConsumer010<>(
        "my-topic",
        new SimpleStringSchema(),
        properties);

kafkaConsumer.setStartFromLatest();

DataStream<String> dataStream = env.addSource(kafkaConsumer);

dataStream.map(new MyMapFunction())
        .addSink(redisSink);

以上就是将 Flink 的偏移量保存到 Redis 中的基本步骤。请根据您的具体需求进行修改和调整。