本文包含redis非集群 和 集群两种方式
有些大佬写的实在太牛，恕我能力不足看不太懂，所以自己尝试了一下
非集群方式
非集群方式 工具类如下：object RedisUtil {
private val jedisPoolConfig: JedisPoolConfig = new JedisPoolConfig()
jedisPoolConfig.setMaxTotal(200) //最大连接数
jedisPoolConfig.setMaxIdle(20) //连接池中最大空闲的连接数
jedisPoolConfig.setMinIdle(20) //最小空闲
jedisPoolConfig.setBlockWhenExhausted(true) //忙碌时是否等待
jedisPoolConfig.setMaxWaitMillis(2000) //忙碌时等待时长 毫秒
jedisPoolConfig.setTestOnBorrow(false) //每次获得连接的进行测试
private val jedisPool: JedisPool = new JedisPool(jedisPoolConfig, "xxx.xxx.xxx.xxx", xxx)
// 直接得到一个 Redis 的连接
def getJedisClient: Jedis = {
jedisPool.getResource
}
//测试通过
def main(args: Array[String]): Unit = {
// println(getJedisClient.hget("flink_redis", "1"))
getJedisClient.hset("flink_redis","20","苏州")
println(getJedisClient.exists("flink_redis"))
println(getJedisClient.exists("flink_redis_false"))
getJedisClient.hkeys("flink_redis")
}
}
测试如下
def main(args: Array[String]): Unit = {
getJedisClient.hset("key","field","value")
println(getJedisClient.exists("key"))
getJedisClient.hkeys("key")
}
直接使用getJedisClient 就可以获取所有redis的方法
由于我是集群，使用的时候，直接改 会报错
redis.clients.jedis.exceptions.JedisMovedDataException: MOVED 1539 xxx.xxx.xxx.xxx:xxx
所以要修改一些
集群方式 工具类object RedisUtil {
private val jedisPoolConfig: JedisPoolConfig = new JedisPoolConfig()
jedisPoolConfig.setMaxTotal(200) //最大连接数
jedisPoolConfig.setMaxIdle(20) //连接池中最大空闲的连接数
jedisPoolConfig.setMinIdle(20) //最小空闲
jedisPoolConfig.setBlockWhenExhausted(true) //忙碌时是否等待
jedisPoolConfig.setMaxWaitMillis(2000) //忙碌时等待时长 毫秒
jedisPoolConfig.setTestOnBorrow(false) //每次获得连接的进行测试
val nodes: util.Set[HostAndPort] = new util.LinkedHashSet[HostAndPort]
//这样写方便后面读取配置文件使用
val nodesList:String = "172.8.10.105:7000,172.8.10.105:7001,172.8.10.106:7002,172.8.10.106:7003,172.8.10.107:7004,172.8.10.107:7005"
nodesList.split(",").foreach(s=>{
val sp: Array[String] = s.split(":")
nodes.add(new HostAndPort(sp.apply(0),sp.apply(1).toInt))
})
private val cluster: JedisCluster = new JedisCluster(nodes, jedisPoolConfig)
//hashGet方法，以下 以此重写其他方法
def hashGet(key:String, field:String):String ={
cluster.hget(key,field)
}
}
其他的方法可以手动重写
测试如下 println(RedisUtil.hashGet("key","field"))
最后补充一下需要的依赖
redis.clients jedis 2.9.0 org.apache.bahir flink-connector-redis_2.11 1.0 org.apache.flink flink-streaming-java_2.11
