Obsługa sieci w Javie

Java posiada dwa API do obsługi sieci: java.io oraz java.nio.

Obsługa wielu gniazd, operacje blokujące / nieblokujące:

java.io pozwala tylko na blokującą obsługę gniazd i nie ma możliwości sprawdzenia czy możliwe jest wykonanie żądanej operacji bez blokowania (tzn. nie ma możliwości skorzystania z funkcji typu select/poll).
W praktyce oznacza to, że dla każdego gniazda potrzebny jest osobny wątek.
java.nio oferuje mechanizm analogiczny w działaniu do mechanizmu epoll. Przy jego wykorzystaniu narzucana jest nieblokująca obsługa gniazd.

Odczyt / zapis danych:

java.io używa strumieni - wysyłanie i odbieranie realizuje się przez obiekty z klas InputStream i OutputStream
java.nio udostępnia metody wysyłające dane z i odbierające dane do buforów – obiektowo opakowanych tablic bajtów.

java.io jest prostszym API, które często utrudnia bądź nawet uniemożliwia napisanie wydajnego kodu.

Dłuższe porównanie IO/NIO: http://tutorials.jenkov.com/java-nio/nio-vs-io.html

java.io

Java I/O – proste i ładne blokujące opakowanie obiektowe gniazd.

Klient TCP

//Socket sock = new Socket();
//sock.connect(new InetSocketAddress("example.com", 13));
Socket sock = new Socket("example.com", 13);
 
InputStream is = sock.getInputStream();
 
byte[] bytearr = new byte[16]; 
while(true){
	int len = is.read(bytearr);
	if(len == -1) break;
	System.out.write(bytearr, 0, len);
}
 
sock.close();

Serwer TCP

ServerSocket ssock = new ServerSocket(1313);
 
Socket sock = ssock.accept();
ssock.close();
 
String s = new Date().toLocaleString() + '\n';
 
sock.getOutputStream().write(s.getBytes());
sock.close();

UDP

Klient

DatagramSocket sock = new DatagramSocket();
 
InetSocketAddress addr = new InetSocketAddress("example.com", 13);
 
DatagramPacket outpacket = new DatagramPacket(new byte [0], 0, addr);
sock.send(outpacket);
 
DatagramPacket inpacket = new DatagramPacket(new byte[1024], 1024);
sock.receive(inpacket);
 
System.out.write(inpacket.getData(), 0, inpacket.getLength());
sock.close();

Serwer

Serwer UDP od klient różni się tym, że używa konstruktora przyjmującego numer portu lub metody bind

Multicast

Multicast używa klasy MulticastSocket która jest klasą DatagramSocket wzbogaconą o kilka metod, m. inn. dołączenie do grupy:

MulticastSocket socket = new MulticastSocket(1313);
 
socket.joinGroup(InetAddress.getByName("239.255.123.45"));

Od Javy 14 metoda MulticastSocket.joinGroup(InetAddress) jest oznaczona jako deprecated i żeby dodać się na wszystkie interfejsy należy ręcznie dodać się do każdego po kolei

MulticastSocket socket = new MulticastSocket(1313);
Enumeration<NetworkInterface> netIfEnum = NetworkInterface.getNetworkInterfaces();
while(netIfEnum.hasMoreElements()){
    NetworkInterface netIf = netIfEnum.nextElement();
    if(netIf.supportsMulticast())
        try {
            socket.joinGroup(new InetSocketAddress(InetAddress.getByName("239.255.123.45"), 1313), netIf);
        } catch(IOException e) {
        }
}

Zadania

Przypomnienie wątków i synchronizacji w Javie: synchronized (czyli zamki)
java.util.concurrent
nowy wątek, używając lambdy:

new Thread(() -> {
	// Kod uruchamiany wewnątrz wątku
}).start();

