大数据技术之_27_电商平台数据分析项目_02_预备知识 + Scala + Spark Core + Spark SQL + Spark Streaming + Java 对象池

object ListDemo01 {
  def main(args: Array[String]): Unit = {
    // 说明
    // 1. 在默认情况下 List 是 scala.collection.immutable.List 即不可变
    // 2. 在 scala 中，List 就是不可变的，如需要使用可变的 List，则需要使用 ListBuffer
    // 3. List 在 package object scala 中做了声明 val List = scala.collection.immutable.List
    // 4. val Nil = scala.collection.immutable.Nil // List()

    val list01 = List(1, 2, 3, "Hello") // 创建时，直接分配元素
    println(list01) // List(1, 2, 3, Hello)

    val list02 = Nil // 空集合
    println(list02) // List()

    // 访问 List 的元素
    val value1 = list01(1) // 1是索引，表示取出第2个元素
    println("value1=" + value1) // 2


    println("====================list追加元素后的效果====================")
    // 通过 :+ 和 +: 给 list 追加元素(本身的集合并没有变化)
    val list1 = List(1, 2, 3, "abc")
    // :+ 运算符表示在列表的最后增加数据
    val list2 = list1 :+ 4 // (1,2,3,"abc", 4)
    println(list1) // list1 没有变化 (1, 2, 3, "abc")，说明 list1 还是不可变
    println(list2) // 新的列表结果是 [1, 2, 3, "abc", 4]

    val list3 = 10 +: list1 // (10, 1, 2, 3, "abc")
    println("list3=" + list3)

    // :: 符号的使用
    val list4 = List(1, 2, 3, "abc")
    // 说明 val list5 = 4 :: 5 :: 6 :: list4 :: Nil 步骤：
    // 1. List()
    // 2. List(List(1, 2, 3, "abc"))
    // 3. List(6, List(1, 2, 3, "abc"))
    // 4. List(5, 6, List(1, 2, 3, "abc"))
    // 5. List(4, 5, 6, List(1, 2, 3, "abc"))
    val list5 = 4 :: 5 :: 6 :: list4 :: Nil
    println("list5=" + list5)

    // ::: 符号的使用
    // 说明 val list6 = 4 :: 5 :: 6 :: list4 ::: Nil 步骤：
    // 1. List()
    // 2. List(1, 2, 3, "abc")
    // 3. List(6, 1, 2, 3, "abc")
    // 4. List(5, 6, 1, 2, 3, "abc")
    // 5. List(4, 5, 6, 1, 2, 3, "abc")
    // 下面等价 4 :: 5 :: 6 :: list4
    val list6 = 4 :: 5 :: 6 :: list4 ::: Nil
    println("list6=" + list6)
  }
}

List(1, 2, 3, Hello)
List()
value1=2
====================list追加元素后的效果====================
List(1, 2, 3, abc)
List(1, 2, 3, abc, 4)
list3=List(10, 1, 2, 3, abc)
list5=List(4, 5, 6, List(1, 2, 3, abc))
list6=List(4, 5, 6, 1, 2, 3, abc)

object zipTest01 {
  def main(args: Array[String]): Unit = {
    val prices1 = List(5.0, 20.0, 9.95)
    val quantities1 = List(10, 2, 1)
    println(prices1.zip(quantities1))

    println("----------------------------------")

    val prices2 = List(5.0, 20.0, 9.95)
    val quantities2 = List(10, 2)
    println(prices2.zip(quantities2))

    println("----------------------------------")

    val prices3 = List(5.0, 20.0, 9.95)
    val quantities3 = List(10, 2)
    println(prices3.zipAll(quantities3, 9.95, 1))

    println("----------------------------------")

    val prices4 = List(5.0, 20.0, 9.95)
    val quantities4 = List(10, 2, 1)
    println(prices4.zipWithIndex)
    println(quantities4.zipWithIndex)
  }
}

