disruptor 入门笔记

disruptor入门例子：

package com.taobao.eagleeye.rt.storm;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.taobao.eagleeye.digest.handler.DigestJob;
import com.taobao.eagleeye.digest.handler.DigesterWorkerHanlder;
import com.taobao.eagleeye.disruptor.EventHandler;
import com.taobao.eagleeye.disruptor.RingBuffer;
import com.taobao.eagleeye.disruptor.dsl.Disruptor;
import com.taobao.tlog.util.NamedThreadFactory;

public class Test {

	private static final int RING_BUFFER_SIZE = 8;

	public static void main(String[] args) {
		ExecutorService es = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                60L, TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>(),new NamedThreadFactory("disruptor-process"));
		Disruptor<DigestJob> disruptor = new Disruptor<DigestJob>(DigestJob.EVENT_FACTORY, RING_BUFFER_SIZE, es);
		DigesterWorkerHanlder[] workers = new DigesterWorkerHanlder[12];
		disruptor.handleEventsWith(new EventHandler<DigestJob>() {

			@Override
			public void onEvent(DigestJob event, long sequence, boolean endOfBatch) throws Exception {
				System.out.println("event=" + event);
				Thread.sleep(10000);
			}

		});
		RingBuffer<DigestJob> ringBuffer = disruptor.start();
		while (true) {
			long seq = ringBuffer.next();
			DigestJob job = ringBuffer.get(seq);
			job.context = null;
			ringBuffer.publish(seq);
			System.out.println("emit..."+seq);
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
}

 

package com.taobao.tlog.runtime;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.taobao.tlog.disruptor.EventFactory;
import com.taobao.tlog.disruptor.EventHandler;
import com.taobao.tlog.disruptor.RingBuffer;
import com.taobao.tlog.disruptor.WorkHandler;
import com.taobao.tlog.disruptor.dsl.Disruptor;

public class ActorTest {

	static class ActorHandler implements
			WorkHandler<String>,
			EventHandler<String> {

		@Override
		public void onEvent(String event, long sequence, boolean endOfBatch) throws Exception {
			onEvent(event);
		}

		@Override
		public void onEvent(String event) throws Exception {
			System.out.println(event);
			throw new RuntimeException("fuck");
		}

	}

	public static void main(String[] args) {
		ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat(
				"Action" + "-thread-%d").build();

		EventFactory<String> EVENT_FACTORY = new EventFactory<String>() {
			@Override
			public String newInstance() {
				return new String();
			}
		};

		ExecutorService es = Executors.newCachedThreadPool(namedThreadFactory);

		Disruptor<String> disruptor = new Disruptor<String>(EVENT_FACTORY, 8, es);

		disruptor.handleEventsWith(new ActorHandler());
		RingBuffer<String> ringBuffer = disruptor.start();
		long seq = ringBuffer.next();
		String k = ringBuffer.get(seq);
		k = "test";
		ringBuffer.publish(seq);
	}
}