试图实现一个quickSend()

This commit is contained in:
free will committed 2021-10-25 23:38:09 +08:00
1 parent e71261132d
commit e231aa8661
3 files changed
+129 -2

No files matched your search

+79
View File
@@ -7,6 +7,7 @@ import packet.*;
import util.ByteHelper;
import java.util.Arrays;
import java.util.concurrent.ConcurrentLinkedQueue;
/*
* @Author: Wang Feng
@@ -25,6 +26,9 @@ public class LinkService {
public int lpPacketHeadSize; // lpPacket 编码成数组时的头部大小
public long lpPacketId; // 发送的lpPacket 编码
// 测试去掉一个encoder之后,是否发包变快
public byte[] cpacketBytes;
/**
* 计算lpPacket头部大小
*/
@@ -57,6 +61,37 @@ public class LinkService {
this.mtu=mtu;
this.calculateLpPacketHeadSize();
this.lpPacketId=0;
// todo:测试
Identifier serverIden = null;
try {
serverIden=new Identifier("/testsrc");
} catch (ComponentException e) {
e.printStackTrace();
return;
}
CPacket cPacket=new CPacket();
cPacket.setSrcIdentifier(serverIden);
cPacket.setDstIdentifier(serverIden);
cPacket.ttl.setTtl(64);
byte[] pay=new byte[1000];
cPacket.payload.setValue(pay);
ConcurrentLinkedQueue<CPacket> readTunCPackets = new ConcurrentLinkedQueue<>();
for (int i = 0; i < 100000; i++) {
readTunCPackets.offer(cPacket);
}
Encoder encoder=new Encoder();
try {
if(!encoder.encoderReset(new SizeT(Encoder.MaxPacketSize),new SizeT(0))){
}
} catch (EncoderException e) {
e.printStackTrace();
}
try {
int bufLen=cPacket.wireEncode(encoder);
} catch (PacketException e) {
e.printStackTrace();
}
cpacketBytes=encoder.getBuffer();
}
/**
@@ -206,6 +241,50 @@ public class LinkService {
}
}
/**
* 快发一个推式包:直接默认要发的cpacket只打包成一个lppacket,快速将它发出去
* @param cPacket
* @return
* @throws LogicFaceException
*/
public boolean sendQuickCPacket(CPacket cPacket) throws LogicFaceException {
try {
LpPacket lpPacket=getLpPacketFromCPacket(cPacket);
if(lpPacket!=null) {
return this.transport.send(lpPacket);
}else {
return false;
}
} catch (LogicFaceException e) {
throw new LogicFaceException("LinkService.sendCPacket: "+e.getMessage());
}
}
private LpPacket getLpPacketFromCPacket(CPacket cPacket) throws LogicFaceException {
try {
// // 1.CPacket => byte[]
// Encoder encoder=new Encoder();
// if(!encoder.encoderReset(new SizeT(Encoder.MaxPacketSize),new SizeT(0))){
// return null;
// }
// int bufLen=cPacket.wireEncode(encoder);
// byte[] buf=encoder.getBuffer();
// 2.byte[] => LpPacket
LpPacket lpPacket=new LpPacket();
lpPacket.setId(this.lpPacketId);
lpPacket.setFragmentNum(1); //分片数
lpPacket.setFragmentSeq(0); //片序号
lpPacket.setValue(cpacketBytes);
this.lpPacketId++;
return lpPacket;
// } catch (PacketException | EncoderException e) {
} catch (Exception e) {
throw new LogicFaceException("LinkService.getMINPacketFromLpPacket: "+e.getMessage());
}
}
/**
* 发送一个MIN网络包
* @param minPacket
+10
View File
@@ -302,6 +302,16 @@ public class LogicFace {
return this.linkService.sendCPacket(cPacket);
}
/**
* 快发包(暂时仅支持TCP):减少打包过程,迅速将包发送出去
* @param cPacket
* @return
* @throws LogicFaceException
*/
public boolean sendQuickCPacketByTCP(CPacket cPacket) throws LogicFaceException {
return this.linkService.sendQuickCPacket(cPacket);
}
/**
* 发送一个兴趣包
* @param interest
@@ -7,8 +7,11 @@ import examples.TestKeyManagerExample;
import logicface.LogicFace;
import logicface.LogicFaceException;
import org.junit.Test;
import packet.CPacket;
import security.KeyChain;
import java.util.concurrent.ConcurrentLinkedQueue;
/*
* @Author: Wang Feng
@@ -24,8 +27,7 @@ public class RegisterPrefixHelperTest {
Identifier identifier=new Identifier("/hello/wefree"+System.currentTimeMillis());
// 初始化face
LogicFace face=new LogicFace();
// face.initWithUdp("127.0.0.1",14922);
face.initWithUdp("14.215.134.202",14922);
face.initWithTcp("14.215.134.202",14922);
// 初始化KeyChain
// KeyChain keyChain=new KeyChain();
@@ -33,6 +35,42 @@ public class RegisterPrefixHelperTest {
KeyManagerExample.INSTANCE.initKeyChain("/wefree/test","D://");
face.setKeyChain(KeyManagerExample.INSTANCE.getKeyChain());
helper.registerPrefix(identifier,face,KeyManagerExample.INSTANCE.getKeyChain());
// 测试发包速率
Identifier serverIden = null;
try {
serverIden=new Identifier("/testsrc");
} catch (ComponentException e) {
e.printStackTrace();
return;
}
CPacket cPacket=new CPacket();
cPacket.setSrcIdentifier(serverIden);
cPacket.setDstIdentifier(serverIden);
cPacket.ttl.setTtl(64);
byte[] pay=new byte[1000];
cPacket.payload.setValue(pay);
ConcurrentLinkedQueue<CPacket> readTunCPackets = new ConcurrentLinkedQueue<>();
for (int i = 0; i < 100000; i++) {
readTunCPackets.offer(cPacket);
}
// 发包计数
int nums=100000;
int error=0;
long startTime=System.currentTimeMillis();
for (int i = 0; i < nums; i++) {
// 出队
CPacket newPacket = readTunCPackets.poll();
if(newPacket!=null) {
face.sendQuickCPacketByTCP(newPacket);
// face.sendCPacket(newPacket);
}
}
System.out.println("nums: "+nums);
long endtime=System.currentTimeMillis();
System.out.println("发包"+nums+"耗时: "+(endtime-startTime)+"毫秒");
System.out.println(readTunCPackets.size());
}
@Test