为什么Netty变慢了？

疯狂的KK

发布于 2020-01-17 15:34:11

2K00

代码可运行

文章被收录于专栏：Java项目实战Java项目实战

运行总次数：0

代码可运行

体验过众多Netty的demo以后，在跟朋友分享后实现了页面昵称输入，消息可回车发送，页面保持输入框在底部，消息页面支持滚动，但是为什么当多客户端接入，消息会延迟，并注册变慢呢？

传统JAVA BIO 阻塞+同步

特点:

1.客户端启动一个socket 每个客户建立一个链接

2.判断服务器是否有线程响应，没有会等待或被拒绝

3.有线程，等待请求响应结束

体验代码

package com.chat.util;

import java.io.IOException;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.time.LocalDate;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author zhaokkstart
 * @create 2020-01-12 21:00
 */
public class Bio {

    public static void main(String[] args) throws IOException {

        ExecutorService executorService = Executors.newCachedThreadPool();

        ServerSocket serverSocket = new ServerSocket(8088);
        System.out.println("服务器启动....");
        while (true) {
            final Socket accept = serverSocket.accept();
            System.out.println("客户端接入....."+ LocalDate.now());
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    handler(accept);
                }
            });

        }
    }

    public static void handler(Socket socket) {
        try {
            byte[] bytes = new byte[1024];
            InputStream inputStream = socket.getInputStream();
            while (true) {
                int read = inputStream.read(bytes);
                if (read != -1) {
                    System.out.println("接受到数据" + new String(bytes, 0, read));
                } else {
                    System.out.println("读取完毕");
                    break;
                }

            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            System.out.println("关闭客户端连接");
            try {
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }
}

启动：

cmd  telnet 127.0.0.1 8088
ctrl ]
send hello netty

Connected to the target VM, address: '127.0.0.1:51231', transport: 'socket'
服务器启动....
客户端接入.....2020-01-12
接受到数据hello netty

同样打开一个窗口发送hello netty2

客户端接入.....2020-01-12
接受到数据hello netty 2

而Nio是如何做的呢？

NIO 非阻塞IO java1.4

channel buffer Selector

线程Thread

↓

Selector 根据不同的事件在各个channel切换

↓

Channel(read/write) 多个连接 Event决定切换到那个channel

↑↓

Buffer buffer底层有一个数组数据的读/写双向的flip切换

↓

Client(socket）

目前没有数据可用就什么都不做，不会阻塞

Http2.0使用多路IO复用技术同一个技术处理多个请求

public class BasicBuffer {

    public static void main(String[] args) {
        IntBuffer intBuffer =IntBuffer.allocate(5);
        intBuffer.put(10);
        intBuffer.put(11);
        intBuffer.put(12);
        intBuffer.put(13);
        intBuffer.put(14);
        //读取数据  读写切换
        intBuffer.flip();
        while (intBuffer.hasRemaining()){
            System.out.println(intBuffer.get());
        }
    }
}

Buffer源码

public abstract class Buffer {

    /**
     * The characteristics of Spliterators that traverse and split elements
     * maintained in Buffers.
     */
    static final int SPLITERATOR_CHARACTERISTICS =
        Spliterator.SIZED | Spliterator.SUBSIZED | Spliterator.ORDERED;

    // Invariants: mark <= position <= limit <= capacity
    mark标记
    private int mark = -1;
     position下一次执行位置
    private int position = 0;
    放入长度限制
    private int limit;
    容量
    private int capacity;

    // Used only by direct buffers
    // NOTE: hoisted here for speed in JNI GetDirectBufferAddress
    long address;
}

Buffer子类

ByteBuffer，存储字节数据到缓冲区
ShortBuffer，存储字符串数据到缓冲区
CharBuffer，存储字符数据到缓冲区
IntBuffer，存储整数数据到缓冲区
LongBuffer，存储长整型数据到缓冲区
DoubleBuffer，存储小数到缓冲区
FloatBuffer，存储小数到缓冲区

以上子类中每一个源码中都有一个数组

    final int[] hb;                  // Non-null only for heap buffers
    final int offset;
    boolean isReadOnly;

对比StringBuffer

StringBuilder 相较于 StringBuffer 有速度优势，所以多数情况下建议使用 StringBuilder 类。然而在应用程序要求线程安全的情况下，则必须使用 StringBuffer 类。

如何做到线程安全的？

@Override
    public synchronized int length() {
        return count;
    }

    @Override
    public synchronized int capacity() {
        return value.length;
    }


    @Override
    public synchronized void ensureCapacity(int minimumCapacity) {
        super.ensureCapacity(minimumCapacity);
    }

    /**
     * @since      1.5
     */
    @Override
    public synchronized void trimToSize() {
        super.trimToSize();
    }

    /**
     * @throws IndexOutOfBoundsException {@inheritDoc}
     * @see        #length()
     */
    @Override
    public synchronized void setLength(int newLength) {
        toStringCache = null;
        super.setLength(newLength);
    }

    /**
     * @throws IndexOutOfBoundsException {@inheritDoc}
     * @see        #length()
     */
    @Override
    public synchronized char charAt(int index) {
        if ((index < 0) || (index >= count))
            throw new StringIndexOutOfBoundsException(index);
        return value[index];
    }

    /**
     * @since      1.5
     */
    @Override
    public synchronized int codePointAt(int index) {
        return super.codePointAt(index);
    }

    /**
     * @since     1.5
     */
    @Override
    public synchronized int codePointBefore(int index) {
        return super.codePointBefore(index);
    }

    /**
     * @since     1.5
     */
    @Override
    public synchronized int codePointCount(int beginIndex, int endIndex) {
        return super.codePointCount(beginIndex, endIndex);
    }

    /**
     * @since     1.5
     */
    @Override
    public synchronized int offsetByCodePoints(int index, int codePointOffset) {
        return super.offsetByCodePoints(index, codePointOffset);
    }

    /**
     * @throws IndexOutOfBoundsException {@inheritDoc}
     */
    @Override
    public synchronized void getChars(int srcBegin, int srcEnd, char[] dst,
                                      int dstBegin)
{
        super.getChars(srcBegin, srcEnd, dst, dstBegin);
    }

    /**
     * @throws IndexOutOfBoundsException {@inheritDoc}
     * @see        #length()
     */
    @Override
    public synchronized void setCharAt(int index, char ch) {
        if ((index < 0) || (index >= count))
            throw new StringIndexOutOfBoundsException(index);
        toStringCache = null;
        value[index] = ch;
    }

    @Override
    public synchronized StringBuffer append(Object obj) {
        toStringCache = null;
        super.append(String.valueOf(obj));
        return this;
    }

大部分的方法都是synchronized

Netty特点是一个线程对应多个channel，不需要启动多个线程连接的啊？为什么会变慢呢？

public class HappyChatMain {
    private static final Logger logger = LoggerFactory.getLogger(HappyChatMain.class);

    public static void main(String[] args) {
        final HappyChatServer server = new HappyChatServer(Constants.DEFAULT_PORT);
        server.init();
        System.out.println("服务器等待连接.............");
        System.out.println("当前服务器CPU核数"+Runtime.getRuntime().availableProcessors());
        server.start();
        // 注册进程钩子，在JVM进程关闭前释放资源
        Runtime.getRuntime().addShutdownHook(new Thread(){
            @Override
            public void run(){
                server.shutdown();
                logger.warn(">>>>>>>>>> jvm shutdown");
                System.exit(0);
            }
        });
    }
}

消息处理

@Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame)
            throws Exception {
        UserInfo userInfo = UserInfoManager.getUserInfo(ctx.channel());
        String content = frame.text();
        System.out.println("服务器在------" + LocalDate.now() + "收到消息" + content +
                "当前线程的id----" + Thread.currentThread().getId() + "线程名称" + Thread.currentThread().getName());
        Channel incoming = ctx.channel();
        clients.forEach(channel -> {
            if (channel != incoming) {
                channel.writeAndFlush(new TextWebSocketFrame("[" + incoming.remoteAddress() + "]" + frame.text()));
            } else {
                channel.writeAndFlush(new TextWebSocketFrame("[you]" + frame.text()));
            }
        });
        if (userInfo != null && userInfo.isAuth()) {
            JSONObject json = JSONObject.parseObject(frame.text());
            // 广播返回用户发送的消息文本
            UserInfoManager.broadcastMess(userInfo.getUserId(), userInfo.getNick(), json.getString("mess"));
        }
    }

启动：

服务器等待连接.............
当前服务器CPU核数4//本地电脑
userCount:

服务器在------2020-01-12收到消息{'code':10086,'mess':'2020年1月12日21:35:35  登录用户'}
当前线程的id----19线程名称DEFAULTEVENTLOOPGROUP_1

再启动一个客户端连接

服务器在------2020-01-12收到消息{'code':10086,'mess':'祖儿  登录2020年1月12日21:37:23'}
当前线程的id----22线程名称DEFAULTEVENTLOOPGROUP_2

UserInfoManager.java:124, broadCastPing userCount: 2个用户

但是当服务器接收多个客户端消息会出现延迟

服务器在------2020-01-12收到消息{'code':10086,'mess':'迪丽热巴 2020年1月12日21:39:47'}
当前线程的id----24线程名称DEFAULTEVENTLOOPGROUP_3

？？？？？？？？？

不是说好的Nio异步执行，不需要每次启动线程保持连接的吗？

现在为何每个客户端都进行了线程启动？

接受到的数据：[服务器在]2020-01-10T10:16:14.943接受到消息, 消息为：1002-------KK-------{'code':10086,'mess':'测试'}
2020-01-10 10:16:30.409  INFO 5514 --- [pool-1-thread-1] com.wolfbe.chat.handler.UserInfoManager  : broadCastPing userCount: 2
2020-01-10 10:16:30.443  INFO 5514 --- [VENTLOOPGROUP_1] com.wolfbe.chat.handler.UserAuthHandler  : receive pong message, address: 39.155.237.42:13007
2020-01-10 10:16:30.621  INFO 5514 --- [VENTLOOPGROUP_2] com.wolfbe.chat.handler.UserAuthHandler  : receive pong message, address: 223.104.3.131:25731
接受到的数据：[服务器在]2020-01-10T10:16:37.634接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'缺点不能去重'}
接受到的数据：[服务器在]2020-01-10T10:16:42.171接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'1'}
接受到的数据：[服务器在]2020-01-10T10:16:42.307接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'1'}
接受到的数据：[服务器在]2020-01-10T10:16:42.567接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'1'}
接受到的数据：[服务器在]2020-01-10T10:16:42.764接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'11'}
接受到的数据：[服务器在]2020-01-10T10:16:42.890接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'1'}
接受到的数据：[服务器在]2020-01-10T10:16:43.118接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'1'}
接受到的数据：[服务器在]2020-01-10T10:16:43.242接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'1'}
接受到的数据：[服务器在]2020-01-10T10:16:43.327接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'1'}
接受到的数据：[服务器在]2020-01-10T10:16:43.450接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'1'}
接受到的数据：[服务器在]2020-01-10T10:16:43.902接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'1'}
接受到的数据：[服务器在]2020-01-10T10:16:43.987接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'1'}
接受到的数据：[服务器在]2020-01-10T10:16:44.101接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'1'}
接受到的数据：[服务器在]2020-01-10T10:16:44.222接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'1'}
接受到的数据：[服务器在]2020-01-10T10:16:44.359接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'1'}
接受到的数据：[服务器在]2020-01-10T10:16:44.459接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'1'}
2020-01-10 10:16:50.389  INFO 5514 --- [pool-1-thread-1] com.wolfbe.chat.HappyChatServer          : scanNotActiveChannel --------
接受到的数据：[服务器在]2020-01-10T10:17:09.481接受到消息, 消息为：1002-------KK-------{'code':10086,'mess':'有没有延迟'}
接受到的数据：[服务器在]2020-01-10T10:17:18.279接受到消息, 消息为：1001-------kk-------{'code':10086,'mess':'有延迟'}
2020-01-10 10:17:20.409  INFO 5514 --- [pool-1-thread-1] com.wolfbe.chat.handler.UserInfoManager  : broadCastPing userCount: 2
2020-01-10 10:17:20.434  INFO 5514 --- [VENTLOOPGROUP_1] com.wolfbe.chat.handler.UserAuthHandler  : receive pong message, address: 39.155.237.42:13007
2020-01-10 10:17:20.484  INFO 5514 --- [VENTLOOPGROUP_2] com.wolfbe.chat.handler.UserAuthHandler  : receive pong message, address: 223.104.3.131:25731
2020-01-10 10:17:50.389  INFO 5514 --- [pool-1-thread-1] com.wolfbe.chat.HappyChatServer          : scanNotActiveChannel --------
2020-01-10 10:17:51.332  WARN 5514 --- [VENTLOOPGROUP_2] com.wolfbe.chat.handler.UserInfoManager  : channel will be remove, address is :223.104.3.131:25731
2020-01-10 10:17:51.333  WARN 5514 --- [VENTLOOPGROUP_2] com.wolfbe.chat.handler.UserInfoManager  : channel will be remove, address is :223.104.3.131:25731
2020-01-10 10:18:10.409  INFO 5514 --- [pool-1-thread-1] com.wolfbe.chat.handler.UserInfoManager  : broadCastPing userCount: 1
2020-01-10 10:18:10.437  INFO 5514 --- [VENTLOOPGROUP_1] com.wolfbe.chat.handler.UserAuthHandler  : receive pong message, address: 39.155.237.42:13007
2020-01-10 10:18:24.418  WARN 5514 --- [VENTLOOPGROUP_1] com.wolfbe.chat.handler.UserInfoManager  : channel will be remove, address is :39.155.237.42:13007

第一有延迟

而我可怜的阿里云私人服务器1 vCPU 2 GiB (I/O优化)！！！

1核啊，不管是cpu密集型，还是io密集型4个线程就够服务器响应的了，而且每个客户端为了保持长连接，还不会断开，所以当客户端多了，消息多了以后就会变慢，这严重违背了Nio的思想。

修改：

mainGroup = new NioEventLoopGroup();
subGroup = new NioEventLoopGroup();

启用Nio事件驱动组即可解决！！！

Connected to the target VM, address: '127.0.0.1:52594', transport: 'socket'
当前服务器CPU核数4
服务器等待连接.............

接收到的数据是:     hello  netty
TextWebSocketFrame(data: UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf(ridx: 0, widx: 78, cap: 156))
当前执行线程名字nioEventLoopGroup-5-1