Redis IO多路复用

news2024/11/17 2:41:20

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());
			}
		}
	}
}