当前位置：首页 > news >正文

Redis IO多路复用

news 2025/7/21 10:43:41

0、前言

本文所有代码可见 => 【gitee code demo】
本文涉及的主题：

1、BIO、NIO的业务实践和缺陷

2、Redis IO多路复用：redis快的主要原因

3、epoll 架构

部分图片 via 【epoll 原理分析】

1、BIO单线程版

1.1 业务代码

client client代码相同启动多个即可

public class RedisClient1 {public static void main(String[] args) throws IOException {SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");Socket socket = new Socket("127.0.0.1", 6300);log("{} {}> 尝试连接服务 {}", sdf.format(new Date()) ,socket.getLocalPort(), socket.getPort());OutputStream outputStream = socket.getOutputStream();while (true) {Scanner scanner = new Scanner(System.in);log("{} {}> ", sdf.format(new Date()) ,socket.getLocalPort());String string = scanner.nextLine();if (string.equalsIgnoreCase("quit")) {break;}socket.getOutputStream().write(string.getBytes());log("{} {}> 发送数据：{}", sdf.format(new Date()) ,socket.getLocalPort(), string);}outputStream.close();socket.close();}}

server

public class RedisServerBIO {public static void main(String[] args) throws IOException {SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");ServerSocket serverSocket = new ServerSocket(6300);while (true) {log("{} {}> ", sdf.format(new Date()), serverSocket.getLocalPort());Socket socket = serverSocket.accept();//阻塞1 ,等待客户端连接log("{} {}> {} 连接到服务", sdf.format(new Date()), socket.getLocalPort(), socket.getPort());InputStream inputStream = socket.getInputStream();int length = -1;byte[] bytes = new byte[1024];log("{} {}> ", sdf.format(new Date()), serverSocket.getLocalPort());while ((length = inputStream.read(bytes)) != -1)//阻塞2 ,等待客户端发送数据{log("{} {}> 收到 {} 的消息：{}", sdf.format(new Date()), serverSocket.getLocalPort(), socket.getPort(), new String(bytes, 0, length));}inputStream.close();socket.close();}}
}

1.2 结果演示

现象：

1、client1 连接到server，client2尝试连接被阻塞

2、client2 先发送的消息未被server接受，client1后发送的repeat消息被server接受

结论：

BIO会一直阻塞，单线程下只能处理一个socket连接

存在的问题：

多 client 访问时效率低

在这里插入图片描述

2、BIO多线程版

2.1 业务代码

public class RedisServerBIOMultiThread {public static void main(String[] args) throws IOException {ServerSocket serverSocket = new ServerSocket(6300);SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");log("{} {}> ", sdf.format(new Date()), serverSocket.getLocalPort());while (true) {Socket socket = serverSocket.accept();//阻塞1 ,等待客户端连接log("{} {}> {} 连接到服务", sdf.format(new Date()), socket.getLocalPort(), socket.getPort());new Thread(() -> {try {InputStream inputStream = socket.getInputStream();int length = -1;byte[] bytes = new byte[1024];while ((length = inputStream.read(bytes)) != -1)//阻塞2 ,等待客户端发送数据{log("{} {}> 收到 {} 的消息：{}", sdf.format(new Date()), serverSocket.getLocalPort(), socket.getPort(), new String(bytes, 0, length));}inputStream.close();socket.close();} catch (IOException e) {e.printStackTrace();}}, Thread.currentThread().getName()).start();System.out.println(Thread.currentThread().getName());}}
}

2.1 结果演示

现象：

client1 、client2 都能正常连接到 server且正常发送、接受消息

结论：

BIO多线程提高处理能力，可以同时处理多个socket连接

存在的问题：

每个线程只能处理一个socket，当client数量大时，需要消耗大量线程资源

在这里插入图片描述

3、NIO

3.1 业务代码

当一个客户端与服务端进行连接，这个socket就会加入到一个容器中，隔一段时间遍历一次，看这个socket的read()方法能否读到数据，这样一个线程就能处理多个客户端的连接和读取了

public class RedisServerNIO {static ArrayList<SocketChannel> socketList = new ArrayList<>();static ByteBuffer byteBuffer = ByteBuffer.allocate(1024);public static void main(String[] args) throws IOException {ServerSocketChannel serverSocket = ServerSocketChannel.open();SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");serverSocket.bind(new InetSocketAddress("127.0.0.1", 6300));serverSocket.configureBlocking(false);//设置为非阻塞模式while (true) {for (SocketChannel element : socketList) {int read = element.read(byteBuffer);if (read > 0) {byteBuffer.flip();byte[] bytes = new byte[read];byteBuffer.get(bytes);//System.out.println(JSONUtil.toJsonStr(element));log("{} {}> 收到 {} 的消息：{}", sdf.format(new Date()), element.socket().getLocalPort(), element.socket().getPort(), new String(bytes, 0, read));byteBuffer.clear(); }}SocketChannel socketChannel = serverSocket.accept();if (socketChannel != null) {log("{} {}> {} 连接到服务", sdf.format(new Date()), socketChannel.socket().getLocalPort(), socketChannel.socket().getPort());socketChannel.configureBlocking(false);//设置为非阻塞模式socketList.add(socketChannel);log("{} {}> socket 数量： {} ", sdf.format(new Date()), socketChannel.socket().getLocalPort(), socketList.size());}}}
}