搭建百万连接服务，使用netty完成websocke的推送

野性酷女 2022-10-16 07:33 522阅读 0赞

**目录**

（一）使用websocket

① 介绍

② websocket实现

③ websocket示例

（二）说说netty如何实现百万连接

① 介绍

② 解决方案

③ client端 和 server端需要修改的代码

（三）linux下的百万配置

① 环境配置

② centos服务器配置

③ centos服务器配置

④ 启动

（四）百万连接配置说明

--------------------

## （一）使用websocket ##

*  ### ① 介绍 ###

> webSocket协议是基于TCP的一种新的网络协议。他的出现实现了网络和浏览器全双工通信，允许服务器主动发送信息给客户端。客户端 给 服务器发消息是半双工，服务器给客户端也发送消息就是全双工。
> 
> 多客户端多语言多浏览器支持：浏览器，php，Java，ruby，nginx，python，Tomcat，erlang，.net等等。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE3MDEwMTkz_size_16_color_FFFFFF_t_70][]

*  ### ② websocket实现 ###

> 服务端和客户端交流，通过的是websocket这种协议，内部传输的协议，通过websocket这种方式和普通的socket没有什么区别，唯一一点就是协议不同。

*  ### ③ websocket示例 ###

> 绑定9001，这是浏览器js方式完成的，可以发送消息和接收消息，生成一个随机数userId，在设计这个功能的时候，考虑到消息推送是否需要先登录。登录过跳转到登录成功，成功之后拿到userId，在建立websocket连接的时候就可以携带userId，最终知道是那个用户。

<html>
     <head>
      <meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
      <title>Web Socket Test</title>
     </head> 
     <body> 
      <script type="text/javascript">
    var socket;
    if (!window.WebSocket) {
      window.WebSocket = window.MozWebSocket;
    }
    if (window.WebSocket) {
      // 随机数
      var random = Math.floor(Math.random()*(10000 - 10 +1) + 10)
      socket = new WebSocket("ws://127.0.0.1:9001/websocket?userId=" + random);
      socket.onmessage = function(event) {
        var ta = document.getElementById('responseText');
        ta.value = ta.value + '\n' + event.data
      };
      socket.onopen = function(event) {
        var ta = document.getElementById('responseText');
        ta.value = "Web Socket opened!";
      };
      socket.onclose = function(event) {
        var ta = document.getElementById('responseText');
        ta.value = ta.value + "Web Socket closed"; 
      };
    } else {
      alert("Your browser does not support Web Socket.");
    }
    
    function send(message) {
      if (!window.WebSocket) { return; }
      if (socket.readyState == WebSocket.OPEN) {
        socket.send(message);
      } else {
        alert("The socket is not open.");
      }
    }
    </script> 
      <form onsubmit="return false;"> 
       <input type="text" name="message" value="Hello, World!" />
       <input type="button" value="Send Web Socket Data" onclick="send(this.form.message.value)" /> 
       <h3>Output</h3> 
       <textarea id="responseText" style="width:500px;height:300px;"></textarea> 
      </form> 
     </body>
    </html>

> java代码示例server端

import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    public final class WebSocketServer {
    
        static int PORT = 9000;
    
        public static void main(String[] args) throws Exception {
    
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_REUSEADDR, true)
                        .childHandler(new WebSocketServerInitializer())
                        .childOption(ChannelOption.SO_REUSEADDR, true);
                    b.bind(++PORT).addListener(new ChannelFutureListener() {
                        public void operationComplete(ChannelFuture future) throws Exception {
                            if ("true".equals(System.getProperty("netease.debug")))
                                System.out.println("端口绑定完成：" + future.channel().localAddress());
                        }
                    });
                
    
                // 端口绑定完成，启动消息随机推送(测试)
                TestCenter.startTest();
                System.in.read();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

> server端的 channel类。绑定完成相关的业务逻辑处理， 1.解析http协议编解码 2.最大的解析数据包拦截 3.自己的业务处理，得到msg对象，变成了一个request的对象。

import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.handler.codec.http.HttpObjectAggregator;
    import io.netty.handler.codec.http.HttpServerCodec;
    
    /**
     */
    public class WebSocketServerInitializer extends ChannelInitializer<SocketChannel> {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            //  职责链， 数据处理流程
            ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast(new HttpServerCodec()); //
            pipeline.addLast(new HttpObjectAggregator(65536));
            pipeline.addLast(new WebSocketServerHandler());
            pipeline.addLast(new NewConnectHandler());
        }
    }