List((5.0,10), (20.0,2), (9.95,1))
----------------------------------
List((5.0,10), (20.0,2))
----------------------------------
List((5.0,10), (20.0,2), (9.95,1))
----------------------------------
List((5.0,0), (20.0,1), (9.95,2))
List((10,0), (2,1), (1,2))

package com.atguigu.session

import org.apache.spark.util.AccumulatorV2

import scala.collection.mutable

/**
  * 自定义累加器
  */
class SessionStatisticAccumulator extends AccumulatorV2[String, mutable.HashMap[String, Int]]() {

  // 自定义累加器：要求要在类的里面维护一个 mutable.HashMap 结构
  val countMap = new mutable.HashMap[String, Int]()

  // 判断累加器是否为空
  override def isZero: Boolean = {
    this.countMap.isEmpty
  }

  // 复制一个一模一样的累加器
  override def copy(): AccumulatorV2[String, mutable.HashMap[String, Int]] = {
    val acc = new SessionStatisticAccumulator
    acc.countMap ++= this.countMap // 将两个 Map 拼接在一起
    acc
  }

  // 重置累加器
  override def reset(): Unit = {
    this.countMap.clear()
  }

  // 向累加器中添加 KV 对（K 存在，V 累加1，K 不存在，重新创建）
  override def add(k: String): Unit = {
    if (!this.countMap.contains(k)) {
      this.countMap += (k -> 0)
    }

    this.countMap.update(k, this.countMap(k) + 1)
  }

  // 两个累加器进行合并（先判断两个累加器是否是同一类型的，再将两个 Map 进行合并(是个小难点)）
  override def merge(other: AccumulatorV2[String, mutable.HashMap[String, Int]]): Unit = {
    other match {
      // (1 : 100).foldLeft(0) 等价于 (0 : (1 to 100))(_+_) 又等价于 { case (int1, int2) => int1 + int2 }
      // acc.countMap.foldLeft(this.countMap) 等价于 this.countMap : acc.countMap  又等价于 this.countMap 和 acc.countMap 的每一个 KV 做操作
      case acc: SessionStatisticAccumulator => acc.countMap.foldLeft(this.countMap) {
        case (map, (k, v)) => map += (k -> (map.getOrElse(k, 0) + v))
      }
    }
  }

  override def value: mutable.HashMap[String, Int] = {
    this.countMap
  }
}

import org.apache.spark.SparkConf
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

case class Person(name: String, age: Int)

object SparkRDDtoDF {

  def main(agrs: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]")
    conf.set("spark.sql.warehouse.dir", "file:D:\\learn\\JetBrains\\workspace_idea3\\commerce_basic\\analyse\\product\\src\\test\\")
    conf.set("spark.sql.shuffle.partitions", "20")

    val sparkSession = SparkSession.builder().appName("RDD to DataFrame").config(conf).getOrCreate()

    // 通过代码的方式，设置 Spark log4j 的级别
    sparkSession.sparkContext.setLogLevel("WARN")

    import sparkSession.implicits._
    // use case class convert RDD to DataFrame
    // 通过反射的方式完成 RDD 向 DataFrame 的转换
    // val peopleDF = rddToDFCase(sparkSession)

    // 通过编程的方式完成 RDD 向 DataFrame 的转换
    val peopleDF = rddToDF(sparkSession)

    peopleDF.show()
    peopleDF.select($"name", $"age").filter($"age" > 20).show()
  }

  // 1. 通过编程获取 Schema
  def rddToDF(sparkSession: SparkSession): DataFrame = {
    // 设置 schema 结构
    val schema = StructType(Seq(StructField("name", StringType, true), StructField("age", IntegerType, true)))
    val rowRDD = sparkSession.sparkContext
      .textFile("file:D:\\learn\\JetBrains\\workspace_idea3\\commerce_basic\\analyse\\product\\src\\test\\people.txt", 2)
      .map(x => x.split(",")).map(x => Row(x(0), x(1).trim().toInt))

    sparkSession.createDataFrame(rowRDD, schema)
  }

  // 2. 通过反射获取 Schema
  def rddToDFCase(sparkSession: SparkSession): DataFrame = {
    // 导入隐饰操作，否则 RDD 无法调用 toDF 方法
    import sparkSession.implicits._
    val peopleRDD = sparkSession.sparkContext
      .textFile("file:D:\\learn\\JetBrains\\workspace_idea3\\commerce_basic\\analyse\\product\\src\\test\\people.txt", 2)
      .map(x => x.split(",")).map(x => Person(x(0), x(1).trim().toInt)).toDF()

    peopleRDD
  }
}

