我只想说我是套接字序列化的新手,在投票之前请建议我可以添加什么来编辑。我试图将代码分解得尽可能小,因为项目非常大。
我正在尝试创建一个非常简单的 RPC 类型中间件,其中客户端可以调用服务器上的方法并在该方法生成这样的对象时检索对象。我正在使用FST-Serializer库来序列化我的对象并通过网络/套接字发送它们。
虽然我已经启动并运行它,但我发现了一个非常奇怪的问题,当这种情况发生时,序列化的性能会显着下降(下面的代码示例):
Client Serialize params/method name -> Server retrieves information and executes method.
这意味着网络速度慢或实现不能足够快地调用它,但是如果发生这种情况,它会显着加速:
Client Serialize params/method name -> Server retrieves information and executes method -> Server serializes an object sends it back to client.
虽然一开始这似乎没什么大不了的,但对于 void 方法,它可能会变得很烦人,因为我必须发回虚拟数据,这意味着我必须无缘无故地进行一次网络旅行。
我想也许这可能是缓冲区不够大,但我已经尝试过不同的缓冲区大小,但似乎没有任何解决办法,我还想澄清一下,应用程序仍然可以正常工作(即没有阻塞)只是性能时间受到影响。所以我的问题是什么会导致速度变慢,它是否可以预防/修复?
我运行 YouKit 性能来查看热点分析(这也是新的),似乎 BufferedReader.Read 方法减慢了很多。
监听传入调用的 ServerThread 片段(客户端代码类似创建具有相同缓冲区大小的输入/输出):
public ServerThread(Socket connection){
this.connection = connection;
isActive = true;
}
@Override
public void run() {
try {
input = new BufferedInputStream(connection.getInputStream(), BUFFER_SIZE);
output = new BufferedOutputStream(connection.getOutputStream(), BUFFER_SIZE); //Buffersize is 512_000 atm
do{
Method m = (Method) readObject(input);
//Invoke method via reflection
Server.methodNames.get(m.getName()).invoke(this, m.getParams());
}while(isActive);
connection.close();
} catch (Exception e) {
e.printStackTrace();
}
}
@Override
public void transactionWillEnd() {
this.currentTransactionLog = null;
// This sends dummy data back to client or slowdown will occur
try{
writeObject(output, "SUCCESS");
output.flush();
} catch(Exception e){
e.printStackTrace();
}
}
从客户端序列化发送到 Serverthread 的方法类
public class Method implements Serializable{
private String name;
private Object[] params;
public Method(String name, Object...params) {
this.name = name;
this.params = params;
}
public String getName() {
return name;
}
public Object[] getParams() {
return params;
}
}
基于 FST 中 TCPObjectSocket 的 TCPSerializer(由 serverthread&client 继承:
public class TCPSerializer {
private static FSTConfiguration config;
public static int BUFFER_SIZE = 512_000;
AtomicBoolean readLock = new AtomicBoolean(false);
AtomicBoolean writeLock = new AtomicBoolean(false);
public TCPSerializer() {
config = FSTConfiguration.createDefaultConfiguration();
}
public Object readObject(InputStream input) throws Exception{
try {
while ( !readLock.compareAndSet(false,true) );
return config.decodeFromStream(input);
} finally {
readLock.set(false);
}
}
public void writeObject(OutputStream output, Object toWrite) throws Exception{
try {
while ( !writeLock.compareAndSet(false,true) );
config.encodeToStream(output, toWrite);
} finally {
writeLock.set(false);
}
}
客户端如何调用方法的示例:
@Override
public void transactionWillEnd() {
String methodName = Helpers.getMethodName(0);
Method m = new Method(methodName);
try {
client.writeObject(client.getOutput(), m);
client.flush();
//Read dummy data before continuing.
String verify = (String) client.readObject(client.getInput());
if(!verify.equals("SUCCESS")) throw new Exception(verify);
} catch (Exception e) {
e.printStackTrace();
}
}