Non Blocking I/O – DECODE X NET

シングルスレッドで複数のクライアントからのリクエストをサーバで処理するとき、ノンブロッキングI/Oを使った方法がとられます。一般的なHTTPサーバーなどはスレッドごとに受信処理を割り当てますが、小規模でデータサイズが小さい処理など場合は多数のリクエストに対して高いパフォーマンスを発揮します。
UDPは、もともとステートレスなのでこのようなしくみを使う必要はないのでは、という疑問もありますが、サーバーサイドプログラミングというのは、いろんなケースを想定する必要があるため、セレクタによる監視機能を使った方が安定するという気がしています。
最近使うことがあったので、今後の参考としてTCP、UDPの両方を置いておきたいと思います。

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
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.util.Iterator;
import java.lang.Thread;

public class TcpServer {
    public static void main(String[] args) {
        try {
            Selector selector = Selector.open();
            ServerSocketChannel channel = ServerSocketChannel.open();
            channel.configureBlocking(false);
            channel.socket().bind(new InetSocketAddress(9999));
            channel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("Launch Server Port=" + channel.socket().getLocalPort());
            while (selector.select() > 0) {
                for (Iterator<SelectionKey> it = selector.selectedKeys().iterator(); it.hasNext();) {
                    SelectionKey key = (SelectionKey) it.next();
                    it.remove();
                    if (key.isAcceptable()) {
                        ServerSocketChannel serverChannel2 = (ServerSocketChannel) key.channel();
                        SocketChannel channel2 = serverChannel2.accept();
                        doAccept(channel2);
                        channel2.register(selector, SelectionKey.OP_READ);
                    } else if (key.isReadable()) {
                        doRead((SocketChannel) key.channel());
                    }
                }
            }
        } catch (IOException e) {
        }
    }
    private static void doAccept(SocketChannel channel) {
        try {
            String remoteAddress = channel.socket().getRemoteSocketAddress().toString();
            System.out.println(remoteAddress + ": Accepted");
            channel.configureBlocking(false);
        } catch (IOException e) {
        }
    }
    private static void doRead(SocketChannel channel) {
        ByteBuffer buf = ByteBuffer.allocate(1024);
        Charset charset = Charset.forName("UTF-8");
        String remote = channel.socket().getRemoteSocketAddress().toString();
        try {
            if (channel.read(buf) < 0) {
                return;
            }
            buf.flip();
            System.out.print(remote + ":" + charset.decode(buf).toString());
            buf.flip();
            Thread.sleep(100);
            channel.write(buf);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            System.out.println(remote + ": disconnected");
            try {
                channel.close();
            } catch (IOException e) {}
        }
    }
}

import java.io.IOException;

import java.net.InetSocketAddress;

import java.nio.ByteBuffer;

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.util.Iterator;

import java.lang.Thread;