val rdd1=testDF.rdd
val rdd2=testDS.rdd

  import spark.implicits._
  val testDF = rdd.map {
    line =>
      (line._1, line._2)
  }.toDF("col1", "col2")

import spark.implicits._
val testDF = testDS.toDF

import spark.implicits._
// 定义字段名和类型
case class Coltest(col1: String, col2: Int) extends Serializable 
val testDS = testDF.as[Coltest]

package com.atguigu.product

import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, StringType, StructField, StructType}

/**
  * 自定义弱类型的聚合函数（UDAF）
  */
class GroupConcatDistinct extends UserDefinedAggregateFunction {
  // 设置 UDAF 函数的输入类型为 String
  override def inputSchema: StructType = StructType(StructField("cityInfoInput", StringType) :: Nil)

  // 设置 UDAF 函数的缓冲区类型为 String
  override def bufferSchema: StructType = StructType(StructField("cityInfoBuffer", StringType) :: Nil)

  // 设置 UDAF 函数的输出类型为 String
  override def dataType: DataType = StringType

  // 设置一致性检验，如果为 true，那么输入不变的情况下计算的结果也是不变的
  override def deterministic: Boolean = true

  // 初始化自定义的 UDAF 函数
  // 设置聚合中间 buffer 的初始值
  // 需要保证这个语义：两个初始 buffer 调用下面实现的 merge 方法后也应该为初始 buffer，
  // 即如果你初始值是1，然后你 merge 是执行一个相加的动作，两个初始 buffer 合并之后等于 2，不会等于初始 buffer 了，
  // 这样的初始值就是有问题的，所以初始值也叫"zero value"
  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = ""
  }

  // 设置 UDAF 函数的缓冲区更新：实现一个字符串带去重的拼接
  // 用输入数据 input 更新 buffer 值，类似于 combineByKey
  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    // 缓冲中的已经拼接过的城市信息串
    var cityInfoBuffer = buffer.getString(0)
    // 刚刚传递进来的某个城市信息
    val cityInfoInput = input.getString(0)

    // 在这里要实现去重的逻辑
    // 判断：之前没有拼接过某个城市信息，那么这里才可以接下去拼接新的城市信息
    if (!cityInfoBuffer.contains(cityInfoInput)) {
      if ("".equals(cityInfoBuffer)) {
        cityInfoBuffer += cityInfoInput
      } else {
        // 比如 1:北京
        // 1:北京,2:上海
        cityInfoBuffer += "," + cityInfoInput
      }
    }

    buffer.update(0, cityInfoBuffer)
  }

  // 把两个自定义的 UDAF 函数的值合并在一起
  // 合并两个 buffer, 将 buffer2 合并到 buffer1. 在合并两个分区聚合结果的时候会被用到, 类似于 reduceByKey
  // 这里要注意该方法没有返回值，在实现的时候是把 buffer2 合并到 buffer1 中去，你需要实现这个合并细节
  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    // cityInfoBuffer1: cityId1:cityName1, cityId2:cityName2, cityId3:cityName3, ...
    var cityInfoBuffer1 = buffer1.getString(0)
    // cityInfoBuffer2: cityId1:cityName1, cityId2:cityName2, cityId3:cityName3, ...
    val cityInfoBuffer2 = buffer2.getString(0)

    // 将 cityInfoBuffer2 中的数据带去重的加入到 cityInfoBuffer1 中
    for (cityInfo <- cityInfoBuffer2.split(",")) {
      if (!cityInfoBuffer1.contains(cityInfo)) {
        if ("".equals(cityInfoBuffer1)) {
          cityInfoBuffer1 += cityInfo
        } else {
          cityInfoBuffer1 += "," + cityInfo
        }
      }
    }

    buffer1.update(0, cityInfoBuffer1)
  }

  // 计算并返回最终的聚合结果
  override def evaluate(buffer: Row): Any = {
    buffer.getString(0)
  }
}

