最近在写spark调用redis的时候总是遇到如下错误：

18/03/28 16:36:40 ERROR yarn.ApplicationMaster: User class threw exception: java.lang.IllegalStateException: failed to create a child event loop
java.lang.IllegalStateException: failed to create a child event loop
	at com.lambdaworks.io.netty.util.concurrent.MultithreadEventExecutorGroup.<init>(MultithreadEventExecutorGroup.java:68)
	at com.lambdaworks.io.netty.channel.MultithreadEventLoopGroup.<init>(MultithreadEventLoopGroup.java:49)
	at com.lambdaworks.io.netty.channel.nio.NioEventLoopGroup.<init>(NioEventLoopGroup.java:68)
	at com.lambdaworks.io.netty.channel.nio.NioEventLoopGroup.<init>(NioEventLoopGroup.java:63)
	at com.lambdaworks.io.netty.channel.nio.NioEventLoopGroup.<init>(NioEventLoopGroup.java:54)
	at com.lambdaworks.redis.resource.DefaultEventLoopGroupProvider.createEventLoopGroup(DefaultEventLoopGroupProvider.java:125)
	at com.lambdaworks.redis.resource.DefaultEventLoopGroupProvider.getOrCreate(DefaultEventLoopGroupProvider.java:99)
	at com.lambdaworks.redis.resource.DefaultEventLoopGroupProvider.allocate(DefaultEventLoopGroupProvider.java:48)
	at com.lambdaworks.redis.AbstractRedisClient.getEventLoopGroup(AbstractRedisClient.java:183)
	at com.lambdaworks.redis.AbstractRedisClient.channelType(AbstractRedisClient.java:167)
	at com.lambdaworks.redis.RedisClient.connectAsync(RedisClient.java:448)
	at com.lambdaworks.redis.RedisClient.connectAsync(RedisClient.java:429)
	at com.lambdaworks.redis.RedisClient.connect(RedisClient.java:371)
	at com.lambdaworks.redis.RedisClient.connect(RedisClient.java:344)
	at com.lambdaworks.redis.RedisClient.connect(RedisClient.java:329)
	at Util.Redis$.Init(Redis.scala:28)
	at Util.Redis$.getResource(Redis.scala:34)
	at LogTools.Format.Normal.BasicTask$.FileIsProcessed(BasicTask.scala:97)
	at LogTools.Format.Normal.BasicTask$.filesNotUpdate(BasicTask.scala:635)
	at LogTools.Format.Normal.BasicTask$.processSplitFile(BasicTask.scala:257)
	at LogTools.Format.Normal.BasicTask$.main(BasicTask.scala:221)
	at LogTools.Format.Normal.BasicTask.main(BasicTask.scala)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:606)
	at org.apache.spark.deploy.yarn.ApplicationMaster$$anon$2.run(ApplicationMaster.scala:552)
Caused by: com.lambdaworks.io.netty.channel.ChannelException: failed to open a new selector
	at com.lambdaworks.io.netty.channel.nio.NioEventLoop.openSelector(NioEventLoop.java:151)
	at com.lambdaworks.io.netty.channel.nio.NioEventLoop.<init>(NioEventLoop.java:142)
	at com.lambdaworks.io.netty.channel.nio.NioEventLoopGroup.newChild(NioEventLoopGroup.java:94)
	at com.lambdaworks.io.netty.util.concurrent.MultithreadEventExecutorGroup.<init>(MultithreadEventExecutorGroup.java:64)
	... 26 more
Caused by: java.io.IOException: Too many open files
	at sun.nio.ch.IOUtil.makePipe(Native Method)
	at sun.nio.ch.EPollSelectorImpl.<init>(EPollSelectorImpl.java:65)
	at sun.nio.ch.EPollSelectorProvider.openSelector(EPollSelectorProvider.java:36)
	at com.lambdaworks.io.netty.channel.nio.NioEventLoop.openSelector(NioEventLoop.java:149)
	... 29 more

原因：在init里面创建redis的链接之后，用完之后没有释放资源，没有close将链接释放，导致在程序在创建过多的redis链接之后，就报too many open files 的错误。最开始我用得是com.lambdaworks.redis来调用redis，用的时候发现这个包在调用close()之后没有释放connection,导致redis的链接越来越多。

修改：使用连接池来维护redis的connection，或者我最后用的是jedis来调用redis。

参考代码如下：

object JRedis extends Serializable{
	case class RedisConfig(
		                      Host: String,
		                      Port: Int,
		                      Password: String
	                      ) extends Serializable
	@transient
	private var j: Jedis = _
	private def Init(host: String, port: Int, password: String): Unit = {
		if (j != null) return
		j = new Jedis(host, port)
		if (password.nonEmpty) j.auth(password)
		j.select(0)
	def getResource(RedisConfig: RedisConfig) = {
		this.synchronized {
			Init(RedisConfig.Host: String, RedisConfig.Port: Int, RedisConfig.Password: String)
	def updateValueByKey(key: String, value: Long, RedisConfig: RedisConfig): Unit = {
		val redis = getResource(RedisConfig: RedisConfig)
		redis.incrBy(key, value)
		redis.close
	def getValueByKey(key: String, RedisConfig: RedisConfig): Long = {
		val redis = getResource(RedisConfig: RedisConfig)
		val num = redis.get(key).toLong
		redis.close()
                    文章最后发布于: 2018-04-17