> TestCenter 类

import io.netty.channel.Channel;
    import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
    
    import java.util.Random;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    // 正常情况是，后台系统通过接口请求，把数据丢到对应的MQ队列，再由推送服务器读取
    public class TestCenter {
        // 此处假设一个用户一台设备，否则用户的通道应该是多个。
        // TODO 还应该有一个定时任务，用于检测失效的连接(类似缓存中的LRU算法，长时间不使用，就拿出来检测一下是否断开了)；
        static ConcurrentHashMap<String, Channel> userInfos = new ConcurrentHashMap<String, Channel>();
    
        // 保存信息
        public static void saveConnection(String userId, Channel channel) {
            userInfos.put(userId, channel);
        }
    
        // 退出的时候移除掉
        public static void removeConnection(Object userId) {
            if (userId != null) {
                userInfos.remove(userId.toString());
            }
        }

> server handler 服务

import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.handler.codec.http.FullHttpRequest;
    import io.netty.handler.codec.http.QueryStringDecoder;
    import io.netty.util.AttributeKey;
    
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.atomic.AtomicInteger;
    
    // 新连接建立了
    public class NewConnectHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
            // 解析请求，判断token，拿到用户ID。
            Map<String, List<String>> parameters = new QueryStringDecoder(req.uri()).parameters();
            // String token = parameters.get("token").get(0);  不是所有人都能连接，比如需要登录之后，发放一个推送的token
            String userId = parameters.get("userId").get(0);
            ctx.channel().attr(AttributeKey.valueOf("userId")).getAndSet(userId); // channel中保存userId
            TestCenter.saveConnection(userId, ctx.channel()); // 保存连接
    
            // 结束
        }
    }