import org.apache.spark.sql.{Encoder, Encoders}
import org.apache.spark.sql.expressions.Aggregator

// 定义 case 类
case class Employee(name: String, salary: Long)

case class Average(var sum: Long, var count: Long)

object MyAverage extends Aggregator[Employee, Average, Double] {

  /**
    * 计算并返回最终的聚合结果
    */
  def zero: Average = Average(0L, 0L)


  /**
    * 根据传入的参数值更新 buffer 值
    */
  def reduce(buffer: Average, employee: Employee): Average = {
    buffer.sum += employee.salary
    buffer.count += 1 buffer
  }

  /**
    * 合并两个 buffer 值，将 buffer2 的值合并到 buffer1
    */
  def merge(b1: Average, b2: Average): Average = {
    b1.sum += b2.sum
    b1.count += b2.count b1
  }

  /**
    * 计算输出
    */
  def finish(reduction: Average): Double = reduction.sum.toDouble / reduction.count

  /**
    * 设定中间值类型的编码器，要转换成 case 类
    * Encoders.product 是进行 scala 元组和 case 类转换的编码器
    */
  def bufferEncoder: Encoder[Average] = Encoders.product

  /**
    * 设定最终输出值的编码器
    */
  def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("score").setMaster("local[*]")

    val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()

    import sparkSession.implicits._
    val scoreDF = sparkSession.sparkContext
      .makeRDD(Array(
        Score("a1", 1, 80),
        Score("a2", 1, 78),
        Score("a3", 1, 95),
        Score("a4", 2, 74),
        Score("a5", 2, 92),
        Score("a6", 3, 99),
        Score("a7", 3, 99),
        Score("a8", 3, 45),
        Score("a9", 3, 55),
        Score("a10", 3, 78)
      ))
      .toDF("name", "grade", "score")

    scoreDF.createOrReplaceTempView("score")
    scoreDF.show()
  }

+----+-----+-----+
|name|grade|score|
+----+-----+-----+
|  a1|    1|   80|
|  a2|    1|   78|
|  a3|    1|   95|
|  a4|    2|   74|
|  a5|    2|   92|
|  a6|    3|   99|
|  a7|    3|   99|
|  a8|    3|   45|
|  a9|    3|   55|
| a10|    3|   78|
+----+-----+-----+

sparkSession.sql("select name, grade, score, count(name) over() name_count from score").show()

+----+-----+-----+----------+
|name|grade|score|name_count|
+----+-----+-----+----------+
|  a1|    1|   80|        10|
|  a2|    1|   78|        10|
|  a3|    1|   95|        10|
|  a4|    2|   74|        10|
|  a5|    2|   92|        10|
|  a6|    3|   99|        10|
|  a7|    3|   99|        10|
|  a8|    3|   45|        10|
|  a9|    3|   55|        10|
| a10|    3|   78|        10|
+----+-----+-----+----------+

sparkSession.sql("select name, grade, score, count(name) over(partition by grade) name_count from score").show()

+----+-----+-----+----------+
|name|grade|score|name_count|
+----+-----+-----+----------+
|  a1|    1|   80|         3|
|  a2|    1|   78|         3|
|  a3|    1|   95|         3|
|  a6|    3|   99|         5|
|  a7|    3|   99|         5|
|  a8|    3|   45|         5|
|  a9|    3|   55|         5|
| a10|    3|   78|         5|
|  a4|    2|   74|         2|
|  a5|    2|   92|         2|
+----+-----+-----+----------+

