iMessageExecutorPool 关于线程模型 about java-kcp HOT 33 OPEN

    long start = System.currentTimeMillis();
    // Do the coding.
    LOOP.codeSomeShards(
            parityRows,
            shards, dataShardCount,
            outputs, parityShardCount,
            offset, byteCount);
    long end = System.currentTimeMillis();
    if ((end - start) > 0)
        System.out.println("codeSomeShards use time = " + (end - start));

   ChannelConfig channelConfig = new ChannelConfig();
    channelConfig.nodelay(true,40,2,true);
    channelConfig.setSndwnd(1024);
    channelConfig.setRcvwnd(1024);
    channelConfig.setMtu(1400);
    channelConfig.setFecDataShardCount(10);
    channelConfig.setFecParityShardCount(3);
    channelConfig.setAckNoDelay(true);
    channelConfig.setTimeoutMillis(10000);
    channelConfig.setUseConvChannel(true);
    channelConfig.setCrc32Check(false);

public class KcpRttExampleClient implements KcpListener {
private static final Logger log = LoggerFactory.getLogger(KcpRttExampleClient.class);
private final ByteBuf data;
private int[] rtts;
private volatile int count;
private ScheduledExecutorService scheduleSrv;
private ScheduledFuture<?> future = null;
private final long startTime ;

public KcpRttExampleClient() {
    data = Unpooled.buffer(200000);
    for (int i = 0; i < data.capacity(); i++) {
        data.writeByte((byte) i);
    }
    rtts = new int[30000];
    for (int i = 0; i < rtts.length; i++) {
        rtts[i] = -1;
    }
    startTime = System.currentTimeMillis();
    scheduleSrv = new ScheduledThreadPoolExecutor(1);
}

public static void main(String[] args) {
    ChannelConfig channelConfig = new ChannelConfig();
    channelConfig.nodelay(true,10,2,true);
    channelConfig.setSndwnd(1024);
    channelConfig.setRcvwnd(1024);
    channelConfig.setMtu(1400);
    channelConfig.setAckNoDelay(true);
    channelConfig.setConv(55);
    channelConfig.setFecDataShardCount(10);
    channelConfig.setFecParityShardCount(3);
    channelConfig.setCrc32Check(false);
    KcpClient kcpClient = new KcpClient();
    kcpClient.init(channelConfig);
    KcpRttExampleClient kcpClientRttExample = new KcpRttExampleClient();
    kcpClient.connect(new InetSocketAddress("127.0.0.1",20003),channelConfig,kcpClientRttExample);
}

@Override
public void onConnected(Ukcp ukcp) {
    future = scheduleSrv.scheduleWithFixedDelay(() -> {
        ByteBuf byteBuf = rttMsg(++count);
        ukcp.write(byteBuf);
        byteBuf.release();
        if (count >= rtts.length) {
            // finish
            future.cancel(true);
            byteBuf = rttMsg(-1);
            ukcp.write(byteBuf);
            byteBuf.release();

        }
    }, 5, 5, TimeUnit.MILLISECONDS);
}

@Override
public void handleReceive(ByteBuf byteBuf, Ukcp ukcp) {
    int curCount = byteBuf.readShort();
    if (curCount == -1) {
        scheduleSrv.schedule(new Runnable() {
            @Override
            public void run() {
                int sum = 0;
                for (int rtt : rtts) {
                    sum += rtt;
                }
                System.out.println("average: "+ (sum / rtts.length));
                System.out.println(Snmp.snmp.toString());
                ukcp.close();
                //ukcp.setTimeoutMillis(System.currentTimeMillis());
                System.exit(0);
            }
        }, 3, TimeUnit.SECONDS);
    } else {
        int idx = curCount - 1;
        long time = byteBuf.readInt();
        if (rtts[idx] != -1) {
            System.out.println("???");
        }
        //log.info("rcv count {} {}", curCount, System.currentTimeMillis());
        rtts[idx] = (int) (System.currentTimeMillis() - startTime - time);
        System.out.println("rtt : "+ curCount+"  "+ rtts[idx]);
    }
}

@Override
public void handleException(Throwable ex, Ukcp kcp)
{
    ex.printStackTrace();
}

@Override
public void handleClose(Ukcp kcp) {
    scheduleSrv.shutdown();
    try {
        scheduleSrv.awaitTermination(3, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

    int sum = 0;
    int max = 0;
    for (int rtt : rtts) {
        if(rtt>max){
            max = rtt;
        }
        sum += rtt;
    }
    System.out.println("average: "+ (sum / rtts.length)+" max:"+max);
    System.out.println(Snmp.snmp.toString());
    System.out.println("lost percent: "+(Snmp.snmp.RetransSegs.doubleValue()/Snmp.snmp.OutPkts.doubleValue()));


}


/**
 * count+timestamp+dataLen+data
 *
 * @param count
 * @return
 */
public ByteBuf rttMsg(int count) {
    ByteBuf buf = Unpooled.buffer(200000);
    buf.writeShort(count);
    buf.writeInt((int) (System.currentTimeMillis() - startTime));

    //int dataLen = new Random().nextInt(200);
    //buf.writeBytes(new byte[dataLen]);

    int dataLen = data.readableBytes();
    buf.writeShort(dataLen);
    buf.writeBytes(data, data.readerIndex(), dataLen);
    return buf;
}

private final ByteBuf data;

private int[] rtts;

private volatile int count;

private ScheduledExecutorService scheduleSrv;

private ScheduledFuture<?> future = null;

private final long startTime ;

public KcpRttExampleClient() {
    data = Unpooled.buffer(200000);
    for (int i = 0; i < data.capacity(); i++) {
        data.writeByte((byte) i);
    }

    rtts = new int[30000];
    for (int i = 0; i < rtts.length; i++) {
        rtts[i] = -1;
    }
    startTime = System.currentTimeMillis();
    scheduleSrv = new ScheduledThreadPoolExecutor(1);
}

public static void main(String[] args) {
    ChannelConfig channelConfig = new ChannelConfig();
    channelConfig.nodelay(true,40,2,true);
    channelConfig.setSndwnd(1024);
    channelConfig.setRcvwnd(1024);
    channelConfig.setMtu(1400);
    channelConfig.setAckNoDelay(true);
    channelConfig.setConv(55);

    channelConfig.setFecAdapt(new FecAdapt(10,3));
    channelConfig.setCrc32Check(true);
    //channelConfig.setTimeoutMillis(10000);
    //channelConfig.setAckMaskSize(32);
    KcpClient kcpClient = new KcpClient();
    kcpClient.init(channelConfig);

    KcpRttExampleClient kcpClientRttExample = new KcpRttExampleClient();
    kcpClient.connect(new InetSocketAddress("127.0.0.1",20003),channelConfig,kcpClientRttExample);

    //kcpClient.connect(new InetSocketAddress("10.60.100.191",20003),channelConfig,kcpClientRttExample);
}

@Override
public void onConnected(Ukcp ukcp) {
    future = scheduleSrv.scheduleWithFixedDelay(() -> {
        ByteBuf byteBuf = rttMsg(++count);
        ukcp.write(byteBuf);
        byteBuf.release();
        if (count >= rtts.length) {
            // finish
            future.cancel(true);
            byteBuf = rttMsg(-1);
            ukcp.write(byteBuf);
            byteBuf.release();

        }
    }, 20, 20, TimeUnit.MILLISECONDS);
}

@Override
public void handleReceive(ByteBuf byteBuf, Ukcp ukcp) {
    int curCount = byteBuf.readShort();

    if (curCount == -1) {
        scheduleSrv.schedule(new Runnable() {
            @Override
            public void run() {
                int sum = 0;
                for (int rtt : rtts) {
                    sum += rtt;
                }
                System.out.println("average: "+ (sum / rtts.length));
                System.out.println(Snmp.snmp.toString());
                ukcp.close();
                //ukcp.setTimeoutMillis(System.currentTimeMillis());
                System.exit(0);
            }
        }, 3, TimeUnit.SECONDS);
    } else {
        int idx = curCount - 1;
        long time = byteBuf.readInt();
        if (rtts[idx] != -1) {
            System.out.println("???");
        }
        //log.info("rcv count {} {}", curCount, System.currentTimeMillis());
        rtts[idx] = (int) (System.currentTimeMillis() - startTime - time);
        System.out.println("rtt : "+ curCount+"  "+ rtts[idx]);
    }
}

@Override
public void handleException(Throwable ex, Ukcp kcp)
{
    ex.printStackTrace();
}

@Override
public void handleClose(Ukcp kcp) {
    scheduleSrv.shutdown();
    try {
        scheduleSrv.awaitTermination(3, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

    int sum = 0;
    int max = 0;
    for (int rtt : rtts) {
        if(rtt>max){
            max = rtt;
        }
        sum += rtt;
    }
    System.out.println("average: "+ (sum / rtts.length)+" max:"+max);
    System.out.println(Snmp.snmp.toString());
    System.out.println("lost percent: "+(Snmp.snmp.RetransSegs.doubleValue()/Snmp.snmp.OutPkts.doubleValue()));


}


/**
 * count+timestamp+dataLen+data
 *
 * @param count
 * @return
 */
public ByteBuf rttMsg(int count) {
    ByteBuf buf = Unpooled.buffer(200000);
    buf.writeShort(count);
    buf.writeInt((int) (System.currentTimeMillis() - startTime));

    //int dataLen = new Random().nextInt(200);
    //buf.writeBytes(new byte[dataLen]);

    int dataLen = data.readableBytes();
    buf.writeShort(dataLen);
    buf.writeBytes(data, data.readerIndex(), dataLen);

    return buf;
}