> 解析 handler，http解码失败，如果不是websocket协议升级的请求，现在开发的是一个websocket服务器，对于一切不符合我要求的请求，可以不管理它，这跟开发springmvc，要接触的东西，不是url，http协议底层，请求的头部协议，如果不符合就返回400，如果是就返回当前处理我可以接收。正常开始通信了，握手，返回响应的状态码。先处理websocket的握手，后处理websocket的消息。握手就是你同意给我交往了，

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelFutureListener;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.handler.codec.http.*;
    import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
    import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
    import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
    import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
    import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
    import io.netty.handler.codec.http.websocketx.WebSocketFrame;
    import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
    import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
    import io.netty.util.AttributeKey;
    import io.netty.util.CharsetUtil;
    
    import java.util.concurrent.atomic.AtomicInteger;
    import java.util.concurrent.atomic.LongAdder;
    
    import static io.netty.handler.codec.http.HttpMethod.*;
    import static io.netty.handler.codec.http.HttpResponseStatus.*;
    import static io.netty.handler.codec.http.HttpVersion.*;
    
    /**
     * Handles handshakes and messages
     */
    public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
    
        private static final String WEBSOCKET_PATH = "/websocket";
    
        private WebSocketServerHandshaker handshaker;
    
        public static final LongAdder counter = new LongAdder();
    
        @Override
        public void channelRead0(ChannelHandlerContext ctx, Object msg) {
            counter.add(1);
            if (msg instanceof FullHttpRequest) {
                // 处理websocket握手
                handleHttpRequest(ctx, (FullHttpRequest) msg);
            } else if (msg instanceof WebSocketFrame) {
                // 处理websocket后续的消息
                handleWebSocketFrame(ctx, (WebSocketFrame) msg);
            }
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) {
            ctx.flush();
        }
    
        private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
            // Handle a bad request. //如果http解码失败 则返回http异常 并且判断消息头有没有包含Upgrade字段(协议升级)
            if (!req.decoderResult().isSuccess() || req.method() != GET || (!"websocket".equals(req.headers().get("Upgrade")))) {
                sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
                return;
            }
    
            // 构造握手响应返回
            WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
                    getWebSocketLocation(req), null, true, 5 * 1024 * 1024);
            handshaker = wsFactory.newHandshaker(req);
            if (handshaker == null) {
                // 版本不支持
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
            } else {
                handshaker.handshake(ctx.channel(), req);
                ctx.fireChannelRead(req.retain()); // 继续传播
            }
        }
    
        private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
            // Check for closing frame 关闭
            if (frame instanceof CloseWebSocketFrame) {
                Object userId = ctx.channel().attr(AttributeKey.valueOf("userId")).get();
                TestCenter.removeConnection(userId);
                handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
                return;
            }
            if (frame instanceof PingWebSocketFrame) { // ping/pong作为心跳
                System.out.println("ping: " + frame);
                ctx.write(new PongWebSocketFrame(frame.content().retain()));
                return;
            }
            if (frame instanceof TextWebSocketFrame) {
                // Echo the frame
                // TODO 处理具体的数据请求（... 聊天室，推送给其他的用户）
                //发送到客户端websocket
                ctx.channel().write(new TextWebSocketFrame(((TextWebSocketFrame) frame).text()
                        + ", 欢迎使用Netty WebSocket服务， 现在时刻:"
                        + new java.util.Date().toString()));
    
                return;
            }
            // 不处理二进制消息
            if (frame instanceof BinaryWebSocketFrame) {
                // Echo the frame
                ctx.write(frame.retain());
            }
        }
    
        private static void sendHttpResponse(
                ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
            // Generate an error page if response getStatus code is not OK (200).
            if (res.status().code() != 200) {
                ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
                res.content().writeBytes(buf);
                buf.release();
                HttpUtil.setContentLength(res, res.content().readableBytes());
            }
    
            // Send the response and close the connection if necessary.
            ChannelFuture f = ctx.channel().writeAndFlush(res);
            if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) {
                f.addListener(ChannelFutureListener.CLOSE);
            }
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            ctx.close();
        }
    
        private static String getWebSocketLocation(FullHttpRequest req) {
            String location = req.headers().get(HttpHeaderNames.HOST) + WEBSOCKET_PATH;
            return "ws://" + location;
        }
    }

> 上边是通过html的客户端，这里用java的netty写一个websocket的客户端

import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.http.HttpClientCodec;
    import io.netty.handler.codec.http.HttpObjectAggregator;
    import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketClientCompressionHandler;
    
    import java.util.concurrent.atomic.AtomicInteger;
    
    public final class WebSocketClient {
    
    
        public static void main(String[] args) throws Exception {
            final String host = System.getProperty("netease.pushserver.host", "127.0.0.1");
            final String maxSize = System.getProperty("netease.client.port.maxSize", "100");
            final String maxConnections = System.getProperty("netease.client.port.maxConnections", "60000");
            int port = 9001;
    
            EventLoopGroup group = new NioEventLoopGroup();
            try {
    
                Bootstrap b = new Bootstrap();
                b.group(group).channel(NioSocketChannel.class);
                b.option(ChannelOption.SO_REUSEADDR, true);
                b.handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new HttpClientCodec());
                        p.addLast(new HttpObjectAggregator(8192));
                        p.addLast(WebSocketClientCompressionHandler.INSTANCE);
                        p.addLast("webSocketClientHandler", new WebSocketClientHandler());
                    }
                });
                // tcp 建立连接
    
                        b.connect(host, port).sync().get();
                    
                System.in.read();
            } finally
    
            {
                group.shutdownGracefully();
            }
        }
    }

> 对应客户端的handler，先http获取消息，然后服务端给客户端发送消息，客户端收到了，如果是http消息，可能不能解析，如果是websocket消息，进行消息的打印。客户端收到了消息，把消息的内容给打印出来。