sparkSession.sql("select name, grade, score, row_number() over(order by score) as row_number from score").show()
sparkSession.sql("select name, grade, score, rank() over(order by score) as rank from score").show()
sparkSession.sql("select name, grade, score, dense_rank() over(order by score) as dense_rank from score").show()
sparkSession.sql("select name, grade, score, ntile(6) over(order by score) as ntile from score").show()

+----+-----+-----+----------+
|name|grade|score|row_number|
+----+-----+-----+----------+
|  a8|    3|   45|         1|
|  a9|    3|   55|         2|
|  a4|    2|   74|         3|
|  a2|    1|   78|         4|
| a10|    3|   78|         5|
|  a1|    1|   80|         6|
|  a5|    2|   92|         7|
|  a3|    1|   95|         8|
|  a6|    3|   99|         9|
|  a7|    3|   99|        10|
+----+-----+-----+----------+

+----+-----+-----+----+
|name|grade|score|rank|
+----+-----+-----+----+
|  a8|    3|   45|   1|
|  a9|    3|   55|   2|
|  a4|    2|   74|   3|
|  a2|    1|   78|   4|
| a10|    3|   78|   4|
|  a1|    1|   80|   6|
|  a5|    2|   92|   7|
|  a3|    1|   95|   8|
|  a6|    3|   99|   9|
|  a7|    3|   99|   9|
+----+-----+-----+----+

+----+-----+-----+----------+
|name|grade|score|dense_rank|
+----+-----+-----+----------+
|  a8|    3|   45|         1|
|  a9|    3|   55|         2|
|  a4|    2|   74|         3|
|  a2|    1|   78|         4|
| a10|    3|   78|         4|
|  a1|    1|   80|         5|
|  a5|    2|   92|         6|
|  a3|    1|   95|         7|
|  a6|    3|   99|         8|
|  a7|    3|   99|         8|
+----+-----+-----+----------+

+----+-----+-----+-----+
|name|grade|score|ntile|
+----+-----+-----+-----+
|  a8|    3|   45|    1|
|  a9|    3|   55|    1|
|  a4|    2|   74|    2|
|  a2|    1|   78|    2|
| a10|    3|   78|    3|
|  a1|    1|   80|    3|
|  a5|    2|   92|    4|
|  a3|    1|   95|    4|
|  a6|    3|   99|    5|
|  a7|    3|   99|    6|
+----+-----+-----+-----+

sparkSession.sql("select name, grade, score, row_number() over(partition by grade order by score) as row_number from score").show()
sparkSession.sql("select name, grade, score, rank() over(partition by grade order by score) as rank from score").show()
sparkSession.sql("select name, grade, score, dense_rank() over(partition by grade order by score) as dense_rank from score").show()
sparkSession.sql("select name, grade, score, ntile(6) over(partition by grade order by score) as ntile from score").show()

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

case class Score(name: String, grade: Int, score: Int)

object WindowFunctionTest {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("score").setMaster("local[*]")
    val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()

    import sparkSession.implicits._
    val scoreDF = sparkSession.sparkContext
      .makeRDD(Array(
        Score("a1", 1, 80),
        Score("a2", 1, 78),
        Score("a3", 1, 95),
        Score("a4", 2, 74),
        Score("a5", 2, 92),
        Score("a6", 3, 99),
        Score("a7", 3, 99),
        Score("a8", 3, 45),
        Score("a9", 3, 55),
        Score("a10", 3, 78)
      ))
      .toDF("name", "grade", "score")

    scoreDF.createOrReplaceTempView("score")

    scoreDF.show()

    // 1、聚合开窗函数
    // sparkSession.sql("select name, grade, score, count(name) over() name_count from score").show()

    // 显示按照班级分组后每组的人数
    // sparkSession.sql("select name, grade, score, count(name) over(partition by grade) name_count from score").show()


    // 2、排序开窗函数
    sparkSession.sql("select name, grade, score, row_number() over(order by score) as row_number from score").show()
    sparkSession.sql("select name, grade, score, rank() over(order by score) as rank from score").show()
    sparkSession.sql("select name, grade, score, dense_rank() over(order by score) as dense_rank from score").show()
    sparkSession.sql("select name, grade, score, ntile(6) over(order by score) as ntile from score").show()