public class TcpServer {

public static void main(String[] args) {

try {

Selector selector = Selector.open();

ServerSocketChannel channel = ServerSocketChannel.open();

channel.configureBlocking(false);

channel.socket().bind(new InetSocketAddress(9999));

channel.register(selector, SelectionKey.OP_ACCEPT);

System.out.println("Launch Server Port=" + channel.socket().getLocalPort());

while (selector.select() > 0) {

for (Iterator<SelectionKey> it = selector.selectedKeys().iterator(); it.hasNext();) {

SelectionKey key = (SelectionKey) it.next();

it.remove();

if (key.isAcceptable()) {

ServerSocketChannel serverChannel2 = (ServerSocketChannel) key.channel();

SocketChannel channel2 = serverChannel2.accept();

doAccept(channel2);

channel2.register(selector, SelectionKey.OP_READ);

} else if (key.isReadable()) {

doRead((SocketChannel) key.channel());

}

} catch (IOException e) {

}

private static void doAccept(SocketChannel channel) {

try {

String remoteAddress = channel.socket().getRemoteSocketAddress().toString();

System.out.println(remoteAddress + ": Accepted");

channel.configureBlocking(false);

} catch (IOException e) {

}

private static void doRead(SocketChannel channel) {

ByteBuffer buf = ByteBuffer.allocate(1024);

Charset charset = Charset.forName("UTF-8");

String remote = channel.socket().getRemoteSocketAddress().toString();

try {

if (channel.read(buf) < 0) {

return;

}

buf.flip();

System.out.print(remote + ":" + charset.decode(buf).toString());

buf.flip();

Thread.sleep(100);

channel.write(buf);

} catch (Exception e) {

e.printStackTrace();

} finally {

System.out.println(remote + ": disconnected");

try {

channel.close();

} catch (IOException e) {}

}

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.lang.Thread;

public class UdpServer {
    public static void main(String[] args) {
        try {
            Selector selector = Selector.open();
            DatagramChannel channel = DatagramChannel.open();
            InetSocketAddress inetSocketAddress = new InetSocketAddress(9999);
            channel.socket().bind(inetSocketAddress);
            channel.configureBlocking(false);
            channel.register(selector, SelectionKey.OP_READ);
            System.out.println("Launch Server Port=" + channel.socket().getLocalPort());
            while (selector.select() > 0) {
                for (Iterator<SelectionKey> iterator 
                     = selector.selectedKeys().iterator();
                                      iterator.hasNext();) {
                    SelectionKey key = (SelectionKey) iterator.next();
                    iterator.remove();
                    if (key.isReadable()) {
                        doReceive((DatagramChannel)key.channel());
                    }
                }
            }
        } catch (Exception e) {
        } 
    }
    private static void doReceive(DatagramChannel channel) {
        ByteBuffer buf = ByteBuffer.allocate(1024);
        Charset charset = Charset.forName("UTF-8");
        try {
            channel.receive(buf);
            buf.flip();
            System.out.print( "Receive: " + charset.decode(buf).toString());
            Thread.sleep(100);
            channel.write(buf);

        } catch (Exception e) {
        }
    }
}

import java.net.InetSocketAddress;

import java.nio.ByteBuffer;

import java.nio.channels.DatagramChannel;

import java.nio.channels.SelectionKey;

import java.nio.channels.Selector;

import java.nio.charset.Charset;

import java.util.Iterator;

import java.lang.Thread;

public class UdpServer {

public static void main(String[] args) {

try {

Selector selector = Selector.open();

DatagramChannel channel = DatagramChannel.open();

InetSocketAddress inetSocketAddress = new InetSocketAddress(9999);

channel.socket().bind(inetSocketAddress);

channel.configureBlocking(false);

channel.register(selector, SelectionKey.OP_READ);

System.out.println("Launch Server Port=" + channel.socket().getLocalPort());

while (selector.select() > 0) {

for (Iterator<SelectionKey> iterator

= selector.selectedKeys().iterator();

iterator.hasNext();) {

SelectionKey key = (SelectionKey) iterator.next();

iterator.remove();

if (key.isReadable()) {

doReceive((DatagramChannel)key.channel());

}

} catch (Exception e) {

}

private static void doReceive(DatagramChannel channel) {

ByteBuffer buf = ByteBuffer.allocate(1024);

Charset charset = Charset.forName("UTF-8");

try {

channel.receive(buf);

buf.flip();

System.out.print( "Receive: " + charset.decode(buf).toString());

Thread.sleep(100);

channel.write(buf);

} catch (Exception e) {

}

これまではC言語で実装することが多かったので、Javaでは初めてでした。(仕事ではよく使うものの、このブログもJava関係はめずらしい)

クライアントには、ncコマンドを使用。

nc localhost 9999
nc localhost 9999 -u

ソケットレベルのサーバプログラム自作して長く安定して動かすには、クライアントアプリよりも10倍以上難しいと思っています。(それゆえ実用サーバでは枯れた技術を組みあわせて使われる)
最近はプラットホームがすでに出来上がっていたりするため、なかなかこのレベルから作ることは少ないですが、私は相変わらず面白い分野だと思っています。