获取kafka和redis连接的工具类

蔚落 2022-10-27 01:36 229阅读 0赞

## 1）PropertiesUtil 获取配置文件的工具类 ##

import java.io.InputStreamReader
    import java.util.Properties
    
    object PropertiesUtil { 
    
       def load(propertieName:String): Properties ={ 
         val prop=new Properties()
         prop.load(new InputStreamReader(Thread.currentThread().getContextClassLoader.getResourceAsStream(propertieName) , "UTF-8"))
         prop
       }
    }

## 2）MykafkaUtil 获取kafak连接的工具类 ##

import java.util.Properties
    
    import org.apache.kafka.clients.consumer.ConsumerRecord
    import org.apache.kafka.common.serialization.StringDeserializer
    import org.apache.spark.streaming.StreamingContext
    import org.apache.spark.streaming.dstream.InputDStream
    import org.apache.spark.streaming.kafka010.{ ConsumerStrategies, KafkaUtils, LocationStrategies}
    
    object MyKafkaUtil { 
    
      //1.创建配置信息对象
      private val properties: Properties = PropertiesUtil.load("config.properties")
    
      //2.用于初始化链接到集群的地址
      val broker_list: String = properties.getProperty("kafka.broker.list")
    
      //3.kafka消费者配置
      val kafkaParam = Map(
        "bootstrap.servers" -> broker_list,
        "key.deserializer" -> classOf[StringDeserializer],
        "value.deserializer" -> classOf[StringDeserializer],
        //消费者组
        "group.id" -> "bigdata2020",
        //如果没有初始化偏移量或者当前的偏移量不存在任何服务器上，可以使用这个配置属性
        //可以使用这个配置，latest自动重置偏移量为最新的偏移量
        "auto.offset.reset" -> "latest",
        //如果是true，则这个消费者的偏移量会在后台自动提交,但是kafka宕机容易丢失数据
        //如果是false，会需要手动维护kafka偏移量
        "enable.auto.commit" -> (true: java.lang.Boolean)
      )
    
      // 创建DStream，返回接收到的输入数据
      // LocationStrategies：根据给定的主题和集群地址创建consumer
      // LocationStrategies.PreferConsistent：持续的在所有Executor之间分配分区
      // ConsumerStrategies：选择如何在Driver和Executor上创建和配置Kafka Consumer
      // ConsumerStrategies.Subscribe：订阅一系列主题
      def getKafkaStream(topic: String, ssc: StreamingContext): InputDStream[ConsumerRecord[String, String]] = { 
    
        val dStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
            ssc,
            LocationStrategies.PreferConsistent,
            ConsumerStrategies.Subscribe[String, String](Array(topic), kafkaParam))
    
        dStream
      }
    }

## 3）RedisUtil 获取redis连接的工具类 ##

import java.util.Properties
    
    import redis.clients.jedis.{ Jedis, JedisPool, JedisPoolConfig}
    
    object RedisUtil { 
    
      var jedisPool: JedisPool = _
    
      def getJedisClient: Jedis = { 
        if (jedisPool == null) { 
          println("开辟一个连接池")
          val config: Properties = PropertiesUtil.load("config.properties")
          val host: String = config.getProperty("redis.host")
          val port: String = config.getProperty("redis.port")
    
          val jedisPoolConfig = new JedisPoolConfig()
          jedisPoolConfig.setMaxTotal(100) //最大连接数
          jedisPoolConfig.setMaxIdle(20) //最大空闲
          jedisPoolConfig.setMinIdle(20) //最小空闲
          jedisPoolConfig.setBlockWhenExhausted(true) //忙碌时是否等待
          jedisPoolConfig.setMaxWaitMillis(500) //忙碌时等待时长 毫秒
          jedisPoolConfig.setTestOnBorrow(true) //每次获得连接的进行测试
    
          jedisPool = new JedisPool(jedisPoolConfig, host, port.toInt)
        }
        println(s"jedisPool.getNumActive = ${jedisPool.getNumActive}")
        println("获得一个连接")
        jedisPool.getResource
      }
    }