java - 通过 TCP 套接字的 Avro 通信

Question

我目前正在开发一个项目，该项目将涉及用 C 和 Java 编写的应用程序的通信。因此，我选择使用 Apache Avro。我在网站上看到 Avro 可以使用 DataFileWriter 类从文件中(反)序列化对象。

但是，就我而言，我想在应用程序之间使用 TCP 套接字。因此，DataFileWriter 类对我不起作用。阅读文档后，我没有找到任何有关如何通过 TCP 套接字发送对象的信息。

关于如何做到这一点有什么想法吗？我特别想知道我应该在 Java 客户端上使用哪种输入和输出流。

我为 Java 服务器开发了以下代码:

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.HashMap;

import middleman.bigpeer.BigPeer;

import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificDatumWriter;

public class MiddleManWorker implements Runnable {

    private InputStream in;

    private OutputStream out;

    private Socket clientSocket;

    public MiddleManWorker(Socket clientSocket, HashMap<Integer, NodeType> dbNodesDirectory, 
            HashMap<Integer, NodeType>  workersDirectory) {
        this.clientSocket = clientSocket;
        try {
            this.in = clientSocket.getInputStream();
            this.out = clientSocket.getOutputStream();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        EncoderFactory encoderFactory = new EncoderFactory();
        DecoderFactory decoderFactory = new DecoderFactory();
    BinaryEncoder binaryEncoder = encoderFactory.binaryEncoder(out, null);
        BinaryDecoder binaryDecoder = decoderFactory.binaryDecoder(in, null);
        SpecificDatumReader<BigPeer> peerDatumReader = new SpecificDatumReader<BigPeer>(BigPeer.class);
        BigPeer bigPeer = null;
        SpecificDatumWriter<BigPeer> writer = new SpecificDatumWriter<BigPeer>();
        try {
            peerDatumReader.read(bigPeer, binaryDecoder);
            System.out.println("Received: " + bigPeer.getType());
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            writer.write(bigPeer, binaryEncoder);
        } catch (IOException e) {
            e.printStackTrace();
        }


    }

} 

示例 Java 客户端如下:

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.Socket;

import middleman.bigpeer.BigPeer;

import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificDatumWriter;

public class SystemClient {

    public static void connect(String serverIPAddress, Integer serverPort) throws IOException, ClassNotFoundException {
        /**
         * Create Connection with the server
         */
        Socket socket = new Socket(serverIPAddress, serverPort);
        InputStream in = socket.getInputStream();
        OutputStream out = socket.getOutputStream();
        EncoderFactory encoderFactory = new EncoderFactory();
        DecoderFactory decoderFactory = new DecoderFactory();
        BinaryEncoder binaryEncoder = encoderFactory.binaryEncoder(out, null);
        BinaryDecoder binaryDecoder = decoderFactory.binaryDecoder(in, null);

        BigPeer bigPeer = new BigPeer();
        bigPeer.setType("test");
        SpecificDatumReader<BigPeer> reader = new SpecificDatumReader<BigPeer>(BigPeer.class);
        SpecificDatumWriter<BigPeer> writer = new SpecificDatumWriter<BigPeer>(BigPeer.class);

        System.out.println("Before: " +  bigPeer.getType());
        writer.write(bigPeer, binaryEncoder);
        System.out.println("Waiting for response...");
        reader.read(bigPeer, binaryDecoder);
        System.out.println("After: " + bigPeer.getType());
    }

}

服务器似乎停止在 peerDatumReader.read(bigPeer, binaryDecoder); 代码行上。有什么想法吗？

谢谢你，尼克

Answer 1

出于性能原因，BinaryEncoder 使用内部缓冲区。您可能需要在编码器上调用flush来通过管道发送数据。

请参阅reference有关此行为的更多信息:

The BinaryEncoder implementation returned may buffer its output. Data may not appear on the underlying OutputStream until Flushable.flush() is called. The buffer size is configured with configureBufferSize(int).

If buffering is not desired, and lower performance is acceptable, use directBinaryEncoder(OutputStream, BinaryEncoder)