Disruptor实际应用示例

简介

Disruptor它是一个开源的并发框架，并获得2011 Duke’s 程序框架创新奖，能够在无锁的情况下实现Queue并发操作。同学们代码中是否会使用到BlockingQueue
queue用于缓存通知消息呢？本文介绍的Disruptor则可以完全取代BlockingQueue，带来更快的速度。其它简单内容可能参考百度，熟悉此类需求的同学知道，我们需要两类核心概念功能，一类是事件，一类是针对事件的处理器。
下图为比较常用的的一类使用：

告警处理需求

• 告警信息
• 处理逻辑

告警信息

在disruptor中实例化的构造函数中，需要指定消息的类型或工厂，如下：

 disruptor = new Disruptor<Event>(Event.FACTORY, 4 * 4, executor, ProducerType.SINGLE, new BlockingWaitStrategy()); 

为了能够通用，毕竟系统的disruptor是想处理任务消息的，而不仅仅是告警信息。所以定义了一个通用的消息结构，而不同的具体消息封装在其中。如下：

package com.zte.sunquan.demo.disruptor.event; import com.lmax.disruptor.EventFactory; public class Event<T> { 
    public static final EventFactory<Event> FACTORY = new EventFactory<Event>() { 
    @Override public Event newInstance() { 
    return new Event(); } }; private T value; private Event() { 
    } public Event(T value) { 
    this.value = value; } public void setValue(T value) { 
    this.value = value; } public T getValue() { 
    return value; } } 

而告警消息的结构则为：

package com.zte.sunquan.demo.disruptor.main; public class AlarmEvent { 
    private AlarmType value; public AlarmEvent(AlarmType value) { 
    this.value = value; } enum AlarmType { 
    NO_POWER, HARDWARE_DAMAGE, SOFT_DAMAGE; } public AlarmType getValue() { 
    return value; } } 

在构建往disruptor发送的消息时，进行AlarmEvent的封装。

Event commEvent = new Event(event);

处理器

在完成了消息的定义后，下面则需要定义消息的处理器，这里利用了@Handler注解用于定义处理器。

package com.zte.sunquan.demo.disruptor.annotation; import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; / * Created by on 2018/9/10. */ @Target({ 
   ElementType.TYPE}) @Retention(RetentionPolicy.RUNTIME) public @interface Handler { 
    String value() default "toString"; } 

通过注解可以方便地在系统的各个地方定义消息处理模块。示例处理逻辑如下：

package com.zte.sunquan.demo.disruptor.main; import com.zte.sunquan.demo.disruptor.annotation.Handler; import com.zte.sunquan.demo.disruptor.handler.AbstractEventHandler; @Handler public class AlarmEventHandler extends AbstractEventHandler<AlarmEvent> { 
    @Override public void handler(AlarmEvent alarmEvent) { 
    AlarmEvent.AlarmType alarmType = alarmEvent.getValue(); System.out.println("Got alarm:" + alarmType.name()); switch (alarmType) { 
    case NO_POWER: System.out.println("charging"); break; case HARDWARE_DAMAGE: System.out.println("repair"); break; case SOFT_DAMAGE: System.out.println("reinstall"); break; default: System.out.println("ignore"); break; } } } 

调度

至此有了消息和消息处理逻辑，如何将两者通过disruptor串联起来？

DisruptorApplication

定义该类，作为调试的核心，继承了AbstractApplication
代码参考：