import io.netty.channel.Channel;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelPromise;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.handler.codec.http.DefaultHttpHeaders;
    import io.netty.handler.codec.http.FullHttpResponse;
    import io.netty.handler.codec.http.websocketx.*;
    import io.netty.util.CharsetUtil;
    
    import java.net.InetSocketAddress;
    import java.net.URI;
    import java.net.URISyntaxException;
    import java.util.Random;
    import java.util.concurrent.atomic.AtomicInteger;
    
    // handler 处理多个~ tcp连接建立之后的事件
    // open websocket
    public class WebSocketClientHandler extends SimpleChannelInboundHandler<Object> {
    
        private WebSocketClientHandshaker handshaker;
        private ChannelPromise handshakeFuture;
    
        public ChannelFuture handshakeFuture() {
            return handshakeFuture;
        }
    
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) {
            handshakeFuture = ctx.newPromise();
        }
    
        static AtomicInteger counter = new AtomicInteger(0);
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            if (handshaker == null) {
                InetSocketAddress address = (InetSocketAddress) ctx.channel().remoteAddress();
                URI uri = null;
                try {
                    uri = new URI("ws://" + address.getHostString() + ":" + address.getPort() + "/websocket?userId=" + counter.incrementAndGet());
                } catch (Exception e) {
                    e.printStackTrace();
                }
                handshaker = WebSocketClientHandshakerFactory.newHandshaker(
                        uri, WebSocketVersion.V13, null, true, new DefaultHttpHeaders());
            }
            handshaker.handshake(ctx.channel());
        }
    
        @Override
        public void channelInactive(ChannelHandlerContext ctx) {
            if ("true".equals(System.getProperty("netease.debug"))) System.out.println("WebSocket Client disconnected!");
        }
    
        @Override
        public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
            Channel ch = ctx.channel();
            if (!handshaker.isHandshakeComplete()) {
                try {
                    handshaker.finishHandshake(ch, (FullHttpResponse) msg);
                    if ("true".equals(System.getProperty("netease.debug")))
                        System.out.println("WebSocket Client connected!");
                    handshakeFuture.setSuccess();
                } catch (WebSocketHandshakeException e) {
                    if ("true".equals(System.getProperty("netease.debug")))
                        System.out.println("WebSocket Client failed to connect");
                    handshakeFuture.setFailure(e);
                }
                return;
            }
    
            if (msg instanceof FullHttpResponse) {
                FullHttpResponse response = (FullHttpResponse) msg;
                throw new IllegalStateException(
                        "Unexpected FullHttpResponse (getStatus=" + response.status() +
                                ", content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
            }
    
            WebSocketFrame frame = (WebSocketFrame) msg;
            if (frame instanceof TextWebSocketFrame) {
                TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
                if ("true".equals(System.getProperty("netease.debug")))
                    System.out.println("WebSocket Client received message: " + textFrame.text());
            } else if (frame instanceof PongWebSocketFrame) {
                if ("true".equals(System.getProperty("netease.debug")))
                    System.out.println("WebSocket Client received pong");
            } else if (frame instanceof CloseWebSocketFrame) {
                if ("true".equals(System.getProperty("netease.debug")))
                    System.out.println("WebSocket Client received closing");
                ch.close();
            }
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            if (!handshakeFuture.isDone()) {
                handshakeFuture.setFailure(cause);
            }
            ctx.close();
        }
    }

> 源码的pom文件

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.idig8.netty</groupId>
        <artifactId>netty-push</artifactId>
        <version>1.0.0</version>
        <build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <configuration>
                        <source>8</source>
                        <target>8</target>
                    </configuration>
                </plugin>
            </plugins>
        </build>
    
        <dependencies>
            <dependency>
                <groupId>io.netty</groupId>
                <artifactId>netty-all</artifactId>
                <version>4.1.33.Final</version>
            </dependency>
        </dependencies>
    </project>

## （二）说说netty如何实现百万连接 ##

*  ### ① 介绍 ###

> 100w连接需要多少台机器才能构建起来，这肯定是很多的，在测试的过程中，不需要几百个服务器完成百万连接。需要注意服务器支持端口的数量是可以支持很多的，但是如何2台服务器要实现百万连接，需要考虑一个TCP层次的一种限制，两台服务器之间建立的连接数量是有限的。网络四元组（客户端IP，客户端端口，服务端IP，服务端端口）。

