5. diel - Java server - Správca spojenia

V predchádzajúcom kvíze, Kvíz - Parametre servera, vlákna a Google Guice, sme si overili nadobudnuté skúsenosti z predchádzajúcich lekcií.

V minulej lekcii, Kvíz - Parametre servera, vlákna a Google Guice , sme si vytvorili základnú kostru vlákna nášho Java servera. Dnes vytvoríme triedu, ktorá sa bude starať o prichádzajúce spojenia od potenciálnych klientov.

Správca spojenie

Opäť začneme tým, že v balíčku core vytvoríme nový balík connection, do ktorého umiestnime triedy, ktoré budú súvisieť so správcom spojenie:

• IConnectionManager - rozhranie poskytujúce metódy správcu spojenia
• IConnectionManagerFactory - rozhranie továrne pre tvorbu správcu spojenia
• ConnectionManager - implementácia rozhrania IConnectionManager
• ConnectionManagerFactory - implementácia rozhrania IConnectionManagerFactory
• IClient - rozhranie poskytujúce metódy pripojeného klienta
• Client - implementácia rozhrania IClient

Teraz začneme plniť jednotlivé rozhrania metódami, začneme rozhraním IConnectionManager:

public interface IConnectionManager {
    void addClient(Socket socket) throws IOException;
    void onServerStart();
    void onServerStop();
}

Rozhranie obsahuje hlavné metódu addClient(), pomocou ktorej by sa mal pridať klient do zoznamu pripojených klientov a dve pomocné metódy onServerStart() a onServerStop(), ktoré sa zavolajú na začiatku (prípadne na konci) života servera.

Rozhranie IConnectionManagerFactory bude obsahovať jedinú metódu getConnectionManager(), pomocou ktorej sa bude tvoriť inštancie rozhrania IConnectionManager:

public interface IConnectionManagerFactory {
    IConnectionManager getConnectionManager(int maxClients, int waitingQueueSize);
}

Metóda prijíma dva parametre, maxClients a waitingQueueSize.

Rozhranie IClient bude predstavovať pripojeného klienta, takže bude obsahovať metódy pre komunikáciu s týmto klientom:

public interface IClient {
    void sendMessageAsync(Object message);
    void sendMessage(Object message) throws IOException;
    void close();
}

Máme tu dve metódy pre odoslanie správy, to preto, že jedna metóda bude blokujúce a druhá asynchrónne. Metódou close() sa bude ukončovať spojenie s klientom.

Implementujeme navrhnutá rozhranie

Client

Pustíme sa do implementácie rozhrania. Začneme triedou Client, pretože tá jediná nebude závislá na správcovi spojenie. Trieda bude implementovať dve rozhrania: IClient a Runnable:

public class Client implements IClient, Runnable {
}

Nadefinujeme si inštančné konštanty:

private final Socket socket;
private final ObjectOutputStream writer;

a jednu inštančné premennú:

private ConnectionClosedListener connectionClosedListener;

ktorej dátový typ vytvoríme o pár riadkov nižšie.

Konštruktor bude mať (zatiaľ) iba jeden parameter typu Socket:

Client(Socket socket) throws IOException {
    this.socket = socket;
    writer = new ObjectOutputStream(socket.getOutputStream());
}

V konstruktoru uložíme referenciu na socket do inštančný konštanty a inicializujeme konštantu writer ako nový ObjectOutputStream.

Ďalej naimplementujeme metódy, ktoré nám predpisujú rozhranie:

@Override
public void close() {
    try {
        socket.close();
    } catch (IOException e) {
        e.printStackTrace
    }
}

Metóda close() iba deleguje volanie tej istej metódy nad Socketom. Odoslanie správy asynchrónne implementujeme v budúcnosti:

@Override
public void sendMessageAsync(Object message) {
    // TODO odeslat zprávu asynchronně
}

Blokujúce verzia odoslanie správy vezme objekt a pošle ho klientovi:

@Override
public void sendMessage(Object message) throws IOException {
    writer.writeObject(message);
}

Výnimka môže vyskočiť v prípade, že bolo spojenie ukončené. Teraz si uveďme spúšťací metódu:

@Override
public void run() {
    try (ObjectInputStream reader = new ObjectInputStream(socket.getInputStream())) {
        Object received;
        while ((received = reader.readObject()) != null) {
            // TODO zpracovat přijatou zprávu
        }
    } catch (EOFException | SocketException e) {
        e.printStackTrace();
    } catch (IOException e) {
        e.printStackTrace();
    } catch (ClassNotFoundException e) {
        // Nikdy by nemělo nastat
        e.printStackTrace();
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        if (connectionClosedListener != null) {
            connectionClosedListener.onConnectionClosed();
        }
        close();
    }
}

Metóda run() vyzerá zložito, ale vlastne nerobí nič iné, než že prijíma správy od klienta. Veľkú časť kódu zaberajú výnimky. Poďme si ich vysvetliť:

• EOFException | SocketException - klient riadne ukončil spojenie
• IOException - nastala neočakávaná výnimka v komunikácii
• ClassNotFoundException - výnimka by nikdy nemala nastať, ak budeme dodržiavať komunikačný protokol, ktorý navrhneme v budúcnosti
• Exception - odchytenie všeobecné výnimky

V bloku finally informujeme poslucháča, že spojenie bolo ukončené a zavoláme metódu close() pre uzavretie socketu a uvoľnenie zdrojov.

