本文是继(Spark 要引入外部数据源，咋整？——(中))的下篇，为了不让你对标题产生审美疲劳，我特意换了个名字。

熟悉「ThreadLocal」的同学应该都知道，这个知识点在纯 Java 界，已经被各种文章讲烂了，但是，这玩意跟 Spark 的结合，却鲜有文章提及。

那今天这篇内容呢，就结合我的实践经历，来跟你聊聊，Spark 怎么样用它，来优雅地解决外部数据源 Redis(其他数据库也一样) 的引入问题。

0. 为什么不能放开连接

前面文章提到，为了降低 Redis 的连接管理成本，于是我们考虑使用「连接单例」，但是如果直接只使用这个连接单例，就会导致在「多线程」情况下，对单个连接的争抢问题(Spark 如果要引入外部数据源，咋整？——(上))。

这个时候，有同学可能觉得，那我根据线程需要来 new 不就好了吗，比如像下面这种：

.foreachPartition(iter => {
                val connection = new RedisConnection()
                while (iter.hasNext){
                    connection...
                    //todo
                }
              }

这样写，确实可以解决问题。

但是，如果在团队中，万一有些水平比较蹩脚的同学，手一哆嗦，就给写成了下面这样：

.foreach(data => {
                val connection = new RedisConnection()
                connection...
                //todo
            })

就问你慌不慌？

根据我的经验，但凡像这种需要你很小心去注意的代码逻辑，一定要把它封装成「工具类」，供团队小伙伴统一使用，否则，你把这些地方的自主权交给个人，指不定会冒出什么样的牛鬼蛇神，分分钟把你的数据库给干崩掉。

1. ThreadLocal 能解决什么

既然连接单例有缺点，而完全放开连接管理给开发的同学，又有失控的风险(就算用线程池，也一样有这个风险)。

那有没有一种，既能控制数据库的连接个数，又可以完全满足多线程情况下，对连接数量的动态需求呢？

有！连接单例 + ThreadLocal 组合。

关于 ThreadLocal 的概念，网上的资料已经烂大街了，这里不赘述，我只想告诉你，它在当前场景里解决问题是：我只需要在一个进程中 new 一个连接对象，然后通过 ThreadLocal 给「按需」复制到各个线程里面，让每个线程都可以拥有属于自己的那份连接。

这里最妙的地方，就在于「按需」这两个字，在代码实现上，你虽然只 new 了一个连接，但如果你在启动任务时，想让每个 executor 并行多跑几个 partition 的数据，这个时候，它就可以根据每个 executor 并行的 partition 数量，自动来创建等量的数据库连接对象。

来，看一眼 Redis 连接单例，跟 ThreadLocal 组合的核心代码：

定义全局的连接对象单例

获取装有 Redis 连接对象的 ThreadLocal 对象

2. 正式代码

Spark 的主体代码如下：

package com.anryg.bigdata.test.variable

import com.anryg.bigdata.RedisClientUtilsV2
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

/**
  * @DESC: 测试Spark利用Redis的连接单例 + ThreadLocal来引入外部数据源
  * @Auther: Anryg
  * @Date: 2024/7/9 17:45
  */
object SparkThreadLocalTest03 {

    def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setAppName("SparkThreadLocalTest03")
        val spark = SparkSession.builder().config(sparkConf).getOrCreate()

        /**构造数据源*/
        val seq = Seq("baidu.com",
            "qq.com",
            "xiaomi.com",
            "12345.com",
            "abc.com",
            "def.net",
            "xxx.com",
            "google.com",
            "sina.com",
            "163.com",
            "yyy.net",
            "360.com",
            "cctv.cn",
            "12306.com")

        spark.sparkContext.makeRDD(seq) /**生成 RDD 对象*/
            .foreach(domain => {
                /**通过ThreadLocal获取Redis连接，并拿到黑名单数据*/
                val blackDomainSet = RedisClientUtilsV2.getSetResult(RedisClientUtilsV2.getRedisThreadLocal.get(),
                    5,
                    "black_domain")

            if (!blackDomainSet.contains(domain)) {
                Thread.sleep(5000) /**为了方便观察Redis的连接情况，sleep几秒*/
                println(domain)
            }
        })
    }
}