1.  同一个IP的端口数不超过65535个，这是个限制，每一个连接不仅仅在服务器上开启一个端口，在客户端也会开启一个端口，每一个TCP连接涉及到端口数量的限制，客户端只有6万多个端口。（不可能搞那么多机器，所以让一个客户端发起100万的连接请求，如果是生产环境就不用考虑这个问题）
2.  服务器只有一个端口的情况下，同一个客户端只能对他发起6万多个连接。客户端每发起一个请求，就需要开启一个端口。客户端没有端口就说明它没办法发起请求。

*  ### ② 解决方案 ###

1.  服务器开启多个端口，网络上区别机器是通过网络四元组来标记的。客户端的端口虽然有限，但是可以复用里面的端口。
2.  举个例子：客户端端口20000，已经连接了服务端9001 这个端口。20000端口在去连接服务端另一个端口9002这个端口。对于客户端指定端口复用，操作系统会自动处理的。
3.  netty里面开启了 地址的复用。客户端也开启复用。参数都是TCP参数。ChannelOption.SO\_REUSEADDR
4.  client 和server端通过循环的方式，增加多个端口的绑定。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE3MDEwMTkz_size_16_color_FFFFFF_t_70 1][]

*  ### ③ client端 和 server端需要修改的代码 ###

> WebSocketClient

// tcp 建立连接
                for (int i = 0; i < 100; i++) {
                    for (int j = 0; j < 60000; j++) {
                        b.connect(host, port).sync().get();
                    }
                    port++;
                }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE3MDEwMTkz_size_16_color_FFFFFF_t_70 2][]

> WebSocketServer

for (int i = 0; i < 100; i++) {
                    b.bind(++PORT).addListener(new ChannelFutureListener() {
                        public void operationComplete(ChannelFuture future) throws Exception {
                            if ("true".equals(System.getProperty("netease.debug")))
                                System.out.println("端口绑定完成：" + future.channel().localAddress());
                        }
                    });
                }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE3MDEwMTkz_size_16_color_FFFFFF_t_70 3][]

## （三）linux下的百万配置 ##

*  ### ① 环境配置 ###

> 测试环境：两台 centos7 jdk8 4核8G

*  ### ② centos服务器配置 ###

> 进程最大文件打开添加参数最大限制

vi /etc/security/limits.conf  
    
    # 修改下面的内容
    
    * soft nofile 1000000
    * hard nofile 1000000

> 全局限制 cat /proc/sys/fs/file-nr

echo 1200000 > /proc/sys/fs/file-max

> 服务设置

vi /etc/sysctl.conf
    #修改下面的内容
    fs.file-max = 1000000

> 问题汇总

# 客户机开不了这么多连接 ，可能的问题原因端口开放数
    linux对外随机分配的端口是有限制,理论上单机对外端口数可达65535,但实际对外可建立的连接默认最大只有28232个
    查看：cat /proc/sys/net/ipv4/ip_local_port_range
    echo "net.ipv4.ip_local_port_range= 1024 65535">> /etc/sysctl.conf
    sysctl -p
    
    # 如果你的机器差，出现了奇怪的问题~
    sysctl -w net.ipv4.tcp_tw_recycle=1 #快速回收time_wait的连接
    sysctl -w net.ipv4.tcp_tw_reuse=1 
    sysctl -w net.ipv4.tcp_timestamps=1

# 如果发现自己的用例跑不上去，就看看linux日志
    tail -f /var/log/messages
    # linux 日志
    1、 nf_conntrack: table full, dropping packet 表示防火墙的表满了，加大 nf_conntrack_max 参数
    echo "net.nf_conntrack_max = 1000000">> /etc/sysctl.conf
    
    # 2、 TCP: too many orphaned sockets 表示内存不太够，拒绝分配，一般就是TCP缓冲区内存不够用，调大一点
    # cat /proc/sys/net/ipv4/tcp_mem 
    echo "net.ipv4.tcp_mem = 786432 2097152 16777216">> /etc/sysctl.conf
    echo "net.ipv4.tcp_rmem = 4096 4096 16777216">> /etc/sysctl.conf
    echo "net.ipv4.tcp_wmem = 4096 4096 16777216">> /etc/sysctl.conf
    sysctl -p

