Java实现非阻塞式服务器的示例代码
作者：低吟不作语 发布时间：2022-01-29 19:29:11
标签：Java,非阻塞式,服务器
1.创建阻塞的服务器
当 ServerSocketChannel 与 SockelChannel 采用默认的阻塞模式时，为了同时处理多个客户的连接，必须使用多线程
public class EchoServer {
private int port = 8000;
private ServerSocketChannel serverSocketChannel = null;
private ExecutorService executorService; //线程池
private static final int POOL_MULTIPLE = 4; //线程池中工作线程的数目
public EchoServer() throws IOException {
//创建一个线程池
executorService = Executors.newFixedThreadPool(
Runtime.getRuntime().availableProcessors() * POOL_MULTIPLE);
//创建一个ServerSocketChannel对象
serverSocketChannel = ServerSocketChannel.open();
//使得在同一个主机上关闭了服务器程序，紧接着再启动该服务器程序时，可以顺利绑定相同的端口
serverSocketChannel.socket().setReuseAddress(true);
//把服务器进程与一个本地端口绑定
serverSocketChannel.socket().bind(new InetSocketAddress(port));
System.out.println("服务器启动");
}
public void service() {
while (true) {
SocketChannel socketChannel = null;
try {
socketChannel = serverSocketChannel.accept();
//处理客户连接
executorService.execute(new Handler(socketChannel));
} catch(IOException e) {
e.printStackTrace();
}
}
}
public static void main(String args[])throws IOException {
new EchoServer().service();
}
//处理客户连按
class Handler implements Runnable {
private SocketChannel socketChannel;
public Handler(SocketChannel socketChannel) {
this.socketChannel = socketChannel;
}
public void run() {
handle(socketChannel);
}
public void handle(SocketChannel socketChannel) {
try {
//获得与socketChannel关联的Socket对象
Socket socket = socketChannel.socket();
System.out.println("接收到客户连接，来自:" + socket.getInetAddress() + ":" + socket.getPort());
BufferedReader br = getReader(socket);
PrintWriter pw = getWriter(socket);
String msg = null;
while ((msg = br.readLine()) != null) {
System.out.println(msg);
pw.println(echo(msg));
if (msg.equals("bye")) {
break;
}
}
} catch (IOException e) {
e.printStackTrace();
} finally {
try {
if(socketChannel != null) {
socketChannel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
private PrintWriter getWriter(Socket socket) throws IOException {
OutputStream socketOut = socket.getOutputStream();
return new PrintWriter(socketOut,true);
}
private BufferedReader getReader(Socket socket) throws IOException {
InputStream socketIn = socket.getInputStream();
return new BufferedReader(new InputStreamReader(socketIn));
}
public String echo(String msg) {
return "echo:" + msg;
}
}
2.创建非阻塞的服务器
在非阻塞模式下，EchoServer 只需要启动一个主线程，就能同时处理三件事：
接收客户的连接
接收客户发送的数据
向客户发回响应数据
EchoServer 委托 Selector 来负责监控接收连接就绪事件、读就绪事件和写就绪事件如果有特定事件发生，就处理该事件
// 创建一个Selector对象
selector = Selector.open();
//创建一个ServerSocketChannel对象
serverSocketChannel = ServerSocketChannel.open();
//使得在同一个主机上关闭了服务器程序，紧接着再启动该服务器程序时
//可以顺利绑定到相同的端口
serverSocketChannel.socket().setReuseAddress(true);
//使ServerSocketChannel工作于非阻塞模式
serverSocketChannel.configureBlocking(false):
//把服务器进程与一个本地端口绑定
serverSocketChannelsocket().bind(new InetSocketAddress(port));
EchoServer 类的 service() 方法负责处理本节开头所说的三件事，体现其主要流程的代码如下：
public void service() throws IOException {
serverSocketChannel.reqister(selector, SelectionKey.OP_ACCEPT);
//第1层while循环
while(selector.select() > 0) {
//获得Selector的selected-keys集合
Set readyKeys = selector.selectedKeys();
Iterator it = readyKeys.iterator();
//第2层while循环
while (it.hasNext()) {
SelectionKey key = null;
//处理SelectionKey
try {
//取出一个SelectionKey
key = (SelectionKey) it.next();
//把 SelectionKey从Selector 的selected-key 集合中删除
it.remove();
1f (key.isAcceptable()) { 处理接收连接就绪事件; }
if (key.isReadable()) { 处理读就绪水件; }
if (key.isWritable()) { 处理写就绪事件; }
} catch(IOException e) {
e.printStackTrace();
try {
if(key != null) {
//使这个SelectionKey失效
key.cancel();
//关闭与这个SelectionKey关联的SocketChannel
key.channel().close();
}
} catch(Exception ex) {
e.printStackTrace();
}
}
}
}
}
首先由
ServerSocketChannel向Selector注册接收连接就绪事件，如果Selector监控到该事件发生,就会把相应的SelectionKey对象加入selected-keys集合第一层 while 循环，不断询问
Selector已经发生的事件，select()方法返回当前相关事件已经发生的SelectionKey的个数，如果当前没有任何事件发生，该方法会阻塞下去，直到至少有一个事件发生。Selector的selectedKeys()方法返回selected-keys集合，它存放了相关事件已经发生的SelectionKey对象第二层 while 循环，从
selected-keys集合中依次取出每个SelectionKey对象并从集合中删除，，然后调用isAcceptable()、isReadable()和isWritable()方法判断到底是哪种事件发生了，从而做出相应的处理
2.1处理接收连接就绪事件
if (key.isAcceptable()) {
//获得与SelectionKey关联的ServerSocketChannel
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
//获得与客户连接的SocketChannel
SocketChannel socketChannel = (SocketChannel) ssc.accept();
//把Socketchannel设置为非阻塞模式
socketChannel.configureBlocking(false);
//创建一个用于存放用户发送来的数据的级冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);
//Socketchannel向Selector注册读就绪事件和写就绪事件
socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, buffer);
}
2.2处理读就绪事件
public void receive(SelectionKey key) throws IOException {
//获得与SelectionKey关联的附件
ByteBuffer buffer = (ByteBuffer) key.attachment();
//获得与SelectionKey关联的Socketchannel
SocketChannel socketChannel = (SocketChannel)key.channel();
//创建一个ByteBuffer用于存放读到的数据
ByteBuffer readBuff = ByteBuffer.allocate(32);
socketChannel.read(readBuff);
readBuff.flip();
//把buffer的极限设为容量
buffer.limit(buffer.capacity());
//把readBuff中的内容拷贝到buffer
buffer.put(readBuff);
}
2.3处理写就绪事件
public void send(SelectionKey key) throws IOException {
//获得与SelectionKey关联的ByteBuffer
ByteBuffer buffer = (ByteBuffer) key.attachment();
//获得与SelectionKey关联的SocketChannel
SocketChannel socketChannel = (SocketChannel) key.channel();
buffer.flip();
//按照GBK编码把buffer中的字节转换为字符串
String data = decode(buffer);
//如果还没有读到一行数据就返回
if(data.indexOf("\r\n") == -1)
return;
//截取一行数据
String outputData = data.substring(0, data.indexOf("\n") + 1);
//把输出的字符串按照GBK编码转换为字节，把它放在outputBuffer中
ByteBuffer outputBuffer = encode("echo:" + outputData);
//输出outputBuffer的所有字节
while(outputBuffer,hasRemaining())
socketChannel.write(outputBuffer);
//把outputData字符审按照GBK编码，转换为字节，把它放在ByteBuffer
ByteBuffer temp = encode(outputData);
//把buffer的位置设为temp的极限
buffer.position(temp.limit()):
//删除buffer已经处理的数据
buffer.compact();
//如果已经输出了字符串“bye\r\n”，就使SelectionKey失效，并关闭SocketChannel
if(outputData.equals("bye\r\n")) {
key.cancel();
socketChannel.close();
}
}
完整代码如下：
public class EchoServer {
private int port = 8000;
private ServerSocketChannel serverSocketChannel = null;
private Selector selector;
private Charset charset = Charset.forName("GBK");
public EchoServer() throws IOException {
// 创建一个Selector对象
selector = Selector.open();
//创建一个ServerSocketChannel对象
serverSocketChannel = ServerSocketChannel.open();
//使得在同一个主机上关闭了服务器程序，紧接着再启动该服务器程序时
//可以顺利绑定到相同的端口
serverSocketChannel.socket().setReuseAddress(true);
//使ServerSocketChannel工作于非阻塞模式
serverSocketChannel.configureBlocking(false):
//把服务器进程与一个本地端口绑定
serverSocketChannelsocket().bind(new InetSocketAddress(port));
}
public void service() throws IOException {
serverSocketChannel.reqister(selector, SelectionKey.OP_ACCEPT);
//第1层while循环
while(selector.select() > 0) {
//获得Selector的selected-keys集合
Set readyKeys = selector.selectedKeys();
Iterator it = readyKeys.iterator();
//第2层while循环
while (it.hasNext()) {
SelectionKey key = null;
//处理SelectionKey
try {
//取出一个SelectionKey
key = (SelectionKey) it.next();
//把 SelectionKey从Selector 的selected-key 集合中删除
it.remove();
1f (key.isAcceptable()) {
//获得与SelectionKey关联的ServerSocketChannel
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
//获得与客户连接的SocketChannel
SocketChannel socketChannel = (SocketChannel) ssc.accept();
//把Socketchannel设置为非阻塞模式
socketChannel.configureBlocking(false);
//创建一个用于存放用户发送来的数据的级冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);
//Socketchannel向Selector注册读就绪事件和写就绪事件
socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, buffer);
}
if (key.isReadable()) { receive(key); }
if (key.isWritable()) { send(key); }
} catch(IOException e) {
e.printStackTrace();
try {
if(key != null) {
//使这个SelectionKey失效
key.cancel();
//关闭与这个SelectionKey关联的SocketChannel
key.channel().close();
}
} catch(Exception ex) {
e.printStackTrace();
}
}
}
}
}
public void receive(SelectionKey key) throws IOException {
//获得与SelectionKey关联的附件
ByteBuffer buffer = (ByteBuffer) key.attachment();
//获得与SelectionKey关联的Socketchannel
SocketChannel socketChannel = (SocketChannel)key.channel();
//创建一个ByteBuffer用于存放读到的数据
ByteBuffer readBuff = ByteBuffer.allocate(32);
socketChannel.read(readBuff);
readBuff.flip();
//把buffer的极限设为容量
buffer.limit(buffer.capacity());
//把readBuff中的内容拷贝到buffer
buffer.put(readBuff);
}
public void send(SelectionKey key) throws IOException {
//获得与SelectionKey关联的ByteBuffer
ByteBuffer buffer = (ByteBuffer) key.attachment();
//获得与SelectionKey关联的SocketChannel
SocketChannel socketChannel = (SocketChannel) key.channel();
buffer.flip();
//按照GBK编码把buffer中的字节转换为字符串
String data = decode(buffer);
//如果还没有读到一行数据就返回
if(data.indexOf("\r\n") == -1)
return;
//截取一行数据
String outputData = data.substring(0, data.indexOf("\n") + 1);
//把输出的字符串按照GBK编码转换为字节，把它放在outputBuffer中
ByteBuffer outputBuffer = encode("echo:" + outputData);
//输出outputBuffer的所有字节
while(outputBuffer,hasRemaining())
socketChannel.write(outputBuffer);
//把outputData字符审按照GBK编码，转换为字节，把它放在ByteBuffer
ByteBuffer temp = encode(outputData);
//把buffer的位置设为temp的极限
buffer.position(temp.limit()):
//删除buffer已经处理的数据
buffer.compact();
//如果已经输出了字符串“bye\r\n”，就使SelectionKey失效，并关闭SocketChannel
if(outputData.equals("bye\r\n")) {
key.cancel();
socketChannel.close();
}
}
//解码
public String decode(ByteBuffer buffer) {
CharBuffer charBuffer = charset.decode(buffer);
return charBuffer.toStrinq();
}
//编码
public ByteBuffer encode(String str) {
return charset.encode(str);
}
public static void main(String args[])throws Exception {
EchoServer server = new EchoServer();
server.service();
}
}
3.阻塞模式与非阻塞模式混合使用
使用非阻塞模式时，ServerSocketChannel 以及 SocketChannel 都被设置为非阻塞模式，这使得接收连接、接收数据和发送数据的操作都采用非阻塞模式，EchoServer 采用一个线程同时完成这些操作
假如有许多客户请求连接，可以把接收客户连接的操作单独由一个线程完成，把接收数据和发送数据的操作由另一个线程完成，这可以提高服务器的并发性能
负责接收客户连接的线程按照阻塞模式工作，如果收到客户连接，就向 Selector 注册读就绪和写就绪事件，否则进入阻塞状态，直到接收到了客户的连接。负责接收数据和发送数据的线程按照非阻塞模式工作，只有在读就绪或写就绪事件发生时，才执行相应的接收数据和发送数据操作
public class EchoServer {
private int port = 8000;
private ServerSocketChannel serverSocketChannel = null;
private Selector selector = null;
private Charset charset = Charset.forName("GBK");
public EchoServer() throws IOException {
selector = Selector.open();
serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().setReuseAddress(true);
serverSocketChannelsocket().bind(new InetSocketAddress(port));
}
public void accept() {
while(true) {
try {
SocketChannel socketChannel = serverSocketChannel.accept();
socketChannel.configureBlocking(false);
ByteBuffer buffer = ByteBuffer.allocate(1024);
synchronized(gate) {
selector.wakeup();
socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, buffer);
}
} catch(IOException e) {
e.printStackTrace();
}
}
}
private Object gate=new Object();
public void service() throws IOException {
while(true) {
synchronized(gate){}
int n = selector.select();
if(n == 0) continue;
Set readyKeys = selector.selectedKeys();
Iterator it = readyKeys.iterator();
while (it.hasNext()) {
SelectionKey key = null;
try {
it.remove();
if (key.isReadable()) {
receive(key);
}
if (key.isWritable()) {
send(key);
}
} catch(IOException e) {
e.printStackTrace();
try {
if(key != null) {
key.cancel();
key.channel().close();
}
} catch(Exception ex) { e.printStackTrace(); }
}
}
}
}
public void receive(SelectionKey key) throws IOException {
...
}
public void send(SelectionKey key) throws IOException {
...
}
public String decode(ByteBuffer buffer) {
...
}
public ByteBuffer encode(String str) {
...
}
public static void main(String args[])throws Exception {
final EchoServer server = new EchoServer();
Thread accept = new Thread() {
public void run() {
server.accept();
}
};
accept.start();
server.service();
}
}
注意一点：主线程的 selector select() 方法和 Accept 线程的 register(...) 方法都会造成阻塞，因为他们都会操作 Selector 对象的共享资源 all-keys 集合，这有可能会导致死锁
导致死锁的具体情形是：Selector 中尚没有任何注册的事件，即 all-keys 集合为空，主线程执行 selector.select() 方法时将进入阻塞状态，只有当 Accept 线程向 Selector 注册了事件，并且该事件发生后，主线程才会从 selector.select() 方法返回。然而，由于主线程正在 selector.select() 方法中阻塞，这使得 Acccept 线程也在 register() 方法中阻塞。Accept 线程无法向 Selector 注册事件，而主线程没有任何事件可以监控，所以这两个线程将永远阻塞下去
为了避免对共享资源的竞争，同步机制使得一个线程执行 register() 时，不允许另一个线程同时执行 select() 方法，反之亦然
来源：https://www.cnblogs.com/Yee-Q/p/17416992.html
0
投稿猜你喜欢
- 数组是一种数据结构，其声明方式如下：type[] arrayName;数组具有以下属性： 1.数组可以是
前言毕业季来临，很多小伙伴在忙于考公或者准备研究生复试等工作，因此从网上下载或者购买了源码。源码在本地运行成功之后，想要稍微修改一下，结果发- PowerPoint幻灯片中可插入公式，用于在幻灯片放映时演示相关内容的论证、推算的依据，能有效地为演讲者提供论述的数据支撑。通过后端程序代
- Android 集成FlutterFlutter 作为 Google 开源的新一代跨平台、高性能 UI 框架，旨在帮助开发者高效地构建出跨平
- 本文提供了基于MD5加密16位和32位的方法，具体内容如下import java.io.IOException;import java.ma
- SpringBoot应用启动run方法SpringApplication.java 中执行的代码@SpringBootApplication
- 想要在Ubuntu上运行java程序，可以将java程序编译成功后打包，然后在Ubuntu上用命令执行jar文件具体操作如下：1、Windo
- 题目要求思路一：枚举 + 二分逐一枚举值域内的所有值，然后二分判断是否合法。Javaclass Solution { &nbs
- 前言感觉Jpa的动态构建查询不好使用，然后mybatis-plus没有动态构建表的功能，有没有可能使两者混合使用，利用Jpa自动建表的优势
- 本文实例讲述了Java正则验证IP的方法。分享给大家供大家参考，具体如下：网上用正则验证IP的表达式有很多，一搜一大堆，可以自己写，但很麻烦
- 本文实例讲述了C#实现对Json字符串处理方法，分享给大家供大家参考。具体分析如下：一般对于web应用开发人员来说对Json字符串都会很熟悉
- ClickHouse应用场景：1.绝大多数请求都是用于读访问的2.数据需要以大批次（大于1000行）进行更新，而不是单行更新；或者根本没有更
- 在POI的使用过程中，对大多数API User来说经常面临两个问题，这也是GridExcel致力解决的问题。问题1. 仅使用简单的导入导出功
- 一、前言最近接到一个任务,需要爬取五级行政区划的所有数据(大概71万条数据在),需要爬取的网站:行政区划 - 行政区划代码查询 发
- 本文实例讲述了Java实现读取及生成Excel文件的方法。分享给大家供大家参考，具体如下：一、读取Excel文件需要先下载poi-3.0.1
- 前言 Windows下实现摄像视频捕捉有多种实现方式；各种方式的优劣，本文不做对比。但是，opencv是一款老牌开发库，在图像处理
- 一、静态静态的定时任务可以直接使用注解@Scheduled，并在启动类上配置@EnableScheduling即可@PostMapping(
- 如果使用了反向代理软件，将http://192.168.1.110:2046/ 的URL反向代理为http://www.xxx.com/ 的
- 问题：1.线程 wait()方法使用有什么前提？2. 多线程之间如何进行通信？3. Java 中 notify 和 notifyAll 有什
- 一、什么是ASMASM是一个java字节码操纵框架，它能被用来动态生成类或者增强既有类的功能。ASM 可以直接产生二进制 class 文件，