    sparkSession.sql("select name, grade, score, row_number() over(partition by grade order by score) as row_number from score").show()
    sparkSession.sql("select name, grade, score, rank() over(partition by grade order by score) as rank from score").show()
    sparkSession.sql("select name, grade, score, dense_rank() over(partition by grade order by score) as dense_rank from score").show()
    sparkSession.sql("select name, grade, score, ntile(6) over(partition by grade order by score) as ntile from score").show()
  }
}

package com.atguigu.stream

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Seconds, StreamingContext}

object UpdateStateByKeyWordCount {

  def main(args: Array[String]): Unit = {
    // 构建 Spark 上下文
    val sparkConf = new SparkConf().setAppName("WordCount").setMaster("local[*]")

    // 创建 Spark 客户端
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    val sc = spark.sparkContext
    val ssc = new StreamingContext(sc, Seconds(5))

    // 设置检查点目录
    ssc.checkpoint("./streaming_checkpoint")

    val lines = ssc.socketTextStream("localhost", 9999)
    val words = lines.flatMap(_.split(" "))
    val pairs = words.map(word => (word, 1))

    val wordCount = pairs.updateStateByKey((values: Seq[Int], state: Option[Int]) => {
      var newValue = state.getOrElse(0)
      for (value <- values) {
        newValue += value
      }
      Option(newValue)
    })

    wordCount.print()

    ssc.start()
    ssc.awaitTermination()
  }
}

/**
  * MySQL 客户端代理对象
  *
  * @param jdbcUrl      MySQL URL
  * @param jdbcUser     MySQL 用户
  * @param jdbcPassword MySQL 密码
  * @param client       默认客户端实现
  */
case class MySqlProxy(jdbcUrl: String, jdbcUser: String, jdbcPassword: String, client: Option[Connection] = None) {
  // 获取客户端连接对象
  private val mysqlClient = client getOrElse {
    DriverManager.getConnection(jdbcUrl, jdbcUser, jdbcPassword)
  }

  /**
    * 执行增删改 SQL 语句
    *
    * @param sql
    * @param params
    * @return 影响的行数
    */
  def executeUpdate(sql: String, params: Array[Any]): Int = {
    var rtn = 0
    var pstmt: PreparedStatement = null

    try {
      // 第一步：关闭自动提交
      mysqlClient.setAutoCommit(false)

      // 第二步：根据传入的 sql 语句创建 prepareStatement
      pstmt = mysqlClient.prepareStatement(sql)

      // 第三步：为 prepareStatement 中的每个参数填写数值
      if (params != null && params.length > 0) {
        for (i <- 0 until params.length) {
          pstmt.setObject(i + 1, params(i))
        }
      }

      // 第四步：执行增删改操作
      rtn = pstmt.executeUpdate()

      // 第五步：手动提交
      mysqlClient.commit()
    } catch {
      case e: Exception => e.printStackTrace
    }

    rtn
  }

  /**
    * 执行查询 SQL 语句
    *
    * @param sql
    * @param params
    */
  def executeQuery(sql: String, params: Array[Any], queryCallback: QueryCallback) {
    var pstmt: PreparedStatement = null
    var rs: ResultSet = null

    try {
      // 第一步：根据传入的 sql 语句创建 prepareStatement
      pstmt = mysqlClient.prepareStatement(sql)

      // 第二步：为 prepareStatement 中的每个参数填写数值
      if (params != null && params.length > 0) {
        for (i <- 0 until params.length) {
          pstmt.setObject(i + 1, params(i))
        }
      }

      // 第三步：执行查询操作
      rs = pstmt.executeQuery()

      // 第四步：处理查询后的结果
      queryCallback.process(rs)
    } catch {
      case e: Exception => e.printStackTrace
    }
  }

  /**
    * 批量执行 SQL 语句
    *
    * @param sql
    * @param paramsList
    * @return 每条SQL语句影响的行数
    */
  def executeBatch(sql: String, paramsList: Array[Array[Any]]): Array[Int] = {
    var rtn: Array[Int] = null
    var pstmt: PreparedStatement = null
    try {
      // 第一步：关闭自动提交
      mysqlClient.setAutoCommit(false)

      pstmt = mysqlClient.prepareStatement(sql)

      // 第二步：为 prepareStatement 中的每个参数填写数值
      if (paramsList != null && paramsList.length > 0) {
        for (params <- paramsList) {
          for (i <- 0 until params.length) {
            pstmt.setObject(i + 1, params(i))
          }
          pstmt.addBatch()
        }
      }

      // 第三步：执行批量的 SQL 语句
      rtn = pstmt.executeBatch()

      // 第四步：手动提交
      mysqlClient.commit()
    } catch {
      case e: Exception => e.printStackTrace
    }

    rtn
  }

  // 关闭 MySQL 客户端
  def shutdown(): Unit = mysqlClient.close()
}

