java socket bio 改造为 netty nio

        公司早些时候接入一款康健监测设备，由于业务原因克日把端口袒露在公网后，每当被恶意连接时系统会创建大量线程，在排盘问题是发现是利用了厂家提供的服务端demo代码，在代码中利用的是java 原生socket，在发现连接后利用独立线程处理后续通讯，占用系统资源造成了服务宕机，因此必要进行改造。
        厂家提供的demo代码如下：

1. import java.io.IOException;
2. import java.net.ServerSocket;
3. import java.net.Socket;
4. import java.util.ArrayList;
5. import java.util.List;
7. public class Demo {
9.     public static void main(String[] args) {
10.         int port = 8003;
11.         if (args.length == 1) {
12.             port = Integer.parseInt(args[0]);
13.         }
14.         ServerSocket ss;
15.         try {
16.             ss = new ServerSocket(port);
17.         }
18.         catch (Exception e) {
19.             System.out.println("服务端socket失败 port = " + port);
20.             return;
21.         }
22.         System.out.println("启动socket监听 端口:" + port);
23.         List<Socket> socketList = new ArrayList<>();
24.         while (true) {
25.             try {
26.                 Socket socket = ss.accept();
27.                 if (socket == null || socket.isClosed()) {
28.                     socketList.remove(socket);
29.                     continue;
30.                 }
31.                 if (socketList.contains(socket)) {
32.                     continue;
33.                 }
34.                 socketList.add(socket);
35.                 System.out.println("socket连接 address = " + socket.getInetAddress().toString() + " port = " + socket.getPort());
36.                 new Thread(new HealthReadThread(socket)).start();
37.             }
38.             catch (IOException e) {
39.                 System.out.println(e.getMessage());
40.             }
41.         }
42.     }
43. }

复制代码

1. import java.io.*;
2. import java.net.Socket;
3. import java.util.ArrayList;
4. import java.util.HashMap;
5. import java.util.List;
7. public class HealthReadThread implements Runnable {
9.     private Socket socket;
11.     HealthReadThread(Socket socket) {
12.         this.socket = socket;
13.     }
15.     private static String message = "";
17.     @Override
18.     public void run() {
19.         try {
20.             //输入
21.             InputStream inPutStream = socket.getInputStream();
22.             BufferedInputStream bis = new BufferedInputStream(inPutStream);
23. //            BufferedReader br = new BufferedReader(new InputStreamReader(inPutStream));
24.             //输出
25.             OutputStream outputStream = socket.getOutputStream();
26.             BufferedOutputStream bw = new BufferedOutputStream(outputStream);
28.             String ip = socket.getInetAddress().getHostAddress();
29.             int port = socket.getPort();
31.             String readStr = "";
32. //            char[] buf;
33.             byte[] buf;
34.             int readLen = 0;
35.             while (true) {
36.                 if (socket.isClosed()) {
37.                     break;
38.                 }
39.                 buf = new byte[1024];
40.                 try {
41.                     readLen = bis.read(buf);
42.                     if (readLen <= 0) {
43. //                        System.out.println(Thread.currentThread().getId() + "线程: " + "ip地址:" + ip + " 端口地址:" + port + "暂无接收数据");
44.                         continue;
45.                     }
46.                     System.out.println(Thread.currentThread().getId() + "线程: " + "ip地址:" + ip + " 端口地址:" + port + " 接收到原始命令长度:" + readLen);
47.                     readStr = StringUtils.byteToHexString(buf, readLen);
48. //                    readStr = new String(buf ,0 , readLen);
49.                 } catch (IOException e) {
50.                     System.out.println(e.getMessage());
51.                     socket.close();
52. //                    continue;
53.                 }
54.                 if (readStr == null || "".equals(readStr)) {
55.                     continue;
56.                 }
57.                 // 省略业务代码
58.             }
59.         }
60.         catch (Exception e) {
61.             System.out.println(e.getMessage());
62.         }
63.     }
64. }

复制代码

利用netty进行改造：