package com.zte.sunquan.demo.disruptor; import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import com.lmax.disruptor.BlockingWaitStrategy; import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.dsl.Disruptor; import com.lmax.disruptor.dsl.ProducerType; import com.zte.sunquan.demo.disruptor.event.Event; import com.zte.sunquan.demo.disruptor.handler.EventHandlerProcess; import com.zte.sunquan.demo.thread.UserThreadFactory; public class AbstractApplication { 
    private static final int CPUS = Runtime.getRuntime().availableProcessors(); private static final UserThreadFactory threadFactory = UserThreadFactory.build("disruptor-test"); private ExecutorService executor; protected Disruptor<Event> disruptor; private EventHandlerProcess process = new EventHandlerProcess(); private Class eventClass; public AbstractApplication(Class eventClass) { 
    this.eventClass = eventClass; init(); try { 
    process(); } catch (InstantiationException e) { 
    e.printStackTrace(); } catch (IllegalAccessException e) { 
    e.printStackTrace(); } } public void init() { 
    executor = Executors.newFixedThreadPool(CPUS * 2, threadFactory); disruptor = new Disruptor<Event>(Event.FACTORY, 4 * 4, executor, ProducerType.SINGLE, new BlockingWaitStrategy()); } private void process() throws InstantiationException, IllegalAccessException { 
    Set<EventHandler> eventHandlers = process.scanHandlers(); for (EventHandler handler : eventHandlers) { 
    // if (filterHandler(handler)) { 
    disruptor.handleEventsWith(handler); // } } disruptor.start(); } private boolean filterHandler(EventHandler handler) { 
    if (handler != null) { 
    //继承类 Type genericSuperclass = handler.getClass().getGenericSuperclass(); if(genericSuperclass==Object.class) return false; ParameterizedType parameterizedType = (ParameterizedType) genericSuperclass; //实现接口handler.getClass().getGenericInterfaces() Type[] actualTypeArguments = parameterizedType.getActualTypeArguments(); return actualTypeArguments[0] == eventClass; } return false; } public void close() { 
    disruptor.shutdown(); executor.shutdown(); } } 

package com.zte.sunquan.demo.disruptor; import com.zte.sunquan.demo.disruptor.event.Event; / * Created by on 2018/9/10. */ public class DisruptorApplication extends AbstractApplication { 
    public DisruptorApplication(Class eventClass) { 
    super(eventClass); } public <T> void publishEvent(T event) { 
    Event commEvent = new Event(event); final long seq = disruptor.getRingBuffer().next(); Event userEvent = (Event) disruptor.get(seq); userEvent.setValue(commEvent.getValue()); disruptor.getRingBuffer().publish(seq); } } 

EventHandlerProcess

需要注意的是在AbstractApplicaton的process中借助EventHandlerProcess处理@Handler注解，具体逻辑如下：

package com.zte.sunquan.demo.disruptor.handler; import java.lang.annotation.Annotation; import java.util.Arrays; import java.util.List; import java.util.Set; import java.util.stream.Collectors; import com.google.common.base.Preconditions; import com.lmax.disruptor.EventHandler; import com.zte.sunquan.demo.disruptor.annotation.Handler; import com.zte.sunquan.demo.disruptor.util.ClassUtil; / * Created by on 2018/9/10. */ public class EventHandlerProcess { 
    public Set<EventHandler> scanHandlers(){ 
    List<Class<?>> clsList = ClassUtil .getAllClassByPackageName("com.zte.sunquan.demo.disruptor"); return clsList.stream() .filter(p->p.getAnnotation(Handler.class)!=null) .map(c-> { 
    try { 
    return (EventHandler)c.newInstance(); } catch (InstantiationException e) { 
    e.printStackTrace(); } catch (IllegalAccessException e) { 
    e.printStackTrace(); } return null; }) .collect(Collectors.toSet()); } public EventHandler scanJavaClass(Class cls) throws IllegalAccessException, InstantiationException { 
    Annotation annotation = cls.getAnnotation(Handler.class); if (annotation != null) { 
    Object instance = cls.newInstance(); //是否可以做成动态字节码生成，框架内部完成接口类的实现 Preconditions.checkState(instance instanceof EventHandler); return (EventHandler) instance; } return null; } public static void main(String[] args) throws InstantiationException, IllegalAccessException { 
    String classPath = (String) System.getProperties().get("java.class.path"); List<String> packages = Arrays.stream(classPath.split(";")).filter(p -> p.contains("demo-disruptor-project")).collect(Collectors.toList()); //List 
   
     > clsList = ClassUtil.getAllClassByPackageName(BlackPeople.class.getPackage()); 
    List<Class<?>> clsList = ClassUtil.getAllClassByPackageName("com.zte.sunquan.demo.disruptor"); System.out.println(clsList); // EventHandlerProcess process = new EventHandlerProcess(); // EventHandler eventHandler = process.scanJavaClass(LogEventHandler.class); // System.out.println(eventHandler); // Properties properties = System.getProperties(); // System.out.println(properties); } } 

ClassUtil

用于查找加了@Handler注解类，该方式可以进一步优先，直接搜索class路径下的所有类，而不是指定文件夹里的。

package com.zte.sunquan.demo.disruptor.util; import java.io.File; import java.io.FileFilter; import java.io.IOException; import java.net.JarURLConnection; import java.net.URL; import java.net.URLDecoder; import java.util.ArrayList; import java.util.Enumeration; import java.util.List; import java.util.jar.JarEntry; import java.util.jar.JarFile; public class ClassUtil { 
    public static List<Class<?>> getAllClassByPackageName(String packageName) { 
    // 获取当前包下以及子包下所以的类 List<Class<?>> returnClassList = getClasses(packageName); return returnClassList; } / * 通过包名获取包内所有类 * * @param pkg * @return */ public static List<Class<?>> getAllClassByPackageName(Package pkg) { 
    String packageName = pkg.getName(); // 获取当前包下以及子包下所以的类 List<Class<?>> returnClassList = getClasses(packageName); return returnClassList; } / * 通过接口名取得某个接口下所有实现这个接口的类 */ public static List<Class<?>> getAllClassByInterface(Class<?> c) { 
    List<Class<?>> returnClassList = null; if (c.isInterface()) { 
    // 获取当前的包名 String packageName = c.getPackage().getName(); // 获取当前包下以及子包下所以的类 List<Class<?>> allClass = getClasses(packageName); if (allClass != null) { 
    returnClassList = new ArrayList<Class<?>>(); for (Class<?> cls : allClass) { 
    // 判断是否是同一个接口 if (c.isAssignableFrom(cls)) { 
    // 本身不加入进去 if (!c.equals(cls)) { 
    returnClassList.add(cls); } } } } } return returnClassList; } / * 取得某一类所在包的所有类名 不含迭代 */ public static String[] getPackageAllClassName(String classLocation, String packageName) { 
    // 将packageName分解 String[] packagePathSplit = packageName.split("[.]"); String realClassLocation = classLocation; int packageLength = packagePathSplit.length; for (int i = 0; i < packageLength; i++) { 
    realClassLocation = realClassLocation + File.separator + packagePathSplit[i]; } File packeageDir = new File(realClassLocation); if (packeageDir.isDirectory()) { 
    String[] allClassName = packeageDir.list(); return allClassName; } return null; } / * 从包package中获取所有的Class * * @param pack * @return */ private static List<Class<?>> getClasses(String packageName) { 
    // 第一个class类的集合 List<Class<?>> classes = new ArrayList<Class<?>>(); // 是否循环迭代 boolean recursive = true; // 获取包的名字 并进行替换 String packageDirName = packageName.replace('.', '/'); // 定义一个枚举的集合 并进行循环来处理这个目录下的things Enumeration<URL> dirs; try { 
    dirs = Thread.currentThread().getContextClassLoader().getResources(packageDirName); // 循环迭代下去 while (dirs.hasMoreElements()) { 
    // 获取下一个元素 URL url = dirs.nextElement(); // 得到协议的名称 String protocol = url.getProtocol(); // 如果是以文件的形式保存在服务器上 if ("file".equals(protocol)) { 
    // 获取包的物理路径 String filePath = URLDecoder.decode(url.getFile(), "UTF-8"); // 以文件的方式扫描整个包下的文件 并添加到集合中 findAndAddClassesInPackageByFile(packageName, filePath, recursive, classes); } else if ("jar".equals(protocol)) { 
    // 如果是jar包文件 // 定义一个JarFile JarFile jar; try { 
    // 获取jar jar = ((JarURLConnection) url.openConnection()).getJarFile(); // 从此jar包 得到一个枚举类 Enumeration<JarEntry> entries = jar.entries(); // 同样的进行循环迭代 while (entries.hasMoreElements()) { 
    // 获取jar里的一个实体 可以是目录 和一些jar包里的其他文件 如META-INF等文件 JarEntry entry = entries.nextElement(); String name = entry.getName(); // 如果是以/开头的 if (name.charAt(0) == '/') { 
    // 获取后面的字符串 name = name.substring(1); } // 如果前半部分和定义的包名相同 if (name.startsWith(packageDirName)) { 
    int idx = name.lastIndexOf('/'); // 如果以"/"结尾 是一个包 if (idx != -1) { 
    // 获取包名 把"/"替换成"." packageName = name.substring(0, idx).replace('/', '.'); } // 如果可以迭代下去 并且是一个包 if ((idx != -1) || recursive) { 
    // 如果是一个.class文件 而且不是目录 if (name.endsWith(".class") && !entry.isDirectory()) { 
    // 去掉后面的".class" 获取真正的类名 String className = name.substring(packageName.length() + 1, name.length() - 6); try { 
    // 添加到classes classes.add(Class.forName(packageName + '.' + className)); } catch (ClassNotFoundException e) { 
    e.printStackTrace(); } } } } } } catch (IOException e) { 
    e.printStackTrace(); } } } } catch (IOException e) { 
    e.printStackTrace(); } return classes; } / * 以文件的形式来获取包下的所有Class * * @param packageName * @param packagePath * @param recursive * @param classes */ private static void findAndAddClassesInPackageByFile(String packageName, String packagePath, final boolean recursive, List<Class<?>> classes) { 
    // 获取此包的目录 建立一个File File dir = new File(packagePath); // 如果不存在或者 也不是目录就直接返回 if (!dir.exists() || !dir.isDirectory()) { 
    return; } // 如果存在 就获取包下的所有文件 包括目录 File[] dirfiles = dir.listFiles(new FileFilter() { 
    // 自定义过滤规则 如果可以循环(包含子目录) 或则是以.class结尾的文件(编译好的java类文件) public boolean accept(File file) { 
    return (recursive && file.isDirectory()) || (file.getName().endsWith(".class")); } }); // 循环所有文件 for (File file : dirfiles) { 
    // 如果是目录 则继续扫描 if (file.isDirectory()) { 
    findAndAddClassesInPackageByFile(packageName + "." + file.getName(), file.getAbsolutePath(), recursive, classes); } else { 
    // 如果是java类文件 去掉后面的.class 只留下类名 String className = file.getName().substring(0, file.getName().length() - 6); try { 
    // 添加到集合中去 classes.add(Class.forName(packageName + '.' + className)); } catch (ClassNotFoundException e) { 
    e.printStackTrace(); } } } } } 