/**
  * 扩展知识：将 MySqlProxy 实例视为对象，MySqlProxy 实例的创建使用对象池进行维护
  *
  * 创建自定义工厂类，继承 BasePooledObjectFactory 工厂类，负责对象的创建、包装和销毁
  *
  * @param jdbcUrl
  * @param jdbcUser
  * @param jdbcPassword
  * @param client
  */
class PooledMySqlClientFactory(jdbcUrl: String, jdbcUser: String, jdbcPassword: String, client: Option[Connection] = None)
  extends BasePooledObjectFactory[MySqlProxy] with Serializable {
  // 用于池来创建对象
  override def create(): MySqlProxy = MySqlProxy(jdbcUrl, jdbcUser, jdbcPassword, client)

  // 用于池来包装对象
  override def wrap(obj: MySqlProxy): PooledObject[MySqlProxy] = new DefaultPooledObject(obj)

  // 用于池来销毁对象
  override def destroyObject(p: PooledObject[MySqlProxy]): Unit = {
    p.getObject.shutdown()
    super.destroyObject(p)
  }
}

/**
  * 创建 MySQL 池工具类
  */
object CreateMySqlPool {
  // 加载 JDBC 驱动，只需要一次
  Class.forName("com.mysql.jdbc.Driver")

  // 在 org.apache.commons.pool2.impl 中预设了三个可以直接使用的对象池：GenericObjectPool、GenericKeyedObjectPool 和 SoftReferenceObjectPool
  // 创建 genericObjectPool 为 GenericObjectPool
  // GenericObjectPool 的特点是可以设置对象池中的对象特征，包括 LIFO 方式、最大空闲数、最小空闲数、是否有效性检查等等
  private var genericObjectPool: GenericObjectPool[MySqlProxy] = null

  // 伴生对象通过 apply 完成对象的创建
  def apply(): GenericObjectPool[MySqlProxy] = {
    // 单例模式
    if (this.genericObjectPool == null) {
      this.synchronized {
        // 获取 MySQL 配置参数
        val jdbcUrl = ConfigurationManager.config.getString(Constants.JDBC_URL)
        val jdbcUser = ConfigurationManager.config.getString(Constants.JDBC_USER)
        val jdbcPassword = ConfigurationManager.config.getString(Constants.JDBC_PASSWORD)
        val size = ConfigurationManager.config.getInt(Constants.JDBC_DATASOURCE_SIZE)

        val pooledFactory = new PooledMySqlClientFactory(jdbcUrl, jdbcUser, jdbcPassword)
        val poolConfig = {
          // 创建标准对象池配置类的实例
          val c = new GenericObjectPoolConfig
          // 设置配置对象参数
          // 设置最大对象数
          c.setMaxTotal(size)
          // 设置最大空闲对象数
          c.setMaxIdle(size)
          c
        }

        // 对象池的创建需要工厂类和配置类
        // 返回一个 GenericObjectPool 对象池
        this.genericObjectPool = new GenericObjectPool[MySqlProxy](pooledFactory, poolConfig)
      }
    }

    genericObjectPool
  }
}