Teraz vytvoríme v triede Client funkcionálne rozhranie ConnectionClosedListener, ktoré bude predstavovať poslucháča ukončenia spojenia:

@FunctionalInterface
public interface ConnectionClosedListener {
    void onConnectionClosed();
}

Rozhranie obsahuje jedinú metódu onConnectionClosed(), ktorú voláme v prípade, že sa ukončilo spojenie medzi klientom a serverom.

Nakoniec pridáme setter tohto poslucháča:

void setConnectionClosedListener(ConnectionClosedListener connectionClosedListener) {
    this.connectionClosedListener = connectionClosedListener;
}

Metóda bude viditeľná iba v balíčku, v ktorom sa trieda nachádza, a nikde inde. Nepotrebujeme, aby nám niekto iný nastavoval listener.

ConnectionManager

Trieda bude iba implementovať rozhranie IConnectionManager a opäť nemusí byť verejná:

class ConnectionManager implements IConnectionManager {
}

V triede sa bude musieť nachádzať kolekcia, ktorá bude obsahovať pripojených klientov:

private final List<IClient> clients = new ArrayList<>();

ďalej threapool pre jednotlivých klientov:

private final ExecutorService pool;

a konštanta reprezentujúci maximálny počet aktívne komunikujúcich klientov:

final int maxClients;

Konštruktor bude prijímať vyššie zmienené neinicializované premenné:

@Inject
public ConnectionManager(ExecutorService pool,int maxClients) {
    this.pool = pool;
    this.maxClients = maxClients;
}

Než začneme implementovať metódy, ktoré po nás vyžaduje rozhranie, vytvoríme si privátne metódu insertClientToListOrQueue(), pomocou ktorej sa budeme rozhodovať, či ak vložíme klienta do kolekcie aktívnych klientov, alebo do čakacej fronty:

private synchronized void insertClientToListOrQueue(Client client) {
    if (clients.size() < maxClients) {
        clients.add(client);
        client.setConnectionClosedListener(() -> {
            clients.remove(client);
        });
        pool.submit(client);
    } else {
        // TODO vložit klienta do čekací fronty
    }
}

Implementáciu vloženie klienta do čakacej fronty necháme na ďalšiu lekciu.

Teraz implementujeme metódy podľa rozhranie:

@Override
public void addClient(Socket socket) throws IOException {
    insertClientToListOrQueue(new Client(socket));
}

Metóda addClient() iba předeleguje volania na metódu insertClientToListOrQueue().

V metóde onServerStart() zatiaľ nebudeme nič robiť:

@Override
public void onServerStart() {}

Pri ukončení servera prejdeme všetkých klientov a ukončíme s nimi spojenie. Nakoniec ukončíme aj samotný threadpool:

@Override
public void onServerStop() {
    for (IClient client : clients) {
        client.close();
    }
    pool.shutdown();
}

Továreň správcu spojenia

Nakoniec nám zostáva implementovať rozhranie IConnectionManagerFactory:

@Singleton
public class ConnectionManagerFactory implements IConnectionManagerFactory {
    @Override
    public IConnectionManager getConnectionManager(int maxClients, int waitingQueueSize) {
        final ExecutorService pool = Executors.newFixedThreadPool(maxClients);
        return new ConnectionManager(pool, maxClients);
    }
}

V metóde vytvoríme threadpool o fixné veľkosti a vrátime novú inštanciu triedy ConnectionManager. Továreň opäť zaregistrujeme v triede ServerModule:

bind(IConnectionManagerFactory.class).to(ConnectionManagerFactory.class);

Úprava továrne vlákna servera

Pretože sme zmenili signatúru konstruktoru v triede ServerThread, musíme upraviť továreň tejto triedy. V triede ServerThreadFactory vytvoríme novú inštančný konštantu typu IConnectionManagerFactory, ktorú budeme inicializovať v konstruktoru, ktorý ju bude prijímať vo svojom parametri:

private final IConnectionManagerFactory connectionManagerFactory;
@Inject
public ServerThreadFactory(IConnectionManagerFactory connectionManagerFactory) {
    this.connectionManagerFactory = connectionManagerFactory;
}

Teraz už máme všetky predispozície pre správne vytvorenie novej inštancie triedy ServerThread:

return new ServerThread(connectionManagerFactory.getConnectionManager(maxClients, waitingQueueSize), port);

Použitia správcu spojenia

V triede ServerThread vytvoríme novú inštančný konštantu typu IConnectionManager. Ďalej pridáme do konstruktoru triedy rovnaký ukazovateľ, ktorú budeme inicializovať vyššie nadefinovanú konštantu. Teraz sa presunieme do metódy run(). Na samom začiatku metódy budeme volať metódu connectionManager.onServerStart(); aby sme dali možnosť správcovi spojenia vykonať inicializáciu (ktorú v budúcnosti napíšeme). Ďalej, keď prijmeme nového klienta pomocou metódy accept(), zavoláme opäť správcu spojenia, tentoraz metódou addClient() a odovzdáme jej prijatý socket. Na konci metódy run() budeme volať metódu connectionManager.onServerStop(), aby sme informovali správcu spojenie, že server sa má ukončiť, tak aby sa postaral o prípadnej pripojených klientov.

V budúcej lekcii, Java server - Client dispatcher , sa postaráme o klientov, ktoré bude potrebné presunú do čakacej fronty.

 

Mal si s čímkoľvek problém? Stiahni si vzorovú aplikáciu nižšie a porovnaj ju so svojím projektom, chybu tak ľahko nájdeš.