Java - ReadObject mit nio

In einem traditionellen blocking-thread-server, ich würde so etwas tun

class ServerSideThread {

    ObjectInputStream in;
    ObjectOutputStream out;
    Engine engine;

    public ServerSideThread(Socket socket, Engine engine) {
        in = new ObjectInputStream(socket.getInputStream());
        out = new ObjectOutputStream(socket.getOutputStream());
        this.engine = engine;
    }

    public void sendMessage(Message m) {
        out.writeObject(m);
    }

    public void run() {
        while(true) {
            Message m = (Message)in.readObject();
            engine.queueMessage(m,this); //give the engine a message with this as a callback
        }
    }
}

Nun, das Objekt kann erwartet werden, ziemlich groß sein. In meinem nio-Schleife, ich kann nicht einfach warten, bis das Objekt, durch zu kommen, alle meine anderen verbindungen (mit viel kleinerer workloads) wartet auf mich.

Wie bekomme ich nur mitgeteilt, dass eine Verbindung für das gesamte Objekt, bevor es erzählt meine nio-Kanal bereit?

InformationsquelleAutor corsiKa | 2011-05-02

7

Schreiben Sie das Objekt in einen ByteArrayOutputStream so dass Sie geben Sie die Länge, bevor ein Objekt gesendet. Auf der empfangenden Seite, Lesen Sie die Menge der Daten, die erforderlich sind, bevor Sie versuchen, Sie zu entschlüsseln.

Jedoch, werden Sie wahrscheinlich finden es viel einfacher und effizienter zu bedienen-blocking IO (anstatt NIO) mit der Object*Stream

