KLALB3.2

src/org/kne/concurrent/DisruptorExecutor.java

@@ -0,0 +1,114 @@
+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);
+	}
+}