在Apache Flink中使用Redis：Bahir的redis-connector与Jedis的区别

问题背景 #

我在使用Apache Flink时需要将处理结果存入Redis，了解到Apache Bahir包含redis-connector可以实现写入。但我也验证了直接用Jedis连接Redis的可行性，无需依赖Bahir的连接器就能成功写入处理后的消息，相关代码示例如下：

数据流处理代码 #

DataStream<String> messageStream = env.addSource(new FlinkKafkaConsumer<>(flinkParams.getRequired("topic"), new SimpleStringSchema(), flinkParams.getProperties())).setParallelism(Math.min(hosts * cores, kafkaPartitions));

messageStream.keyBy(new KeySelector<String, String>() {
    @Override
    public String getKey(String s) throws Exception {
        return s;
    }
}).flatMap(new RedisConnector());

Jedis操作封装类 #

public class ProcessorCommon {

    private static final Logger logger = LoggerFactory.getLogger(ProcessorCommon.class);

    private Jedis jedis;

    private Set<DummyPair> dummy;

    public ProcessorCommon(String redisServerHostName) {
        this.jedis = new Jedis(redisServerHostName);
    }

    public void writeToRedis(String key, String value) {
        this.jedis.set(key, value);
    }

    public String getFromRedis(String key) {
        return this.jedis.get(key);
    }

    public void close() {
        this.jedis.close();
    }
}

想了解这两种方式在Flink中操作Redis有何区别？

核心区别对比 #

• Flink原生集成度差异
  Bahir的Redis Connector是为Flink生态量身打造的，完全适配Flink Runtime的生命周期：

  • 自动完成并行任务的连接生命周期管理，在算子open()时初始化连接、close()时释放，避免手动管理导致的连接泄漏；
  • 原生支持Flink的检查点（Checkpoint）机制，能保证Exactly-Once语义——写入Redis的操作会和Checkpoint绑定，任务故障恢复时不会出现数据重复或丢失；
  • 无需手动处理状态持久化，直接适配Flink的状态后端。
    而直接用Jedis的话，以上所有逻辑都要自己编码实现，很容易出现疏漏，比如忘记在close()中释放连接，或者没处理Checkpoint一致性导致数据不一致。

• Redis操作的封装与扩展性
  Bahir连接器已经封装了Redis多种数据结构（String、Hash、List、Set、Sorted Set等）的操作，通过RedisMapper接口就能定义数据到Redis命令的映射，不用自己拼接底层Redis命令；
  同时原生支持Redis集群、哨兵模式，只需要配置对应的参数即可，无需手动编写集群分片、故障转移的逻辑。
  直接用Jedis的话，所有数据结构的操作都要自己实现，集群模式下还要手动处理节点路由、故障重连，开发成本高且容易出错。

• 容错与可靠性
  Bahir连接器内置了连接重试、超时处理、故障节点自动切换等容错机制，当Redis节点出现故障时会自动尝试恢复连接；
  直接用Jedis的话，这些容错逻辑需要自己编写，比如捕获JedisConnectionException后实现重试逻辑，否则任务可能直接因连接失败而终止。

• 性能优化
  Bahir连接器支持批量写入，会缓存一定量的数据后批量发送到Redis，大幅减少网络IO开销；
  如果直接用Jedis单条写入，性能会差很多，要自己实现批量缓存、批量提交的逻辑。

• 代码复杂度
  使用Bahir连接器只需要实现RedisMapper接口，配置好连接参数就能快速集成，代码简洁易维护；
  直接用Jedis则需要手动管理连接池、处理异常、实现一致性保障等，代码冗余且容易引入bug。

内容的提问来源于stack exchange，提问作者sclee1