青木大师
https://github.com/goudai/gd-rpc

在这个基础上,抽取的Reactor模式简化理解的版本
这个JAVA nio和多线程结合的问题,已经困扰了好几年了.
通过阅读青木的代码,终于有点明白了

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
class NIOTest {
public static void main(String[] args) throws IOException {
Acceptor acceptor = new Acceptor(9090, new Reactor(10));
acceptor.start();
Socket socket = new Socket("127.0.0.1", 9090);
socket.getOutputStream().write("helloworld\n".getBytes());
socket.getOutputStream().write("helloworld\n".getBytes());
socket.getOutputStream().write("helloworld\n".getBytes());
socket.getOutputStream().flush();
BufferedReader br = new BufferedReader(new InputStreamReader(socket.getInputStream()));
String str = null;
while ((str = br.readLine()) != null) {
System.out.println(str);
}
socket.close();
}
}
class Reactor {
private final Worker[] workers;
private final int workerCount;
private final static AtomicInteger SELECTORINDEX = new AtomicInteger();
public Reactor(int workerCount) throws IOException {
this.workerCount = workerCount;
this.workers = new Worker[workerCount];
for (int i = 0; i < workerCount; i++) {
workers[i] = new Worker("worker_" + i);
workers[i].start();
}
}
public void register(SocketChannel socket) {
Channel channel = new Channel(socket, SelectionKey.OP_READ);
this.workers[SELECTORINDEX.getAndIncrement() % this.workerCount].register(channel);
}
class Worker extends Thread {
private final Selector selector;
private final String name;
private final Queue<Channel> asyncRegistrySocketChannels = new ConcurrentLinkedQueue<Channel>();
private final Charset charset;
private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
public Worker(String name) throws IOException {
this.name = name;
this.charset = Charset.forName("utf8");
selector = Selector.open();
}
public void register(Channel channel) {
this.asyncRegistrySocketChannels.offer(channel);
this.selector.wakeup();
}
private void asyncRegister() throws ClosedChannelException {
Channel channel = null;
while ((channel = asyncRegistrySocketChannels.poll()) != null) {
channel.getSocketChannel().register(selector, SelectionKey.OP_READ);
}
}
@Override
public void run() {
while (true) {
try {
asyncRegister();
selector.select();
Set<SelectionKey> selectionKeys = this.selector.selectedKeys();
for (SelectionKey key : selectionKeys) {
if (key.isValid() && key.isReadable()) {
Object obj = key.attachment();
SocketChannel channel = (SocketChannel) key.channel();
StringBuilder sb;
if (obj == null) {
sb = new StringBuilder();
} else {
sb = (StringBuilder) obj;
}
ByteBuffer bb = ByteBuffer.allocate(1024);
int count = 0;
while ((count = channel.read(bb)) > 0) {
bb.flip();
CharBuffer cb = charset.decode(bb);
sb.append(cb.toString());
bb.compact();
}
if (sb.charAt(sb.length() - 1) != '\n') {
channel.register(selector, SelectionKey.OP_READ, sb);
} else {
channel.register(selector, SelectionKey.OP_WRITE, sb);
}
} else if (key.isValid() && key.isWritable()) {
StringBuilder sb = (StringBuilder) key.attachment();
SocketChannel channel = (SocketChannel) key.channel();
for (String s : sb.toString().split("\n")) {
String str = sdf.format(new Date()) + ":" + s + "\n";
channel.write(ByteBuffer.wrap(str.toString().getBytes()));
}
channel.register(selector, SelectionKey.OP_READ, null);
}
}
selectionKeys.clear();
} catch (IOException e) {
e.printStackTrace();
}
}
}
};
class Channel {
public SocketChannel getSocketChannel() {
return socketChannel;
}
public void setSocketChannel(SocketChannel socketChannel) {
this.socketChannel = socketChannel;
}
public int getOps() {
return ops;
}
public void setOps(int ops) {
this.ops = ops;
}
public Channel(SocketChannel socketChannel, int ops) {
this.socketChannel = socketChannel;
this.ops = ops;
}
private SocketChannel socketChannel;
private int ops;
}
}
class Acceptor extends Thread {
private final Selector selector;
private final Reactor reactor;
private final ServerSocketChannel serverSocketChannel;
public Acceptor(int port, Reactor reactor) throws IOException {
this.selector = Selector.open();
this.reactor = reactor;
this.serverSocketChannel = (ServerSocketChannel) ServerSocketChannel.open().bind(new InetSocketAddress(port))
.configureBlocking(false);
this.serverSocketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
this.serverSocketChannel.setOption(StandardSocketOptions.SO_RCVBUF, 1024 * 16 * 2);
this.serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
}
@Override
public void run() {
while (true) {
try {
int i = selector.select();
if (i > 0) {
Iterator<SelectionKey> it = selector.selectedKeys().iterator();
SelectionKey key = it.next();
if (key.isValid() && key.isAcceptable()) {
ServerSocketChannel server = (ServerSocketChannel) key.channel();
SocketChannel socket = server.accept();
socket.configureBlocking(false);
reactor.register(socket);
} else {
key.cancel();
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}

ITPub博客

Reactor模式

最新文章