测试

用例为：

package com.zte.sunquan.demo.disruptor.main; import org.junit.After; import org.junit.Before; import com.zte.sunquan.demo.disruptor.DisruptorApplication; public class Test { 
    private DisruptorApplication application; @Before public void setUp() { 
    application = new DisruptorApplication(AlarmEvent.class); } @org.junit.Test public void testDisruptor() throws InterruptedException { 
    //1.准备AlarmEvent AlarmEvent event = new AlarmEvent(AlarmEvent.AlarmType.HARDWARE_DAMAGE); //2.发送AlarmEvent application.publishEvent(event); } @org.junit.Test public void testDisruptor2() throws InterruptedException { 
    for (int i = 0; i < 100; i++) { 
    AlarmEvent event = new AlarmEvent(AlarmEvent.AlarmType.NO_POWER); application.publishEvent(event); } } @After public void after() throws InterruptedException { 
    Thread.sleep(1000); application.close(); } } 

在AbstractApplication中filterHandler用于对消息过滤，因为按上面的实现一个disruptor只能处理一类消息，相关泛型被擦出，无法匹配判断。所以用例中使用

如果做到这一步，虽然满足了要求，但功能太过封闭。其它的消息类型如果增加？按上述方式，只能重新定义DisruptorApplication。

AbstractEventHandler