1. import io.netty.bootstrap.ServerBootstrap;
2. import io.netty.channel.ChannelFuture;
3. import io.netty.channel.ChannelInitializer;
4. import io.netty.channel.EventLoopGroup;
5. import io.netty.channel.nio.NioEventLoopGroup;
6. import io.netty.channel.socket.SocketChannel;
7. import io.netty.channel.socket.nio.NioServerSocketChannel;
8. import lombok.extern.slf4j.Slf4j;
9. import org.springframework.boot.ApplicationArguments;
10. import org.springframework.boot.ApplicationRunner;
11. import org.springframework.stereotype.Component;
13. @Slf4j
14. @Component
15. public class DeviceNettyServer implements ApplicationRunner {
17.     @Override
18.     public void run(ApplicationArguments args) throws Exception {
19.         start();
20.     }
22.     public void start() {
23.         Thread thread = new Thread(() -> {
24.             // 配置服务端的NIO线程组
25.             EventLoopGroup bossGroup = new NioEventLoopGroup(1);
26.             EventLoopGroup workerGroup = new NioEventLoopGroup(4);
28.             ServerBootstrap b = new ServerBootstrap();
29.             b.group(bossGroup, workerGroup)
30.                     // 使用 NIO 方式进行网络通信
31.                     .channel(NioServerSocketChannel.class)
32.                     .childHandler(new ChannelInitializer<SocketChannel>() {
33.                         @Override
34.                         public void initChannel(SocketChannel ch) throws Exception {
35.                             // 添加自己的处理器
36.                             ch.pipeline().addLast(new DeviceMsgHandler());
37.                         }
38.                     });
40.             try {
41.                 int port1 = 8081;
42.                 int port2 = 8082;
43.                 // 绑定一个端口并且同步，生成一个ChannelFuture对象
44.                 ChannelFuture f1 = b.bind(port1).sync();
45.                 ChannelFuture f2 = b.bind(port2).sync();
46.                 log.info("启动监听, 端口:" + port1 + "、" + port2);
48.                 // 对关闭通道进行监听
49.                 f1.channel().closeFuture().sync();
50.                 f2.channel().closeFuture().sync();
51.             } catch (Exception e) {
52.                 log.error("启动监听失败", e);
53.             } finally {
54.                 workerGroup.shutdownGracefully();
55.                 bossGroup.shutdownGracefully();
56.             }
57.         });
58.         thread.setName("DeviceNettyServer");
59.         thread.start();
60.     }
61. }

复制代码

1. import com.alibaba.fastjson.JSON;
2. import com.alibaba.fastjson.JSONObject;
3. import io.netty.buffer.ByteBuf;
4. import io.netty.buffer.Unpooled;
5. import io.netty.channel.Channel;
6. import io.netty.channel.ChannelHandlerContext;
7. import io.netty.channel.SimpleChannelInboundHandler;
8. import lombok.extern.slf4j.Slf4j;
9. import org.apache.commons.collections.CollectionUtils;
10. import org.apache.commons.lang3.StringUtils;
12. import java.util.*;
13. import java.util.concurrent.ConcurrentHashMap;
15. @Slf4j
16. public class DeviceMsgHandler extends SimpleChannelInboundHandler<ByteBuf> {
18.     /**
19.      * 已连接的设备
20.      */
21.     private static final ConcurrentHashMap<Channel, DeviceDTO> CONNECTION_DEVICE_MAP = new ConcurrentHashMap<>(8);
23.     /**
24.      * 一旦连接，第一个被执行
25.      */
26.     @Override
27.     public void handlerAdded(ChannelHandlerContext ctx) {
28.         String remoteAddress = ctx.channel().remoteAddress().toString();
29.         log.info("发现连接, remoteAddress: " + remoteAddress);
30.         // 发送查询设备信息指令
31.         sendQuery(ctx.channel());
32.     }
34.     /**
35.      * 读取数据
36.      */
37.     @Override
38.     protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
39.         byte[] bytes = new byte[msg.readableBytes()];
40.         msg.readBytes(bytes);
41.         // 忽略业务处理代码
43.         // 传递给下一个处理器
44.         ctx.fireChannelRead(msg);
45.     }
47.     /**
48.      * 连接断开
49.      *
50.      * @param ctx
51.      */
52.     @Override
53.     public void handlerRemoved(ChannelHandlerContext ctx) {
54.         log.info("连接断开, remoteAddress: " + ctx.channel().remoteAddress());
55.         CONNECTION_DEVICE_MAP.remove(ctx.channel());
56.     }
58.     /**
59.      * 连接异常
60.      *
61.      * @param ctx
62.      * @param cause
63.      */
64.     @Override
65.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
66.         log.info("连接异常, remoteAddress: " + ctx.channel().remoteAddress());
67.         CONNECTION_DEVICE_MAP.remove(ctx.channel());
68.       
69.     }

复制代码

经过改造后利用了4个worker线程进行读写，消除了原先恶意连接造成线程数无线扩大的问题，利用nio也极大的提高了系统资源利用率。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。