java socket bio 改造为 netty nio

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

        厂家提供的demo代码如下：

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;public class Demo {public static void main(String[] args) {int port = 8003;if (args.length == 1) {port = Integer.parseInt(args[0]);}ServerSocket ss;try {ss = new ServerSocket(port);}catch (Exception e) {System.out.println("服务端socket失败 port = " + port);return;}System.out.println("启动socket监听 端口:" + port);List<Socket> socketList = new ArrayList<>();while (true) {try {Socket socket = ss.accept();if (socket == null || socket.isClosed()) {socketList.remove(socket);continue;}if (socketList.contains(socket)) {continue;}socketList.add(socket);System.out.println("socket连接 address = " + socket.getInetAddress().toString() + " port = " + socket.getPort());new Thread(new HealthReadThread(socket)).start();}catch (IOException e) {System.out.println(e.getMessage());}}}
}

import java.io.*;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;public class HealthReadThread implements Runnable {private Socket socket;HealthReadThread(Socket socket) {this.socket = socket;}private static String message = "";@Overridepublic void run() {try {//输入InputStream inPutStream = socket.getInputStream();BufferedInputStream bis = new BufferedInputStream(inPutStream);
//            BufferedReader br = new BufferedReader(new InputStreamReader(inPutStream));//输出OutputStream outputStream = socket.getOutputStream();BufferedOutputStream bw = new BufferedOutputStream(outputStream);String ip = socket.getInetAddress().getHostAddress();int port = socket.getPort();String readStr = "";
//            char[] buf;byte[] buf;int readLen = 0;while (true) {if (socket.isClosed()) {break;}buf = new byte[1024];try {readLen = bis.read(buf);if (readLen <= 0) {
//                        System.out.println(Thread.currentThread().getId() + "线程: " + "ip地址:" + ip + " 端口地址:" + port + "暂无接收数据");continue;}System.out.println(Thread.currentThread().getId() + "线程: " + "ip地址:" + ip + " 端口地址:" + port + " 接收到原始命令长度:" + readLen);readStr = StringUtils.byteToHexString(buf, readLen);
//                    readStr = new String(buf ,0 , readLen);} catch (IOException e) {System.out.println(e.getMessage());socket.close();
//                    continue;}if (readStr == null || "".equals(readStr)) {continue;}// 省略业务代码}}catch (Exception e) {System.out.println(e.getMessage());}}
}

使用netty进行改造：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class DeviceNettyServer implements ApplicationRunner {@Overridepublic void run(ApplicationArguments args) throws Exception {start();}public void start() {Thread thread = new Thread(() -> {// 配置服务端的NIO线程组EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup(4);ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup)// 使用 NIO 方式进行网络通信.channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {// 添加自己的处理器ch.pipeline().addLast(new DeviceMsgHandler());}});try {int port1 = 8081;int port2 = 8082;// 绑定一个端口并且同步，生成一个ChannelFuture对象ChannelFuture f1 = b.bind(port1).sync();ChannelFuture f2 = b.bind(port2).sync();log.info("启动监听, 端口:" + port1 + "、" + port2);// 对关闭通道进行监听f1.channel().closeFuture().sync();f2.channel().closeFuture().sync();} catch (Exception e) {log.error("启动监听失败", e);} finally {workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();}});thread.setName("DeviceNettyServer");thread.start();}
}

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;import java.util.*;
import java.util.concurrent.ConcurrentHashMap;@Slf4j
public class DeviceMsgHandler extends SimpleChannelInboundHandler<ByteBuf> {/*** 已连接的设备*/private static final ConcurrentHashMap<Channel, DeviceDTO> CONNECTION_DEVICE_MAP = new ConcurrentHashMap<>(8);/*** 一旦连接，第一个被执行*/@Overridepublic void handlerAdded(ChannelHandlerContext ctx) {String remoteAddress = ctx.channel().remoteAddress().toString();log.info("发现连接, remoteAddress: " + remoteAddress);// 发送查询设备信息指令sendQuery(ctx.channel());}/*** 读取数据*/@Overrideprotected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {byte[] bytes = new byte[msg.readableBytes()];msg.readBytes(bytes);// 忽略业务处理代码// 传递给下一个处理器ctx.fireChannelRead(msg);}/*** 连接断开** @param ctx*/@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) {log.info("连接断开, remoteAddress: " + ctx.channel().remoteAddress());CONNECTION_DEVICE_MAP.remove(ctx.channel());}/*** 连接异常** @param ctx* @param cause*/@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {log.info("连接异常, remoteAddress: " + ctx.channel().remoteAddress());CONNECTION_DEVICE_MAP.remove(ctx.channel());}

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