减小体积
This commit is contained in:
@@ -7,7 +7,6 @@ import java.net.SocketException;
|
||||
|
||||
import org.kne.cloud.network.VirtualDatagramSocket;
|
||||
import org.kne.cloud.network.VirtualDatagramSocketImpl;
|
||||
import org.pcap4j.core.Inets;
|
||||
|
||||
public class KLALBVirtualDatagramSocket extends VirtualDatagramSocket {
|
||||
|
||||
|
||||
@@ -145,7 +145,7 @@ public class Kperf implements Runnable{
|
||||
|
||||
System.out.println("测试2/3:带宽测试");
|
||||
test.setLimitspeed(1000L*1024L*1024L*1024L);
|
||||
testPacket=new TESTNPacket(65535);
|
||||
testPacket=new TESTNPacket(9000);
|
||||
for (int i = 0; i < 60; i++) {
|
||||
Thread.sleep(1000);
|
||||
}
|
||||
|
||||
@@ -52,7 +52,6 @@ import org.kne.cloud.network.klalb.ui.PerformanceStrategyItem;
|
||||
import org.kne.concurrent.HighPerformanceExecutor2;
|
||||
import org.kne.concurrent.TimeoutConcurrentHashMap;
|
||||
import org.kne.io.KNEChannels;
|
||||
import org.pcap4j.packet.IllegalRawDataException;
|
||||
|
||||
import com.google.gson.internal.Pair;
|
||||
|
||||
@@ -483,7 +482,7 @@ public class SRv6Router {
|
||||
|
||||
// 负载均衡路由
|
||||
private boolean routingLoadBalance(IPv6NetworkLink linkfrom, IPv6Packet iPv6Packet,
|
||||
StringBuilder dbg, List<RouteItem> routes) throws IllegalRawDataException, IOException {
|
||||
StringBuilder dbg, List<RouteItem> routes) throws IOException {
|
||||
|
||||
if(routes==null||routes.isEmpty()) {
|
||||
//System.out.println("Route not found:" + iPv6Packet);
|
||||
@@ -512,7 +511,7 @@ public class SRv6Router {
|
||||
|
||||
private boolean loadBalance(IPv6NetworkLink linkfrom, IPv6Packet iPv6Packet, StringBuilder dbg,
|
||||
List<RouteItem> routes, double cscale)
|
||||
throws IllegalRawDataException, IOException {
|
||||
throws IOException {
|
||||
// 遍历可用路由进行负载均衡
|
||||
for (RouteItem tri : routes) {
|
||||
IPv6NetworkLink link = tri.getDestlink();
|
||||
@@ -546,7 +545,7 @@ public class SRv6Router {
|
||||
|
||||
// 处理数据包转发
|
||||
private void processPacket(IPv6NetworkLink linkfrom, IPv6Packet iPv6Packet, RouteItem ri)
|
||||
throws IllegalRawDataException, IOException {
|
||||
throws IOException {
|
||||
int hop = iPv6Packet.getHopLimit();
|
||||
|
||||
if (ri.getDestlink().isLoopBack()) { // 环回链路处理
|
||||
@@ -615,7 +614,7 @@ public class SRv6Router {
|
||||
}
|
||||
|
||||
// 发送ICMP TTL超时消息
|
||||
private void sendTTLExceedPacket(IPv6Packet iPv6Packet) throws IOException, IllegalRawDataException {
|
||||
private void sendTTLExceedPacket(IPv6Packet iPv6Packet) throws IOException {
|
||||
enqueuePacketSendTask(() -> {
|
||||
// 构建IPv6数据包封装ICMP消息
|
||||
IPv6Packet icmpv = new IPv6Packet();
|
||||
@@ -637,7 +636,7 @@ public class SRv6Router {
|
||||
// 处理SRv6数据包
|
||||
private void processSRv6Packet(IPv6NetworkLink linkfrom, IPv6Packet iPv6Packet, RouteItem ri,
|
||||
IPv6SegmentRoutingHeader srhh)
|
||||
throws IllegalRawDataException, IOException {
|
||||
throws IOException {
|
||||
|
||||
|
||||
if (srhh.getSegmentsLeft() <= 0) { // 所有段已处理完毕
|
||||
|
||||
@@ -1,114 +0,0 @@
|
||||
package org.kne.concurrent;
|
||||
|
||||
import java.util.concurrent.Executor;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ThreadFactory;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import com.lmax.disruptor.BlockingWaitStrategy;
|
||||
import com.lmax.disruptor.EventFactory;
|
||||
import com.lmax.disruptor.EventHandler;
|
||||
import com.lmax.disruptor.EventTranslator;
|
||||
import com.lmax.disruptor.LiteBlockingWaitStrategy;
|
||||
import com.lmax.disruptor.RingBuffer;
|
||||
import com.lmax.disruptor.Sequence;
|
||||
import com.lmax.disruptor.SequenceBarrier;
|
||||
import com.lmax.disruptor.SleepingWaitStrategy;
|
||||
import com.lmax.disruptor.YieldingWaitStrategy;
|
||||
import com.lmax.disruptor.dsl.Disruptor;
|
||||
import com.lmax.disruptor.dsl.ProducerType;
|
||||
|
||||
|
||||
public class DisruptorExecutor implements Executor{
|
||||
|
||||
public static final DisruptorExecutor defaultExecutor=new DisruptorExecutor();
|
||||
private RunnableHandler[]rh;
|
||||
|
||||
public DisruptorExecutor() {
|
||||
this(Runtime.getRuntime().availableProcessors(),1024);
|
||||
}
|
||||
public DisruptorExecutor(int threadcount,int bufferSize) {
|
||||
this(threadcount,bufferSize,Executors.defaultThreadFactory());
|
||||
//this(threadcount,bufferSize,Thread.ofVirtual().factory());
|
||||
}
|
||||
public DisruptorExecutor(int threadcount,int bufferSize,ThreadFactory th) {
|
||||
|
||||
|
||||
|
||||
rh=new RunnableHandler[threadcount];
|
||||
|
||||
for (int i = 0; i < rh.length; i++) {
|
||||
rh[i]=new RunnableHandler(bufferSize,th);
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
private class RunnableHandler implements EventHandler<AtomicReference<Runnable>>{
|
||||
|
||||
private Disruptor<AtomicReference<Runnable>>ring;
|
||||
public RunnableHandler(int bufferSize,ThreadFactory tf)
|
||||
{ring=new Disruptor<AtomicReference<Runnable>>(new EventFactory<AtomicReference<Runnable>>() {
|
||||
|
||||
@Override
|
||||
public AtomicReference<Runnable> newInstance() {
|
||||
return new AtomicReference<Runnable>();
|
||||
}
|
||||
}, bufferSize,tf,ProducerType.MULTI, new SleepingWaitStrategy());
|
||||
ring.handleEventsWith(this);
|
||||
ring.start();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onEvent(AtomicReference<Runnable> arg0, long sequence, boolean onEndOfBatch) throws Exception {
|
||||
//System.out.println(sequence);
|
||||
Runnable r =arg0.get();
|
||||
if(r!=null)
|
||||
try {
|
||||
r.run();
|
||||
}catch(Throwable th) {
|
||||
th.printStackTrace();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
/*
|
||||
* public final class MyHandler implements EventHandler<ValueEvent>
|
||||
{
|
||||
|
||||
public void onEvent(final ValueEvent entry, final long sequence, final boolean onEndOfBatch)
|
||||
{
|
||||
if ((sequence % numberOfConsumers) == ordinal)
|
||||
{
|
||||
// Process the event
|
||||
}
|
||||
}
|
||||
}
|
||||
*/
|
||||
private AtomicLong al=new AtomicLong();
|
||||
@Override
|
||||
public void execute(Runnable command) {
|
||||
int v=(int) (al.getAndIncrement()%rh.length);
|
||||
rh[v].ring.publishEvent(new EventTranslator<AtomicReference<Runnable>>() {
|
||||
|
||||
@Override
|
||||
public void translateTo(AtomicReference<Runnable> arg0, long arg1) {
|
||||
arg0.set(command);
|
||||
//arg0.compareAndSet(null, command);
|
||||
}
|
||||
});
|
||||
}
|
||||
public static void main(String[] args) throws InterruptedException {
|
||||
DisruptorExecutor dre=new DisruptorExecutor();
|
||||
for (int i = 0; i < 20; i++) {
|
||||
dre.execute(()->{
|
||||
System.out.println("run");
|
||||
});
|
||||
System.out.println("exc");
|
||||
}
|
||||
Thread.sleep(1000);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user