*  ### ③ centos服务器配置 ###

# 查看某个端口的连接情况
    netstat -nat|grep -i "9001"|wc -l
    netstat -n | awk '/^tcp/ {++S[$NF]} END {for(a in S) print a, S[a]}'
    
    # 网络接口的带宽使用情况 
    #tcpdump https://www.cnblogs.com/maifengqiang/p/3863168.html
    
    # glances工具
    yum install -y glances
    glances 控制台查看
    glances -s 服务器模式查看

*  ### ④ 启动 ###

# 服务端启动
    java -Xmx4096m -Xms4096m -Dnetease.debug=true -cp netty-all-4.1.33.Final.jar:netty-push-1.0.0.jar com.idig8.netty.push.server.WebSocketServer
    
    # 客户端
    java -Xmx4096m -Xms4096m -Dnetease.debug=false -Dnetease.pushserver.host=192.168.100.101 -cp netty-all-4.1.33.Final.jar:netty-push-1.0.0.jar com.idig8.netty.push.client.WebSocketClient
    
    
    # 发送消息服务端启动
    java -Xmx4096m -Xms4096m -Dnetease.debug=true -Dneteae.server.test.sendmsg=true -cp netty-all-4.1.33.Final.jar:netty-push-1.0.0.jar com.idig8.netty.push.server.WebSocketServer

## （四）百万连接配置说明 ##

> 一台机器为什么能支持百万的连接感觉有点科幻，感觉不可能，首先需要理解NIO的概念，NIO如果没有做任何处理的情况下，网络是不需要java程序处理的，java程序在连接没有产生动静的时候，java是不产生事件的时候，只有真正产生事件的时候，selector会通知，在一个海量连接的过程中，只要没有消息的推送，消息传到服务器来，不管有几百万个请求，都可以去接受，这是NIO的特性，不像BIO一个请求要开启一个连接，并非是无限的增长，对于连接这么多，内存资源首先被消耗，占用最大的是内存，操作系统底层有TCP的关联，java的netty里面也有channel，需要保留连接，一个连接产生一个对应的对象，虽然这个对象没有处理但是会占用内存，跟cpu没有太大管理，只是java程序要处理占用了cpu和内存。
> 
> 有很多连接有一个误解，百万连接需要很多机器，百万连接的关键是NIO网络的机制，对NIO没有直观认识的时候，不知道NIO能带来什么。在真实的生产环境的情况下，服务端不需要这么多端口，开100个端口，为了让测试服务器可以连接，上边有个命令是发送消息服务器启动，一旦涉及到百万连接和发送消息的话肯定设计到大量的资源消耗，netty调用handler，channel，这些都是消耗资源的，服务器cpu资源会挂掉，4核 linux top的方式来查看cpu最大是400%。

PS：最好是通过代码，自己试一下，了解下百万连接的思路，按照正常是分布式的架构，单机始终是有瓶颈的，100万用户的连接的话单机8g4核轻轻松松，分布式系统就要设计到分布式，注册中心的概念，推送使用netty方便系统的开发，沾包和拆包的问题方法去解决，而不是自己写一个socket程序很复杂，netty是通过责任链的方式，通过pipline控制之后的步骤。netty的底层是基于NIO，NIO的底层是基于多路复用的机制，多路复用机制是依托于操作系统的，百万连接这个是拼操作系统参数的，java代码是使用的NIO，如果不是使用的NIO，不好意思你达不到，设置到一些系统操作的配置。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE3MDEwMTkz_size_16_color_FFFFFF_t_70]: /images/20221014/204d52ca19224f47b058ce88df7823ec.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE3MDEwMTkz_size_16_color_FFFFFF_t_70 1]: /images/20221014/1a71cadc37f248feb157a4e328dbfd3f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE3MDEwMTkz_size_16_color_FFFFFF_t_70 2]: /images/20221014/782bb14de0ff4cd2a9fb3da1f3aea72e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE3MDEwMTkz_size_16_color_FFFFFF_t_70 3]: /images/20221014/98e0532aa2004d27b72fc428bc31f876.png