一.引言
现在有一批数据写入多台 Redis 相同 key 的队列中，需要消费 Redis 队列作为 Flink Source，为了提高可用性，下面基于 JedisPool 进行队列的消费。队列数据示例: 1,2,3,4,5、A,B,C,D,E，程序将字符串解析并 split(",") 然后分别写到下游。
二.Flink Source By JedisPool
1.初始化 JedisPool
由于数据量较大，所以同时写入 N 台 Redis 队列，key 均相同，注意这里是 JedisPool 不是 JedisCluster，需要区分二者的概念。
def initJedisPool(host: String, port: Int): JedisPool = { val config = new JedisPoolConfig config.setMaxTotal(4) config.setMaxIdle(2) config.setMaxWaitMillis(1000) config.setTestOnBorrow(true) config.setTestOnReturn(true) jedisPool = new JedisPool(config, host, port) jedisPool }
需要导入 Jedis 依赖:
<dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>2.7.2</version> </dependency>
这里 JedisPool 的几个参数可以参考之前的文章：JedisPool - Java.net.SocketException: Broken pipe (write failed)，里面介绍了 TestOnBorrow、TestOnReturn 的含义。
2.实现 RedisSource
自定义 Source 需要继承 org.apache.flink.streaming.api.functions.source.RichSourceFunction 实现 run 方法和 cancel 方法即可，主要逻辑在 run 方法中，run 方法负责从队列中不断获取数据并 collect 到下游，除此之外，由于需要读取 redis client，所以还需要新增 open 方法进行 client 的初始化。
class JedisPoolSourceTest(host: String, port: Int) extends RichSourceFunction[String] { var jedisPool: JedisPool = _ // JedisPool val listKey = "testListKey" // 公用队列 Key override def open(parameters: Configuration): Unit = { initJedisPool(host, port) // 初始化 JedisPool } // 消费 Redis 队列产出数据 override def run(sourceContext: SourceFunction.SourceContext[String]): Unit = { collectData(sourceContext) } // 关闭 Source override def cancel(): Unit = { jedisPool.close() } }
3.产出数据 CollectData
上面是继承 RichSourceFunction 完成了整体的架构，下面单独把 collectData 的数据产出逻辑分析一下:
def collectData(out: SourceFunction.SourceContext[String]): Unit = { var resource: Jedis = null try { resource = jedisPool.getResource // 获取 Jedis 实例 // while True 保证持续消费 while (true) { // 解析并产出 val info = resource.lpop(listKey) if (info != null) { val sendRe = info.split(",") if (sendRe.nonEmpty) { sendRe.foreach(out.collect) } } // 如果队列为空则 Sleep 1s if (resource.llen(listKey) == 0) { TimeUnit.SECONDS.sleep(1) } } } catch { case e: Exception => { e.printStackTrace() resource.close() // 关闭资源 collectData(out) // 出现异常递归重启，保证 Source 的可用性 } } }
collectData 为 run 方法的主类，主要用于消费队列并产出数据到下游，有几个点需要注意：
A.获取 Jedis 实例
由于本例中使用 JedisPool，所以这里采用 pool.getResource 的方式获取 Jedis，网上也有一些 demo 在这里直接初始化单独 JedisClient，这里采用 pool 的形式主要是考虑到稳定性的情况。
B.While True
这里没啥太多说的，通过 while true 实现不间断的消费队列数据。
C.Sleep(1)
这里考虑到队列为空时，如果频繁 while true 访问 Jedis 会造成高 QPS 且无用操作，所以这里检测到队列为空时会加入 1s 的延迟，这个时间也可以根据自己的任务场景灵活修改。
D.Try Catch + 递归
try-catch 逻辑内除关闭 resource 外还递归调用了 CollectData，该方法参考了 Spark-Streaming 的 receiver 方法，在故障时能够重新申请 redis 连接并重新读取数据，保证 Source 的稳定性，Spark-Streaming 的 receiver 可以参考：Spark Streaming Receiver restart 重启。
4.合并多条数据源
由于数据量较大，所以数据分布式的写入到多台 Redis 队列中，上面已经实现了 JedisSource，下面则需要将多台 Redis 分别接入 JedisSource 并绑定合成一个统一的 Source。
var dataStream: DataStream[String] = null // 初始化原始流 // 遍历多台 Redis 的 Host && Port testHostAndPorts.foreach { case (host, port) => { if (dataStream == null) { dataStream = env.addSource[String](new JedisPoolSourceTest(host, port)) } else { val newStream = env.addSource[String](new JedisPoolSourceTest(host, port)) dataStream = dataStream.union(newStream) // union 合并流 } } }
通过 var 构造可变变量，然后不断将 RedisSource union 到 DataStream 中，最终形成统一流。
三.测试多 Redis Source
1.启动多台 client
启动多个 Jedis 连接并向队列 key 推数据:
redis-cli -h $host -p $port
在不同客户端分别执行:
lpush testListKey 0,1,2,3,4 lpush testListKey A,B,C,D,E
2.主函数 main
主函数逻辑比较简单，主要就是接受队列数据，"," split 并将每个元素输出，最终 print sink 到标准输出。
def main(args: Array[String]): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment // 执行环境 // 合并多流 var dataStream: DataStream[String] = null testHostAndPorts.foreach { case (host, port) => { if (dataStream == null) { dataStream = env.addSource[String](new JedisPoolSourceTest(host, port)) } else { val newStream = env.addSource[String](new JedisPoolSourceTest(host, port)) dataStream = dataStream.union(newStream) } } } // 标准输出 dataStream.print() // 执行 env.execute()
上面通过 2 台 client 分别 lpush 了数据，这边接收测试也没有问题：
编辑
四.尝试与优化
1.多线程的尝试
上面采用 JedisPool + 单 Resource 的形式进行接受数据，其实最早使用 JedisPool 是为了接入多线程 Source，即在 run 方法内使用线程池提高生产效率：
def addTask(poolSize: Int, jedisPool: JedisPool, sourceContext: SourceFunction.SourceContext[String]): Unit = { val executor = Executors.newFixedThreadPool(2) // 初始化线程池 (0 until poolSize).foreach(epoch => { executor.submit(new Runnable { override def run(): Unit = { val resource: Jedis = jedisPool.getResource while (true) { val info = resource.lpop(listKey) if (info != null) { info.split(",").foreach(sourceContext.collect) } if (resource.llen(listKey) == 0) { TimeUnit.SECONDS.sleep(1) } } } }) }) }
这里 Runnable 和上面单 Resource 接队列消费逻辑相同，实现后发现可以多线程读取到 Jedis 队列的内容，随后程序发生堵塞，经过调试发现任务卡在 sourceContext.collect，不知道是不是 sourceContext 不支持多线程，有了解的同学也可以评论区讨论一下~
2.空闲数据源的优化
上述多台 redis 队列并不能保证全天无间断都有数据产出，在 EventTime 场景下，会出现流数据空闲的状态从而影响 WaterMark 的更新影响整个任务的时间推进，为了解决这个问题，可以使用新版的时间戳策略即 WatermarkStrategy：
WatermarkStrategy .forBoundedOutOfOrderness[T](Duration.ofSeconds(60)) .withIdleness(Duration.ofMinutes(1))
通过 org.apache.flink.api.common.eventtime.WatermarkStrategy 的 withIdleness 方法对数据流进行标记，当数据源在 Duration 规定的时间内未产出数据时将该 Source 标记为空闲状态，这样下游的数据也不需要等待该 Source 的 WaterMark 从而保证数据流的正常推进，待有新数据推入 Source 中，该数据流切换为活跃状态，重新向下游发送其真实 WaterMark 水印。
3.数据量过大
对于数据过多或过大的场景，lpop 可能会有一定的读取和网络压力，这时候可以采用 lrange + ltrim 的批量读取逻辑，将单条访问改为批量访问，减少 redis 读取压力。其思想与 hscan 代替 hgetAll 批量访问过大数据有一定类似，都是采取分治的思想，只不过前者是由一到多，后者是化整为零。
五.总结
绑定多台 Redis 源上线后，任务没有问题且支持空流处理，除了 Redis Source 外，还有 Redis Sink 相关的实现，大家可以参考: Flink / Scala - 使用 RedisSink 存储数据，这里使用 SharedJedisPool 代替了 Flink 自带的 RedisCommandsContainer，后续也会单独出一期 RedisCommandsContainer 的 Flink-Jedis 教程。
编辑
可以看到多台 Redis Source 上线后，这个作业图实在是 🚄。