Bearbeiten, so etwas wie dieses
```
public static void send(SocketChannel socket,  Serializable serializable) throws IOException {
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    for(int i=0;i<4;i++) baos.write(0);
    ObjectOutputStream oos = new ObjectOutputStream(baos);
    oos.writeObject(serializable);
    oos.close();
    final ByteBuffer wrap = ByteBuffer.wrap(baos.toByteArray());
    wrap.putInt(0, baos.size()-4);
    socket.write(wrap);
}

private final ByteBuffer lengthByteBuffer = ByteBuffer.wrap(new byte[4]);
private ByteBuffer dataByteBuffer = null;
private boolean readLength = true;

public Serializable recv(SocketChannel socket) throws IOException, ClassNotFoundException {
    if (readLength) {
        socket.read(lengthByteBuffer);
        if (lengthByteBuffer.remaining() == 0) {
            readLength = false;
            dataByteBuffer = ByteBuffer.allocate(lengthByteBuffer.getInt(0));
            lengthByteBuffer.clear();
        }
    } else {
        socket.read(dataByteBuffer);
        if (dataByteBuffer.remaining() == 0) {
            ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(dataByteBuffer.array()));
            final Serializable ret = (Serializable) ois.readObject();
            //clean up
            dataByteBuffer = null;
            readLength = true;
            return ret;
        }
    }
    return null;
}
```
- Die Blockierung von IO nicht skalieren. Da der stack-thread Größe ist konstant über die JVM, jede Verbindung thread die gleiche stack-Größe wie mein Motor-thread (was ist gehen zu müssen, Recht groß zu sein.) So auch nur 3000 verbindungen, dies bringt mich auf 384Mb nur für verbindungen. Das lässt nicht viel für die system -, heap -, und Datenbank-auf einem 1Gb-box. Die cpu Anforderungen sind klein genug, dass ich konnte problemlos mit über 10000 verbindungen, abgesehen von der Speichernutzung.
- Also wo ich das senden der Menge von Daten, Lesen Sie zunächst, wie würde ich wissen, wie viel Daten habe ich in eine Warteschlange gestellt, bis... ich es Lesen?
- Verwenden Sie eine new ObjectOutputStream(new ByteArrayOutputStream mit jedem Objekt (oder Gruppe von Objekten) auf diese Weise haben Sie keine Daten in die Warteschlange eingereiht. Senden Sie die Länge des byte-Arrays als int gefolgt von dem byte-array. Auf der Seite Lesen, sollten Sie Lesen, 4 byte, minimum, das Ihnen die Länge der den rest für die Daten für das Objekt.
- Ich verstehe, dass Teil. Sagen wir, ich lese 4 bytes aus und ich bekomme 1000. Also ich weiß, dass ich Lesen müssen, 1000 bytes, bevor ich in der Lage bin, dies zu Lesen-Objekt. Wie weiß ich, Wann kann ich Lesen, diese 1000 bytes, die ohne Blockierung?
- Vielleicht seine Zeit, um mehr Speicher. Ein server mit 4 GB kostet etwa $500 und ein server mit 16 GB kostet etwa $1000. 😉
- Ich habe es geschafft-hosting. Und, dass ist wirklich nicht der Punkt...
- Sie wissen nicht, Wann Sie die Daten Lesen können, ohne zu blockieren. Allerdings, wenn Sie haben Lesen Sie die ersten vier bytes, wissen Sie, der rest ist nicht weit hinter sich. Wenn Sie nicht warten möchten, können Sie Puffern die Daten, die Sie haben und ObjectInputStream die Daten, wenn Sie gelesen haben, die erwartete Anzahl von bytes.
- Kann nicht der Punkt, aber es ist nützlich, um die richtigen Werkzeuge für den job. 😉
- Vielleicht, aber es würde immer noch nicht skalieren indefinately. Wenn ich das hätte, viel mehr Speicher, ich würde auch mehr Rechenleistung und Bandbreite, die mir erlauben würde, eine Skalierung auf bis zu 100k user auf einem server mit nio, aber begrenzen würde mich über 30k mit standard blockieren.
- hier ist ein Gedanke. Wäre es sinnvoll, wenn ich lese diese vier bytes, verwenden Sie einen threadpool und Lesen Sie den rest des bytes gibt es? Ich hätte eine gleich bleibende Anzahl threads verwendet, so dass anstatt 1 pro Verbindung, die mit 99% von Ihnen blockieren?
- Also im Grunde bin ich wie "ooo ich habe vier bytes kommen... hey nächste thread im Einklang, nehmen Sie dieses-stream und Lesen Sie ein Objekt!" mit diesem Gedanken im Hinterkopf, bräuchte ich noch nicht einmal diese vier bytes, könnte ich senden Sie einfach ein einzelnes byte, das sagt "INCOMIIINNNGGG!!!!"
- mit NIO, führen Sie eine Lesen, und es sagt Ihnen, wie viele bytes gelesen wurden. Sie nehmen dann diese bytes, und setzen Sie Sie in einen separaten Puffer pflegen Sie für jede Verbindung, bis die erforderliche Anzahl von bytes, gesammelt wurden. Ich würde empfehlen, wickeln Sie diese Funktionalität in Ihre eigenen paar von input-und output-streams.
- Ich mag es. Ich hatte vergessen, dass. Ich habe es gerade gelesen. Dann kann ich einen ObjectInputStream auf der hand und Pumpen Sie die Objekt-bytes in und out erscheint ein Objekt. 🙂
- Ich würde nicht vorschlagen, Sie gehen über 10K verbindungen pro server auf verwalteten hardware. esp. das ist so Ressource begrenzt. 😉
- Siehe mein Beispiel-code zur Verfügung gestellt.
- Hinweis: Es sind zwei subtile Fallstricke sollte man beachten, wenn Sie mit new ObjectOutputStream() mit jedem Objekt über eine einzige ObjectOutput/InputStream - pair-Mädchen für die gesamte Sitzung: Erstens, Daten müssen nur einmal gesendet werden, wie die Header und verschiedene meta-Informationen über Arten, wird sich ärgern, für jeden senden. Zweite, gemeinsame Instanzen auf sender-Seite werden Unterschied verweist, wenn Sie empfangen auf mehrere sendet.
InformationsquelleAutor Peter Lawrey

Inspiriert durch den code oben habe ich einen (GoogleCode-Projekt)

Es enthält einen einfachen unit-test:

SeriServer server = new SeriServer(6001, nthreads);
final SeriClient client[] = new SeriClient[nclients];

//write the data with multiple threads to flood the server

for (int cnt = 0; cnt < nclients; cnt++) {
    final int counterVal = cnt;
    client[cnt] = new SeriClient("localhost", 6001);
    Thread t = new Thread(new Runnable() {
         public void run() {
             try {
                for (int cnt2 = 0; cnt2 < nsends; cnt2++) {
                   String msg = "[" + counterVal + "]";                       
                   client[counterVal].send(msg);
                 }
             } catch (IOException e) {
                 e.printStackTrace();
                 fail();
             }
         }
         });
    t.start();
 }

 HashMap<String, Integer> counts = new HashMap<String, Integer>();
   int nullCounts = 0;
   for (int cnt = 0; cnt < nsends * nclients;) {
       //read the data from a vector (that the server pool automatically fills
       SeriDataPackage data = server.read();  
       if (data == null) {
              nullCounts++;
              System.out.println("NULL");
              continue;
       }

       if (counts.containsKey(data.getObject())) {
              Integer c = counts.get(data.getObject());
              counts.put((String) data.getObject(), c + 1);
        } else {
              counts.put((String) data.getObject(), 1);
        }
        cnt++;
        System.out.println("Received: " + data.getObject());
   }

   //asserts the results
   Collection<Integer> values = counts.values();
   for (Integer value : values) {
        int ivalue = value;
        assertEquals(nsends, ivalue);
        System.out.println(value);
   }
   assertEquals(counts.size(), nclients);
   System.out.println(counts.size());
   System.out.println("Finishing");
   server.shutdown();

InformationsquelleAutor Feiteira

Schreibe einen Kommentar

Du musst angemeldet sein, um einen Kommentar abzugeben.