netty 大批量发送数据的时候 出现channel关闭异常?报错

用netty建立的长链接，在长链接过多，大批量发送数据的时候。发现好多消息发送失败，并且链接断开很多。

报错的的日志：

    19:23:18.036 [nioEventLoopGroup-3-10] ERROR com.suning.live.service.MsgServiceImpl - chat msg  发送失败, msg is : {"content":"这是测试内容","fromCustNum":"test","fromHeadPic":"test","fromNickName":"test","level":1,"msgType":"text","session_id":"755069b917624224bdefd3a70c3d6c4b","toChatRoomId":"lyRoom","type":"Push2ChatRoom"}     cause: java.nio.channels.ClosedChannelException: null

代码部分：

final Set<String> custNums = redisProcess.smembers(RedisKeys.ROOM_ID_KEY_PREFIXX + roomId);
        LOGGER.info("准备发送chat msg给client， 接收消息的custNum为 :" + custNums);
        if (null == custNums || custNums.size() <= 0) {
            return;
        }
        taskExecutor.execute(new Runnable() {
            @Override
            public void run() {
                for (String custNum : custNums) {
                     sendMsgToCustNum(msg, custNum);
                }
            }
        });

private void sendMsgToCustNum(final String msg, String toCustNum) {
        try {
            Channel channel = channelStore.getChannel(toCustNum);
            if (null != channel && channel.isActive()) {
                channel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (!future.isSuccess()) {
                            LOGGER.error("chat msg  发送失败, msg is : {}     cause: ", msg, future.cause());
                        }
                    }
                });
            }
        } catch (Exception e) {
            LOGGER.error("chat msg 发送失败 wrapper, msg is : {}", e);
        }
    }

netty的链接关闭异常方法回调并没有进入。inactive channel的日志有打印
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        LOGGER.error("Unexpected exception from downstream,channel" + ctx.channel() + " will be closed, exception is "
                + cause);
        // 当有异常时，关闭channel
        ctx.close();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        LOGGER.info("Channel " + ctx.channel() + " inactive.");
        // 删除channelGroup中的已关闭channel
        channelClosed(ctx);
        super.channelInactive(ctx);
    }