Java NIO mrežno programiranje u praksi - soba za ćaskanje: od početnika do majstora
Ranije smo koristili Java sokete Socket i ServerSocket za mrežno programiranje, ali Socket i ServerSocket se zasnivaju na Java IO, pa u mrežnom programiranju imaju lošije performanse. Razlog smo već objasnili ranije.
Kako su onda na performansama SocketChannel i ServerSocketChannel iz Java NIO-a?
SocketChannel i ServerSocketChannel
U prvom času učenja NIO-a već smo predstavili SocketChannel i ServerSocketChannel, ovde ćemo samo kratko dopuniti.
ServerSocketChannel služi za kreiranje soketa na strani servera, dok SocketChannel služi za kreiranje soketa na strani klijenta. Oba podržavaju blokirajući i neblokirajući režim, koji se prebacuje postavljanjem svojstva blocking. U blokirajućem režimu, operacije čitanja/pisanja blokiraće se dok se ne završe, dok će u neblokirajućem režimu operacije čitanja/pisanja odmah vratiti.
Blokirajući režim:
- Prednosti: jednostavno programiranje, pogodno za scenarije sa niskom istovremenošću.
- Mane: lošije performanse, nije pogodno za scenarije sa visokom istovremenošću.
Neblokirajući režim:
- Prednosti: bolje performanse, pogodno za scenarije sa visokom istovremenošću.
- Mane: programiranje je relativno složenije.
Pogledajmo jednostavan primer (u blokirajućem režimu):
Prvo pogledajmo stranu Servera:
public class BlockingServer {
public static void main(String[] args) throws IOException {
// Kreiranje serverskog soketa
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// Povezivanje sa portom
serverSocketChannel.socket().bind(new InetSocketAddress(8080));
// Postavljanje na blokirajući režim (podrazumevano je blokirajući režim)
serverSocketChannel.configureBlocking(true);
while (true) {
// Prihvatanje veze klijenta
SocketChannel socketChannel = serverSocketChannel.accept();
// Dodela bafera
ByteBuffer buffer = ByteBuffer.allocate(1024);
// Čitanje podataka
int bytesRead = socketChannel.read(buffer);
while (bytesRead != -1) {
buffer.flip();
System.out.println(StandardCharsets.UTF_8.decode(buffer));
buffer.clear();
bytesRead = socketChannel.read(buffer);
}
// Zatvaranje soketa
socketChannel.close();
}
}
}Kratko objašnjenje ovog koda, prilično je lako razumeti.
Prvo se kreira serverski soket ServerSocketChannel, zatim se povezuje sa portom 8080, a onda se u while petlji osluškuje na klijentski soket. Ako se primi veza klijenta SocketChannel, čitaju se podaci iz kanala u bafer ByteBuffer, dok god u kanalu ima podataka, a zatim se zatvara taj kanal.
Pri tome serverSocketChannel.configureBlocking(true) postavlja kanal u blokirajući režim (može se izostaviti).
Sada pogledajmo klijentsku stranu:
public class BlockingClient {
public static void main(String[] args) throws IOException {
// Kreiranje klijentskog soketa
SocketChannel socketChannel = SocketChannel.open();
// Povezivanje sa serverom
socketChannel.connect(new InetSocketAddress("localhost", 8080));
// Dodela bafera
ByteBuffer buffer = ByteBuffer.allocate(1024);
// Slanje podataka serveru
buffer.put("Chenmo Wang Er, ovo je poruka sa klijentske strane.".getBytes(StandardCharsets.UTF_8));
buffer.flip();
socketChannel.write(buffer);
// Pražnjenje bafera
buffer.clear();
// Zatvaranje soketa
socketChannel.close();
}
}Kôd klijenta je još jednostavniji: uspostavlja se kanal SocketChannel, povezuje se sa serverom, zatim se u bafer stavi jedan podatak, upisuje se u kanal i zatvara soket.
Prvo pokrenite BlockingServer, a zatim BlockingClient — u konzoli na strani Servera možete videti sledeće.
U redu, pogledajmo sada primer u neblokirajućem režimu.
Prvo strana Servera:
public class NonBlockingServer {
public static void main(String[] args) throws IOException {
// Kreiranje serverskog soketa
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// Povezivanje sa portom
serverSocketChannel.socket().bind(new InetSocketAddress(8080));
// Postavljanje na neblokirajući režim
serverSocketChannel.configureBlocking(false);
// Kreiranje selektora
Selector selector = Selector.open();
// Registracija serverskog soketa kod selektora
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
while (true) {
selector.select();
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectedKeys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
if (key.isAcceptable()) {
// Prihvatanje veze klijenta
SocketChannel socketChannel = serverSocketChannel.accept();
socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_READ);
}
if (key.isReadable()) {
// Čitanje podataka
SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
int bytesRead = socketChannel.read(buffer);
if (bytesRead != -1) {
buffer.flip();
System.out.print(StandardCharsets.UTF_8.decode(buffer));
buffer.clear();
} else {
// Klijent je prekinuo vezu, otkazati ključ selekcije i zatvoriti kanal
key.cancel();
socketChannel.close();
}
}
}
}
}
}Što je isto kao u prethodnom blokirajućem režimu nećemo ponavljati, navešćemo samo razlike.
①,Prvo, kreira se ServerSocketChannel i postavlja se u neblokirajući režim.
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);②,Kreira se instanca Selector, za obradu događaja na više kanala.
Selector selector = Selector.open();③,ServerSocketChannel se registruje kod Selector-a i kao događaj od interesa postavlja se OP_ACCEPT. To znači da će nas Selector obavestiti kada stigne novi zahtev klijenta za vezu.
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);Pogledajte komentar za OP_ACCEPT:
④,U petlji se obrađuju događaji u Selector-u. Prvo se poziva selector.select() da bi se sačekalo na događaj od interesa. Ova metoda blokira dok se ne dogodi barem jedan događaj od interesa.
while (true) {
int readyChannels = selector.select();
if (readyChannels == 0) {
continue;
}
// ...
}⑤,Kada se selector.select() vrati, preko selector.selectedKeys() možemo dobiti sve spremne događaje i iterativno ih obraditi. Pri obradi događaja se na osnovu tipa SelectionKey izvršava odgovarajuća operacija.
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
while (keyIterator.hasNext()) {
SelectionKey key = keyIterator.next();
// Obrada događaja
// ...
keyIterator.remove();
}⑥,Kada je tip SelectionKey jednaki OP_ACCEPT, to znači da postoji novi zahtev klijenta za vezom. Tada treba prihvatiti novu vezu i novonastali SocketChannel postaviti u neblokirajući režim. Zatim taj SocketChannel registrovati kod Selector-a i kao događaj od interesa postaviti OP_READ.
if (key.isAcceptable()) {
ServerSocketChannel server = (ServerSocketChannel) key.channel();
SocketChannel client = server.accept();
client.configureBlocking(false);
client.register(selector, SelectionKey.OP_READ);
}⑦,Kada je tip SelectionKey jednaki OP_READ, to znači da je klijent poslao podatke. Iz SocketChannel-a treba pročitati podatke i izvršiti odgovarajuću obradu.
if (key.isReadable()) {
SocketChannel client = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
int bytesRead = client.read(buffer);
// Obrada pročitanih podataka
// ...
}⑧,(ako je moguće) Kada je tip SelectionKey jednaki OP_WRITE, to znači da se klijentu mogu slati podaci. Podaci koje želimo da pošaljemo mogu se upisati u SocketChannel.
if (key.isWritable()) {
SocketChannel client = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.wrap("Zdravo, klijentu".getBytes());
client.write(buffer);
}Međutim, u ovom primeru nema ovog koraka. Ako zatreba, na ovaj način se mogu upisivati podaci klijentu.
⑨,Kada server prestane sa radom, treba zatvoriti Selector i ServerSocketChannel i osloboditi resurse.
key.cancel();
socketChannel.close();U redu, sad pogledajmo klijentsku stranu.
public class NonBlockingClient {
public static void main(String[] args) throws IOException {
// Kreiranje klijentskog soketa
SocketChannel socketChannel = SocketChannel.open();
// Postavljanje na neblokirajući režim
socketChannel.configureBlocking(false);
// Povezivanje sa serverom
socketChannel.connect(new InetSocketAddress("localhost", 8080));
while (!socketChannel.finishConnect()) {
// Čekanje da se veza uspostavi
}
// Dodela bafera
ByteBuffer buffer = ByteBuffer.allocate(1024);
// Slanje podataka serveru
String message = "Zdravo, Chenmo Wang Er, ovo je poruka sa klijentske strane.";
buffer.put(message.getBytes(StandardCharsets.UTF_8));
buffer.flip();
socketChannel.write(buffer);
// Pražnjenje bafera
buffer.clear();
// Zatvaranje soketa
socketChannel.close();
}
}Kôd klijenta je i dalje prilično jednostavan, pa ćemo ga preskočiti bez dodatnog objašnjenja. Zatim pokrenite Server pa Client. Može se pokrenuti više puta, rezultat je:
Scatter i Gather
Scatter i Gather su dve efikasne I/O operacije u Java NIO-u, koje služe za rasipanje podataka u više bafera ili prikupljanje podataka iz više bafera.
Scatter (rasipanje): podaci pročitani iz Channel-a rasipaju se (upisuju) u više bafera. Ova operacija omogućava da se prilikom čitanja podaci rasipaju u različite bafere, što pomaže u obradi strukturiranih podataka. Na primer, zaglavlje poruke, telo poruke i rep poruke mogu se upisati u različite bafere.
Gather (prikupljanje): suprotno od Scatter-a, prikuplja (čita) podatke iz više bafera i upisuje ih u jedan Channel. Ova operacija dozvoljava da prilikom slanja podataka prikupimo podatke iz više bafera. Na primer, zaglavlje poruke, telo poruke i rep poruke mogu se prikupiti iz različitih bafera i upisati u isti Channel.
Napišimo kompletan demo, prvo Server.
// Kreiranje ServerSocketChannel-a
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().bind(new InetSocketAddress(9000));
// Prihvatanje veze
SocketChannel socketChannel = serverSocketChannel.accept();
// Scatter: rasipanje pročitanih podataka u više bafera
ByteBuffer headerBuffer = ByteBuffer.allocate(128);
ByteBuffer bodyBuffer = ByteBuffer.allocate(1024);
ByteBuffer[] buffers = {headerBuffer, bodyBuffer};
long bytesRead = socketChannel.read(buffers);
// Ispis podataka iz bafera
headerBuffer.flip();
while (headerBuffer.hasRemaining()) {
System.out.print((char) headerBuffer.get());
}
System.out.println();
bodyBuffer.flip();
while (bodyBuffer.hasRemaining()) {
System.out.print((char) bodyBuffer.get());
}
// Gather: prikupljanje podataka iz više bafera i upis u Channel
ByteBuffer headerResponse = ByteBuffer.wrap("Header Response".getBytes());
ByteBuffer bodyResponse = ByteBuffer.wrap("Body Response".getBytes());
ByteBuffer[] responseBuffers = {headerResponse, bodyResponse};
long bytesWritten = socketChannel.write(responseBuffers);
// Zatvaranje veze
socketChannel.close();
serverSocketChannel.close();Sada Client:
// Kreiranje SocketChannel-a
SocketChannel socketChannel = SocketChannel.open();
socketChannel.connect(new InetSocketAddress("localhost", 9000));
// Slanje podataka serveru
String header = "Header Content";
String body = "Body Content";
ByteBuffer headerBuffer = ByteBuffer.wrap(header.getBytes());
ByteBuffer bodyBuffer = ByteBuffer.wrap(body.getBytes());
ByteBuffer[] buffers = {headerBuffer, bodyBuffer};
socketChannel.write(buffers);
// Prijem podataka sa servera
ByteBuffer headerResponseBuffer = ByteBuffer.allocate(128);
ByteBuffer bodyResponseBuffer = ByteBuffer.allocate(1024);
ByteBuffer[] responseBuffers = {headerResponseBuffer, bodyResponseBuffer};
long bytesRead = socketChannel.read(responseBuffers);
// Ispis primljenih podataka
headerResponseBuffer.flip();
while (headerResponseBuffer.hasRemaining()) {
System.out.print((char) headerResponseBuffer.get());
}
bodyResponseBuffer.flip();
while (bodyResponseBuffer.hasRemaining()) {
System.out.print((char) bodyResponseBuffer.get());
}
// Zatvaranje veze
socketChannel.close();U ovom primeru smo koristili Scattering da iz SocketChannel-a rasipamo pročitane podatke u više bafera, i Gathering da iz više bafera prikupimo podatke i upišemo ih u SocketChannel. Na ovaj način možemo lako rukovati podacima u više bafera.
Asinhroni soket kanali AsynchronousSocketChannel i AsynchronousServerSocketChannel
AsynchronousSocketChannel i AsynchronousServerSocketChannel su klase za asinhroni I/O uvedene u Javi 7, koje služe za obradu asinhronog klijentskog Socket-a odnosno serverskog ServerSocket-a. Asinhroni I/O omogućava da se za vreme I/O operacije izvršavaju drugi zadaci, a da se po završetku operacije dobije obaveštenje, čime se povećava moć istovremene obrade.
Pogledajmo jednostavan primer, prvo serverska strana.
public class AsynchronousServer {
public static void main(String[] args) throws IOException, InterruptedException {
AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel.open();
server.bind(new InetSocketAddress("localhost", 5000));
System.out.println("Server je pokrenut");
server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>() {
@Override
public void completed(AsynchronousSocketChannel client, Void attachment) {
// Prihvatanje sledećeg zahteva za vezom
server.accept(null, this);
ByteBuffer buffer = ByteBuffer.allocate(1024);
Future<Integer> readResult = client.read(buffer);
try {
readResult.get();
buffer.flip();
String message = new String(buffer.array(), 0, buffer.remaining());
System.out.println("Primljena poruka: " + message);
} catch (Exception e) {
e.printStackTrace();
}
}
@Override
public void failed(Throwable exc, Void attachment) {
exc.printStackTrace();
}
});
// Da bi server nastavio da radi, moramo sprečiti izlazak main niti
Thread.currentThread().join();
}
}Struktura koda je prilično slična asinhronom kanalu datoteka AsynchronousFileChannel o kom smo ranije govorili: asinhroni serverski soket kanal AsynchronousServerSocketChannel prihvata veze klijenata, a svaki put kada stigne nova veza poziva se completed(), zatim se čitaju podaci koje je poslao klijent i ispisuju u konzolu.
Hajde da ukratko analiziramo.
①,Kreira se i otvara instanca AsynchronousServerSocketChannel. Ovaj kanal će se koristiti za osluškivanje na veze klijenata.
AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel.open();②,Poziva se accept() za prihvatanje veze klijenta. Ovoj metodi je potrebna instanca CompletionHandler; kada klijent uspešno uspostavi vezu, biće pozvan completed().
server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>() { ... });③,Implementira se CompletionHandler: kada I/O operacija uspe, poziva se completed(); kada I/O operacija ne uspe, poziva se failed().
new CompletionHandler<AsynchronousSocketChannel, Void>() {
@Override
public void completed(AsynchronousSocketChannel client, Void attachment) { ... }
@Override
public void failed(Throwable exc, Void attachment) { ... }
}U completed metodi prvo pozivamo server.accept() da prihvatimo sledeći zahtev za vezom. Zatim kreiramo bafer ByteBuffer i pomoću client.read() čitamo podatke od klijenta. U ovom primeru koristimo objekat Future da sačekamo završetak operacije čitanja. Kada se čitanje završi, sadržaj bafera ispisujemo u konzolu.
④,Da bi server nastavio da radi i prima veze klijenata, moramo sprečiti izlazak main niti.
Thread.currentThread().join();Sada pogledajmo klijentsku stranu:
public class AsynchronousClient {
public static void main(String[] args) {
try {
AsynchronousSocketChannel client = AsynchronousSocketChannel.open();
Future<Void> connectResult = client.connect(new InetSocketAddress("localhost", 5000));
connectResult.get(); // Čekanje da se veza uspostavi
String message = "Chenmo Wang Er, jesi li tu?";
ByteBuffer buffer = ByteBuffer.wrap(message.getBytes(StandardCharsets.UTF_8));
Future<Integer> writeResult = client.write(buffer);
writeResult.get(); // Čekanje da se slanje završi
System.out.println("Poruka je poslata");
client.close();
} catch (IOException | InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
}To je prosto povezivanje i upisivanje podataka, pa nema potrebe za dodatnim objašnjenjem. Prvo pokrenite serversku stranu, a zatim klijentsku, da vidite rezultat.
Jednostavna soba za ćaskanje
Pomoću SocketChannel-a i ServerSocketChannel-a implementiraćemo sobu za ćaskanje verzije 0.1. Prvo da kažemo šta se traži, prilično je jednostavno: server se pokreće i osluškuje na zahteve klijenata, a kada klijent pošalje poruku serveru, server je primi i odbaci nazad klijentu (echo). Prilično je glupavo, ali hajde prvo da pogledamo.
Pogledajmo kôd serverske strane:
public class ChatServer {
private Selector selector;
private ServerSocketChannel serverSocketChannel;
private static final int PORT = 8080;
public ChatServer() {
try {
selector = Selector.open();
serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().bind(new InetSocketAddress(PORT));
serverSocketChannel.configureBlocking(false);
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("Server sobe za ćaskanje je pokrenut na portu " + PORT);
} catch (IOException e) {
e.printStackTrace();
}
}
public void start() {
try {
while (true) {
if (selector.select() > 0) {
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
handleKey(key);
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
private void handleKey(SelectionKey key) throws IOException {
if (key.isAcceptable()) {
SocketChannel socketChannel = serverSocketChannel.accept();
socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_READ);
System.out.println("Klijent se povezao: " + socketChannel.getRemoteAddress());
} else if (key.isReadable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
int read = socketChannel.read(buffer);
if (read > 0) {
buffer.flip();
String msg = new String(buffer.array(), 0, read);
System.out.println("Klijent kaže: " + msg);
socketChannel.write(ByteBuffer.wrap(("Odgovor servera: " + msg).getBytes()));
}
}
}
public static void main(String[] args) {
new ChatServer().start();
}
}Objašnjenje logike koda:
1,Kreira se ServerSocketChannel i povezuje se sa zadatim portom.
2,ServerSocketChannel se postavlja u neblokirajući režim.
3,Kreira se Selector i ServerSocketChannel se registruje kod njega, osluškuje na događaj OP_ACCEPT (čeka se veza klijenta).
4,Beskonačna petlja, čeka se da se dogodi događaj od interesa.
5,Koristi se Selector.select() da bi se sačekalo na događaj u registrovanim kanalima.
6,Uzima se SelectionKey kanala na kom se dogodio događaj.
7,Određuje se tip događaja SelectionKey-ja:
- a. Ako je događaj OP_ACCEPT, to znači da je uspostavljena nova veza sa klijentom. Prihvata se nova veza, a SocketChannel nove veze se registruje kod Selector-a i osluškuje na događaj OP_READ.
- b. Ako je događaj OP_READ, to znači da je klijent poslao poruku. Čita se poruka koju je poslao klijent i vraća mu se nazad. Nakon obrade se uklanja obrađeni SelectionKey.
Sada pogledajmo kôd klijentske strane:
public class ChatClient {
private Selector selector;
private SocketChannel socketChannel;
private static final String HOST = "localhost";
private static final int PORT = 8080;
public ChatClient() {
try {
selector = Selector.open();
socketChannel = SocketChannel.open(new InetSocketAddress(HOST, PORT));
socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_READ);
System.out.println("Povezan sa sobom za ćaskanje");
} catch (IOException e) {
e.printStackTrace();
}
}
public void start() {
new Thread(() -> {
try {
while (true) {
if (selector.select() > 0) {
for (SelectionKey key : selector.selectedKeys()) {
selector.selectedKeys().remove(key);
if (key.isReadable()) {
readMessage();
}
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}).start();
try (BufferedReader reader = new BufferedReader(new InputStreamReader(System.in
))) {
String input;
while ((input = reader.readLine()) != null) {
sendMessage(input);
}
} catch (IOException e) {
e.printStackTrace();
}
}
private void sendMessage(String message) throws IOException {
if (message != null && !message.trim().isEmpty()) {
ByteBuffer buffer = ByteBuffer.wrap(message.getBytes());
socketChannel.write(buffer);
}
}
private void readMessage() throws IOException {
ByteBuffer buffer = ByteBuffer.allocate(1024);
int read = socketChannel.read(buffer);
if (read > 0) {
buffer.flip();
String msg = new String(buffer.array(), 0, read);
System.out.println(msg);
}
}
public static void main(String[] args) {
new ChatClient().start();
}
}Objašnjenje logike koda:
1,Kreira se SocketChannel i povezuje se sa zadatom adresom i portom servera.
2,SocketChannel se postavlja u neblokirajući režim.
3,Kreira se Selector i SocketChannel se registruje kod njega, osluškuje na događaj OP_READ (čeka se prijem poruke sa servera).
4,Pokreće se nova nit koja čita poruke koje korisnik unese u konzoli i šalje ih serveru.
5,Beskonačna petlja, čeka se da se dogodi događaj od interesa.
6,Koristi se Selector.select() da bi se sačekalo na događaj u registrovanim kanalima.
7,Uzima se SelectionKey kanala na kom se dogodio događaj.
8,Određuje se tip događaja SelectionKey-ja:
- a. Ako je događaj OP_READ, to znači da je server poslao poruku. Čita se poruka servera i prikazuje u konzoli. Nakon obrade se uklanja obrađeni SelectionKey.
Pogledajmo efekat nakon pokretanja.
U redu, sad da malo nadogradimo zahteve, odnosno sobu za ćaskanje verzije 0.2, gde se zahteva da i serverska strana može unositi poruke iz konzole i aktivno ih slati klijentu.
Pogledajmo kôd serverske strane:
public class Chat2Server {
public static void main(String[] args) throws IOException {
// Kreiranje ServerSocketChannel-a
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.bind(new InetSocketAddress(8080));
// Kreiranje Selector-a
Selector selector = Selector.open();
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("Server sobe za ćaskanje je pokrenut");
// Veza klijenta
AtomicReference<SocketChannel> clientRef = new AtomicReference<>();
// Čitanje unosa iz konzole i slanje klijentu
Thread sendMessageThread = new Thread(() -> {
try (BufferedReader reader = new BufferedReader(new InputStreamReader(System.in))) {
while (true) {
System.out.println("Unesite poruku servera: ");
String message = reader.readLine();
SocketChannel client = clientRef.get();
if (client != null && client.isConnected()) {
ByteBuffer buffer = ByteBuffer.wrap((message + "\n").getBytes());
client.write(buffer);
}
}
} catch (IOException e) {
e.printStackTrace();
}
});
sendMessageThread.start();
while (true) {
int readyChannels = selector.select();
if (readyChannels == 0) {
continue;
}
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
while (keyIterator.hasNext()) {
SelectionKey key = keyIterator.next();
if (key.isAcceptable()) {
// Prihvatanje veze klijenta
SocketChannel client = serverSocketChannel.accept();
System.out.println("Klijent je povezan");
client.configureBlocking(false);
client.register(selector, SelectionKey.OP_READ);
clientRef.set(client);
} else if (key.isReadable()) {
// Čitanje poruke klijenta
SocketChannel channel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
int bytesRead = channel.read(buffer);
if (bytesRead > 0) {
buffer.flip();
byte[] bytes = new byte[buffer.remaining()];
buffer.get(bytes);
String message = new String(bytes).trim();
System.out.println("Poruka klijenta: " + message);
}
}
keyIterator.remove();
}
}
}
}Sada kôd klijenta:
public class Chat2Client {
public static void main(String[] args) throws IOException {
// Kreiranje SocketChannel-a
SocketChannel socketChannel = SocketChannel.open();
socketChannel.configureBlocking(false);
socketChannel.connect(new InetSocketAddress("localhost", 8080));
// Kreiranje Selector-a
Selector selector = Selector.open();
socketChannel.register(selector, SelectionKey.OP_CONNECT);
// Čitanje unosa iz konzole i slanje serveru
Thread sendMessageThread = new Thread(() -> {
try (BufferedReader reader = new BufferedReader(new InputStreamReader(System.in))) {
while (true) {
System.out.println("Unesite poruku klijenta: ");
String message = reader.readLine();
if (socketChannel.isConnected()) {
ByteBuffer buffer = ByteBuffer.wrap((message + "\n").getBytes());
socketChannel.write(buffer);
}
}
} catch (IOException e) {
e.printStackTrace();
}
});
sendMessageThread.start();
while (true) {
int readyChannels = selector.select();
if (readyChannels == 0) {
continue;
}
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
while (keyIterator.hasNext()) {
SelectionKey key = keyIterator.next();
if (key.isConnectable()) {
// Povezivanje sa serverom
socketChannel.finishConnect();
socketChannel.register(selector, SelectionKey.OP_READ);
System.out.println("Povezano sa serverom");
} else if (key.isReadable()) {
// Čitanje poruke sa servera
ByteBuffer buffer = ByteBuffer.allocate(1024);
int bytesRead = socketChannel.read(buffer);
if (bytesRead > 0) {
buffer.flip();
byte[] bytes = new byte[buffer.remaining()];
buffer.get(bytes);
String message = new String(bytes).trim();
System.out.println("Poruka servera: " + message);
}
}
keyIterator.remove();
}
}
}
}Pokrenite Server pa Client, razmena poruka izgleda ovako:
Koristili smo Selector i neblokirajući I/O, što omogućava serveru da istovremeno opslužuje više veza. Zato u Intellij IDEA-i možemo konfigurisati još jednog klijenta, kao na slici (popunite ova četiri polja).
Zatim ga pokrenite i možete ostvariti interakciju između jednog Servera i više Clienta.
OK, o sobi za ćaskanje ćemo tu stati za sada.
Kratak pregled
Ranije smo videli da se performanse Java NIO-a kod datotečnog I/O-a zaista ne razlikuju mnogo od tradicionalnog I/O-a, pa čak i zaostaju pri obradi velikih datoteka, ali glavna uloga NIO-a dolazi do izražaja u mrežnom I/O-u — na primer, ispod haube okvira Netty zapravo je NIO. Hajde da ukratko sumiramo.
SocketChannel (za TCP veze) i ServerSocketChannel (za osluškivanje i prihvatanje novih TCP veza) mogu zameniti tradicionalne klase Socket i ServerSocket i pružaju neblokirajući režim.
NIO podržava i blokirajući i neblokirajući režim. Neblokirajući režim omogućava programu da dok čeka na I/O izvršava druge zadatke, čime se poboljšavaju performanse istovremenosti. Implementacija neblokirajućeg režima se oslanja na Selector, koji može nadgledati I/O događaje na više kanala.
NIO podržava rasipanje podataka u više Buffer-a (Scatter) ili prikupljanje podataka iz više Buffer-a (Gather), čime se pruža efikasniji način prenosa podataka.
Java NIO.2 je uvela AsynchronousSocketChannel i AsynchronousServerSocketChannel; ove klase pružaju asinhroni I/O zasnovan na povratnim pozivima. Asinhroni soket kanali mogu automatski okinuti funkciju povratnog poziva kada završe I/O operaciju, čime se postiže efikasna asinhrona obrada.
Na kraju, pomoću NIO-a implementirali smo jednostavnu funkciju sobe za ćaskanje. Preko ServerSocketChannel-a i SocketChannel-a kreiraju se serverska i klijentska strana i omogućava se uzajamno slanje i prijem poruka. Pri obradi više klijenata može se koristiti Selector za upravljanje vezama više klijenata, čime se poboljšavaju performanse istovremenosti.
Ukratko, mrežno programiranje u Java NIO-u pruža efikasniji, fleksibilniji i skalabilniji način obrade I/O-a, sa značajnom prednošću za velike aplikacije i scenarije sa visokom istovremenošću.