在该类中增加了filter方法

public abstract class AbstractEventHandler<T> implements EventHandler<Event<T>> { 
    @Override public void onEvent(Event<T> tEvent, long l, boolean b) throws Exception { 
    if(filter(tEvent)) { 
    T t = tEvent.getValue(); handler(t); } } public boolean filter(Event event){ 
    return false; } public abstract void handler(T t); } 

这就要求每个消息处理类，都要显示的指定自身能处理的消息

@Handler public class UnknownEventHandler extends AbstractEventHandler<OtherEvent> { 
    @Override public boolean filter(Event event) { 
    if (event.getValue().getClass() == OtherEvent.class) { 
    return true; } return false; } 

在AlarmEventHandler中增加filter

@Handler public class AlarmEventHandler extends AbstractEventHandler<AlarmEvent> { 
    @Override public boolean filter(Event event) { 
    if (event.getValue().getClass() == AlarmEvent.class) { 
    return true; } return false; } 

新的测试用例为：

 @org.junit.Test public void testDisruptorSuper() throws InterruptedException { 
    application = new DisruptorApplication(); //1.准备AlarmEvent AlarmEvent event = new AlarmEvent(AlarmEvent.AlarmType.HARDWARE_DAMAGE); //2.发送AlarmEvent application.publishEvent(event); } 

看到构造函数DisruptorApplication中去掉了入参，同时AlarmEventHandler正确处理了消息。

也许你会好奇ODL中是如何做的，他系统范围内也只有一个Disruptor,在ODL中，disruptor的Handler只是作了转发Handler，该Handler的工作才是寻找对应的EventHandler。

ODL实现

 private DOMNotificationRouter(final ExecutorService executor, final int queueDepth, final WaitStrategy strategy) { 
    this.executor = Preconditions.checkNotNull(executor); disruptor = new Disruptor<>(DOMNotificationRouterEvent.FACTORY, queueDepth, executor, ProducerType.MULTI, strategy); disruptor.handleEventsWith(DISPATCH_NOTIFICATIONS); disruptor.after(DISPATCH_NOTIFICATIONS).handleEventsWith(NOTIFY_FUTURE); disruptor.start(); } 

而Handler的onEvent为

 private static final EventHandler<DOMNotificationRouterEvent> DISPATCH_NOTIFICATIONS = new EventHandler<DOMNotificationRouterEvent>() { 
    @Override public void onEvent(final DOMNotificationRouterEvent event, final long sequence, final boolean endOfBatch) throws Exception { 
    event.deliverNotification(); onEvnetCount.incrementAndGet(); } }; 

默认先执行了 event.deliverNotification()，注意是event中的方法，具体实现：

 void deliverNotification() { 
    LOG.trace("Start delivery of notification {}", notification); for (ListenerRegistration<? extends DOMNotificationListener> r : subscribers) { 
    final DOMNotificationListener listener = r.getInstance(); if (listener != null) { 
    try { 
    LOG.trace("Notifying listener {}", listener); listener.onNotification(notification); LOG.trace("Listener notification completed"); } catch (Exception e) { 
    LOG.error("Delivery of notification {} caused an error in listener {}", notification, listener, e); } } } LOG.trace("Delivery completed"); } 

可以看到其在一个subscribers的列表中寻找对应的Listenen进行方法调用执行。

看到这，应该明白了ODL的handler只简单负责转换，真正的选择执行对象在 event.deliverNotification()，那一个事件，如何知道有哪些定阅者呢？必然要存在一个定阅或注册的过程。代码如下：

@Override public synchronized <T extends DOMNotificationListener> ListenerRegistration<T> registerNotificationListener(final T listener, final Collection<SchemaPath> types) { 
    final ListenerRegistration<T> reg = new AbstractListenerRegistration<T>(listener) { 
    @Override protected void removeRegistration() { 
    final ListenerRegistration<T> me = this; synchronized (DOMNotificationRouter.this) { 
    replaceListeners(ImmutableMultimap.copyOf(Multimaps.filterValues(listeners, new Predicate<ListenerRegistration<? extends DOMNotificationListener>>() { 
    @Override public boolean apply(final ListenerRegistration<? extends DOMNotificationListener> input) { 
    if (input == me) { 
    logDomNotificationChanges(listener, null, "removed"); } return input != me; } }))); } } }; if (!types.isEmpty()) { 
    final Builder<SchemaPath, ListenerRegistration<? extends DOMNotificationListener>> b = ImmutableMultimap.builder(); b.putAll(listeners); for (final SchemaPath t : types) { 
    b.put(t, reg); logDomNotificationChanges(listener, t, "added"); } replaceListeners(b.build()); } return reg; } 

根据YANG中定义的scheam进行了注册。这样在Notification注册时，则绑定好了事件类型与处理逻辑的对应关系。而在封装消息时，将subscribe传递给了event如下：

 private ListenableFuture<Void> publish(final long seq, final DOMNotification notification, final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers) { 
    final DOMNotificationRouterEvent event = disruptor.get(seq); final ListenableFuture<Void> future = event.initialize(notification, subscribers); logDomNotificationExecute(subscribers, notification, seq, "publish"); disruptor.getRingBuffer().publish(seq); publishCount.incrementAndGet(); return future; } 

与上文的实现方式相比，避免了每一个消息，都进行全范围的Handler的filter判断。