Zadanie 1. Napisz klienta TCP z użyciem java.io i wątków.
(Możesz skorzystać z szablonu kodu z TODO's do zrobienia: tcpclienttemplate.java)

Zadanie 2. Napisz program odbierający i wysyłający wiadomości UDP od/do grupy multicastowej z użyciem java.io i wątków.

IP multicast - przypomnienie i komendy do testowania

Zadanie 3. Napisz serwer czatu z użyciem java.io. (Jako klienta użyj np. programu z zadania 1 lub netcat lub socat.)

java.nio

Java New I/O – wydajne opakowanie obiektowe gniazd.
http://tutorials.jenkov.com/java-nio/index.html

Klasy obsługujące gniazda (kanały)

Ważne klasy (sieciowe):

SocketChannel – gniazdo TCP (connect, klient)
ServerSocketChannel – gniazdo TCP (listen, serwer)
DatagramChannel – gniazdo UDP
Selector – multiplekser

Obiekty z tych klas są tworzone metodą statyczną open(), np:

ServerSocketChannel ssc = ServerSocketChannel.open()

Kanały wprawdzie pozwalają na pracę w trybie blokującym i nieblokującym, ale przy użycie selektora wymusza pracę w trybie nieblokującym.

Praca z selektorem

Aby używać kanału NIO z selektorem, należy:

stworzyć selektor i kanały gniazd
przestawić kanały w tryb nieblokujący: configureBlocking(false)
dodać kanały gniazd do selektora – metoda register() na kanale
w pętli
1. wykonywać metodę select() selektora
2. metodą selectedKeys() pobierać kolekcję SelectionKey opisujących zdarzenia
3. obsłużyć zdarzenia
4. wyczyścić kolekcję kluczy

/*  1.  */ Selector selector = Selector.open();
/*  1.  */ DatagramChannel channel = DatagramChannel.open();
/*  1.  */ channel.bind(new InetSocketAddress(12345));
/*  2.  */ channel.configureBlocking(false);
/*  3.  */ selector.register(channel, SelectionKey.OP_READ);
/*  4.  */ …
/* 4.a. */    selector.select();
/* 4.b. */    for(SelectionKey key : selector.selectedKeys())
/* 4.c. */       …
/* 4.d. */    selector.selectedKeys().clear();

Bufory

NIO używa dedykowanych klas buforów – np. ByteBuffer. Klasy te służą opakowaniu tablic (np. tablicy bajtów) w sposób nie ograniczający wydajności.

Bufor można tworzyć metodami statycznymi, np. ByteBuffer.allocate() i ByteBuffer.wrap().

Każdy bufor ma stałą pojemność. Poza pojemnością każdy bufor ma kursor (position) i limit (limit)¹⁾. Pisanie i czytanie odbywa się zawsze od kursora i przesuwa kursor o ilość przeczytanych/zapisanych bajtów.
Do zmiany pozycji i limitu służą np. flip, clear.

ByteBuffer buffer = ByteBuffer.allocate(16); //  |................|
                                             //   ^pos(ition)     ^limit
int bytesRead;
bytesRead = channel.read(buffer);            //  |ala.............|
assert bytesRead == 3;                       //      ^pos         ^limit
 
bytesRead = channel.read(buffer);            //  |alamakota.......|
assert bytesRead == 6;                       //            ^pos   ^limit
 
buffer.flip();                               //  |alamakota.......|
                                             //   ^pos     ^limit
 
int bytesWritten;                            //            vlimit 
bytesWritten = channel2.write(buffer);       //  |alamakota.......|
assert bytesWritten == 9;                    //            ^pos
 
buffer.flip();                               //  |alamakota.......|
                                             //   ^pos     ^limit
 
short useless = buffer.getShort();           //  |alamakota.......|
assert useless == 0x616c;                    //     ^pos   ^limit
 
Charset utf8 = Charset.forName("UTF-8");     //            vlimit 
String str = utf8.decode(buffer).toString(); //  |alamakota.......|
assert str.equals(new String("amakota"));    //            ^pos
 
buffer.clear();                              //  |alamakota.......|
                                             //   ^pos            ^limit

SelectionKey

W selektorze jeden kanał gniazdo może być zarejestrowane co najwyżej raz do oczekiwania na podane zdarzenia przy pomocy metody register(). Metoda ta zwraca klucz – obiekt z klasy SelectionKey. Ten sam obiekt po wywołaniu na selektorze metody select() jest kopiowany do selectedKeys() selektora. Klucz można też wydobyć wywołując na kanale metodę keyFor(selector). Kanał można wyciągnąć z klucza metodą channel().
Aby zmienić listę oczekiwanych zdarzeń, należy wykonać na kluczu metodę interestOps() z nową listą zdarzeń.

Po wywołaniu metody select() selektora każdy klucz ma ustawiane readyOps() - listę gotowych operacji. Dla ułatwienia klasa SelectionKey ma też metody isReadable(), isWriteable(), isAcceptable(), …

Do usunięcia klucza z selektora należy użyć metody cancel() na kluczu (lub zamknąć kanał metodą close()).

SelectionKey key = selector.register(channel, SelectionKey.OP_READ);
 
selector.select();
assert key == selector.selectedKeys().iterator().next();
 
if(key.isValid() && key.isReadable())
  …
 
key.interestOps(SelectionKey.OP_READ|SelectionKey.OP_WRITE);

Każdy klucz może mieć dołączony jeden załącznik, albo podany przy rejestracji kanału w selektorze, albo dołączony metodą attach(object). Można go później pobrać metodą attachment().

MyClass myObj = …
selector.register(myObj.myChannel(), SelectionKey.OP_READ, myObj);
 
selector.select();
SelectionKey key = selector.selectedKeys().iterator().next();
 
assert key.attachment() instanceof MyClass;
((MyClass)key.attachment()).myFunction(key);

"klient tcp"

java.nio nie pozwala na współpracę ze standardowym wejściem.

Przykład obsługujący jedno gniazdo

NioCli.java

package sk2;
 
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
 
public class NioCli {
    private SocketChannel sock;                     // gniazdo w NIO
    private Selector sel;                           // selektor – opakowuje mechanizm typu 'select'
    private SelectionKey sockKey;                   
    private ByteBuffer bb = ByteBuffer.allocate(8); // bufor – odpowiednio opakowana tablica bajtów
 
    public NioCli(String[] args) throws Throwable{
        // Selektory, gniazda, etc. są tworzone przez metody statyczne, używające dostarczanej
        // przez VM implementacji klasy SelectorProvider
        sel = Selector.open();
        sock = SocketChannel.open(new InetSocketAddress(args[0], Integer.parseInt(args[1])));
 
        sock.configureBlocking(false);              // Używanie Selectora wymaga nieblokującego I/O
        sockKey = sock.register(sel, SelectionKey.OP_READ); // Tak każe się czekać na zdarzenie
    }
 
    private void select() throws Throwable{
        sel.select();                               // oczekiwanie na zdarzenie
 
        // selectedKeys() zawiera listę kluczy dla których można wykonać żądaną operację
        // i ustawia im dostępne zdarzenia (readyOps, isReadable, is…)
        assert sel.selectedKeys().size() == 1;      
        assert sel.selectedKeys().iterator().next() == sockKey;
 
        bb.clear();                                 // przygotowanie bufora do pracy
        int count = sock.read(bb);
 
        if(count == -1) {                           // -1 oznacza EOF
            sockKey.cancel();                       // cancel usuwa klucz z selektora
            return;
        }
 
        System.out.write(bb.array(), 0, bb.position());
 
        sel.selectedKeys().clear();                 // klucze są w zbiorze do momentu usunięcia
    }
 
    public static void main(String[] args) throws Throwable{
        NioCli cli = new NioCli(args);
        while(cli.sockKey.isValid())
            cli.select();
    }
}

Serwer czatu

Przykład serwera czatu

NioServ.java

package sk2;
 
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.util.HashSet;
 
public class NioServ {
 
    ServerSocketChannel ssc;
    Selector sel;
    HashSet<SocketChannel> clients = new HashSet<SocketChannel>();
    ByteBuffer commonReadBuffer = ByteBuffer.allocate(256);
 
    public NioServ(int port) throws Throwable {
        sel = Selector.open();
        ssc = ServerSocketChannel.open();
        ssc.bind(new InetSocketAddress(port));
        ssc.configureBlocking(false);
        ssc.register(sel, SelectionKey.OP_ACCEPT);
    }
 
    private void loop() throws Throwable {
        while(true){
            select();
        }
    }
 
    private void select() throws Throwable {
        sel.select();
        for(SelectionKey key : sel.selectedKeys()){
            if(key.isAcceptable()){
                accept();
            }
            if(key.isReadable()){
                SocketChannel sc = (SocketChannel) key.channel();
                read(key, sc);
            }
        }
        sel.selectedKeys().clear();
    }
 
    private void accept() throws Throwable {
        SocketChannel sc = ssc.accept();
        sc.configureBlocking(false);
        sc.register(sel, SelectionKey.OP_READ);
        sendAll("Connected:" + sc.getRemoteAddress().toString() + "\n");
        clients.add(sc);
    }
 
    private void read(SelectionKey key, SocketChannel sc) throws Throwable {
        commonReadBuffer.clear();
        commonReadBuffer.put(sc.getRemoteAddress().toString().getBytes());
        commonReadBuffer.putChar(':');
        commonReadBuffer.putChar(' ');
 
        int count = sc.read(commonReadBuffer);
 
        if(count == 0) return;
        if(count == -1){
            clients.remove(sc);
            sendAll("Disconnected:" + sc.getRemoteAddress().toString() + "\n");
            key.cancel();
            sc.close();
            return;
        }
 
        sendAll(commonReadBuffer);
    }
 
    private void sendAll(String msg) throws Throwable {
        ByteBuffer bb = ByteBuffer.wrap(msg.getBytes());
        bb.position(bb.limit());
        sendAll(bb);
    }
 
    private void sendAll(ByteBuffer bb) throws Throwable {
        bb.flip();
        for(SocketChannel sc : clients){
            sc.write(bb);
            bb.rewind();
        }
    }
 
    public static void main(String[] args) throws Throwable{
        NioServ s = new NioServ(Integer.parseInt(args[0]));
        s.loop();
    }
 
}

Zadania

Zadanie 4. Pobierz kod prostego serwera key-value store: simplekv.java. (https://en.wikipedia.org/wiki/Key-value_database)
Przeczytaj kod, połącz się programem netcat / socat i przetestuj działanie.
Następnie dodaj do programu brakujące:

komunikaty diagnostyczne – pojawił się nowy klient, klient się rozłączył, zmieniono wartość
obsługę błędów – tzn. usuń throws Throwable i dodaj kod łapiący i obsługujący wyjątki
buforowanie wejścia – obsłuż sytuację, w której read zwróci tylko część linii
Testuj używając jako klienta:
''socat tcp:host:port,ignoreeof - ''
Wpisanie EOF (ctrl+d) spowoduje wysłanie dotychczas wpisanych danych
buforowanie wyjścia – obsłuż sytuację, w której write zapisze tylko część żądanych danych

¹⁾

oraz znacznik, mark

Jan Kończak

Table of Contents

Obsługa sieci w Javie

java.io

Klient TCP

Serwer TCP

UDP

Klient

Serwer

Multicast

Zadania

java.nio

Klasy obsługujące gniazda (kanały)

Praca z selektorem

Bufory

SelectionKey

"klient tcp"

Serwer czatu

Zadania