单线程Reactor
package org.example.utils.echo.single;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
public class EchoServerReactor implements Runnable{
Selector selector;
ServerSocketChannel serverSocketChannel;
EchoServerReactor() throws IOException {
//Reactor初始化
selector = Selector.open();
serverSocketChannel = ServerSocketChannel.open();
InetSocketAddress address =
new InetSocketAddress("localhost",
8848);
//非阻塞
serverSocketChannel.configureBlocking(false);
//分步处理,第一步,接收accept事件
SelectionKey sk =
serverSocketChannel.register(selector,0,new AcceptorHandler());
// SelectionKey.OP_ACCEPT
serverSocketChannel.socket().bind(address);
System.out.println("服务端已经开始监听:"+address);
sk.interestOps(SelectionKey.OP_ACCEPT);
}
@Override
public void run() {
try {
while (!Thread.interrupted()){
selector.select();
Set<SelectionKey> selected=selector.selectedKeys();
Iterator<SelectionKey> it=selected.iterator();
while (it.hasNext()){
SelectionKey sk=it.next();
dispatch(sk);
}
selected.clear();
}
} catch (IOException e) {
throw new RuntimeException(e);
}
}
private void dispatch(SelectionKey sk) {
Runnable handler=(Runnable) sk.attachment();
if (handler!=null){
handler.run();
}
}
class AcceptorHandler implements Runnable{
@Override
public void run() {
try {
SocketChannel channel=serverSocketChannel.accept();
if (channel!=null)
new EchoHandler(selector,channel);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
public static void main(String[] args) throws IOException {
new Thread(new EchoServerReactor()).start();
}
}
package org.example.utils.echo.single;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
public class EchoHandler implements Runnable{
final SocketChannel channel;
final SelectionKey sk;
final ByteBuffer byteBuffer=ByteBuffer.allocate(1024);
static final int RECIEVING=0,SENDING=1;
int state=RECIEVING;
EchoHandler(Selector selector,SocketChannel c) throws IOException {
channel=c;
c.configureBlocking(false);
sk=channel.register(selector,0);
sk.attach(this);
sk.interestOps(SelectionKey.OP_READ);
selector.wakeup();
}
@Override
public void run() {
try {
if (state==SENDING){
channel.write(byteBuffer);
byteBuffer.clear();
sk.interestOps(SelectionKey.OP_READ);
state=RECIEVING;
}else if (state==RECIEVING){
int length=0;
while ((length=channel.read(byteBuffer))>0)
{
System.out.println(new String(byteBuffer.array(),0,length));
}
byteBuffer.flip();
sk.interestOps(SelectionKey.OP_WRITE);
state=SENDING;
}
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
结果:
原理无非就是:
多线程,无非就是搞多个Reactor , 一个专门接受accept , 一个专门dispatch , 再搞一个多线程池处理handle
这里面最主要的就是
handle类,sk.attach(this);把对象传回reactor
参考文献:
java高并发核心编程. 卷1,NIO、Netty、Redis、ZooKeeper (尼恩)