【Guava学习】EventBus

2018-02-03  本文已影响0人  农蓝

1,简单使用

首先引入guave的依赖;

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>23.6-jre</version>
</dependency>

再来看一下基础的使用:

EventBus eventBus = new EventBus();
eventBus.register(new Object() {
    @Subscribe
    public void hehe(String name) {
        System.out.println(name);
    }

    @Subscribe
    public void haha(Object object) {
        System.out.println(object);
    }
});
eventBus.post("huang");//输出两遍huang;
eventBus.post(17);//输出一遍17;

注意点:

EventBus eventBus = new EventBus();
eventBus.register(new Object() {
    @Subscribe
    public void hehe(Integer num) throws InterruptedException {
        System.out.println(num + ":" + System.currentTimeMillis());
        Thread.currentThread().sleep(100);
    }
});
for (int i = 0; i < 10; i++) {
    int finalI = i;
    new Thread(new Runnable() {
        @Override
        public void run() {
            eventBus.post(finalI);
        }
    }).start();
}
//4:1517538494324
//1:1517538494425
//8:1517538494529
//0:1517538494632
//5:1517538494735
//2:1517538494838
//9:1517538494942
//6:1517538495043
//3:1517538495147
//7:1517538495250

2,关于EventBus的并行

线程安全问题是每个java程序猿都应该时刻注意的,当带有@Subscribe的方法被多个方法同时执行,且该方法内部逻辑涉及到更改成员变量时,就会出现线程安全问题,在默认情况下,如果只有@Subscribe注解时,方法时是异步执行的,即使多个线程同时调用,也需要竞争方法的同步锁,然后依次执行;

当需要@Subscribe标注的方法能被多个线程同时调用,需要配合@AllowConcurrentEvents注解使用,该注解表示允许并行执行该方法,当有多个线程同时调用方法时,因为方法无锁,所以线程可以同时进入执行,有锁和无锁,取决于@AllowConcurrentEvents,当没有该注解时,EventBus在生成Subscriber时,使用了SynchronizedSubscriber,该类型在真实调用带有@Subscribe方法时,使用了同步锁,具体后面讲解;

EventBus eventBus = new EventBus();
eventBus.register(new Object() {
    @Subscribe
    @AllowConcurrentEvents
    public void hehe(Integer num) throws InterruptedException {
        System.out.println(num + ":" + System.currentTimeMillis());
        Thread.currentThread().sleep(100);
    }
});
for (int i = 0; i < 10; i++) {
    int finalI = i;
    new Thread(new Runnable() {
        @Override
        public void run() {
            eventBus.post(finalI);
        }
    }).start();
}
//7:1517543101034
//9:1517543101034
//1:1517543101034
//4:1517543101034
//5:1517543101034
//8:1517543101034
//0:1517543101034
//6:1517543101034
//2:1517543101034
//3:1517543101034

关于多线程调用:

EventBus eventBus = new AsyncEventBus(Executors.newFixedThreadPool(3));
eventBus.register(new Object() {
    @Subscribe
    @AllowConcurrentEvents
    public void hehe(Integer num) throws InterruptedException {
        Thread.currentThread().sleep(100);
        System.out.println(Thread.currentThread().getName() + "-" + num + "-" + System.currentTimeMillis());
    }
});
for (int i = 0; i < 9; i++) {
    eventBus.post(i);
}
//pool-1-thread-2-1-1517546343599
//pool-1-thread-1-0-1517546343599
//pool-1-thread-3-2-1517546343599
//pool-1-thread-2-3-1517546343704
//pool-1-thread-3-5-1517546343704
//pool-1-thread-1-4-1517546343704
//pool-1-thread-3-7-1517546343807
//pool-1-thread-1-8-1517546343807
//pool-1-thread-2-6-1517546343807

3,源码解析

基本组件:

3.1,Executor

private enum DirectExecutor implements Executor {
    INSTANCE;
    
    @Override
    public void execute(Runnable command) {
      command.run();
    }
    
    @Override
    public String toString() {
      return "MoreExecutors.directExecutor()";
    }
}

3.2,Dispatcher

private static final class PerThreadQueuedDispatcher extends Dispatcher {
    private final ThreadLocal<Queue<Event>> queue =
            new ThreadLocal<Queue<Event>>() {
                @Override
                protected Queue<Event> initialValue() {
                    return Queues.newArrayDeque();
                }
            };

    private final ThreadLocal<Boolean> dispatching =
            new ThreadLocal<Boolean>() {
                @Override
                protected Boolean initialValue() {
                    return false;
                }
            };

    @Override
    void dispatch(Object event, Iterator<Subscriber> subscribers) {
        checkNotNull(event);
        checkNotNull(subscribers);
        Queue<Event> queueForThread = queue.get();
        queueForThread.offer(new Event(event, subscribers));

        if (!dispatching.get()) {
            dispatching.set(true);
            try 
                Event nextEvent;
                while ((nextEvent = queueForThread.poll()) != null) {
                    while (nextEvent.subscribers.hasNext()) {
                        nextEvent.subscribers.next().dispatchEvent(nextEvent.event);
                    }
                }
            } finally {
                dispatching.remove();
                queue.remove();
            }
        }
    }
}

下面的代码是具体的调用逻辑,使用的是Executor进行具体调用,并执行方法:

final void dispatchEvent(final Object event) {
    executor.execute(
            new Runnable() {
                @Override
                public void run() {
                    try {
                        invokeSubscriberMethod(event);
                    } catch (InvocationTargetException e) {
                        bus.handleSubscriberException(e.getCause(), context(event));
                    }
                }
            });
}

@VisibleForTesting
void invokeSubscriberMethod(Object event) throws InvocationTargetException {
    try {
        method.invoke(target, checkNotNull(event));
    } catch (IllegalArgumentException e) {
        throw new Error("Method rejected target/argument: " + event, e);
    } catch (IllegalAccessException e) {
        throw new Error("Method became inaccessible: " + event, e);
    } catch (InvocationTargetException e) {
        if (e.getCause() instanceof Error) {
            throw (Error) e.getCause();
        }
        throw e;
    }
}

3.3,SubscriberRegistry

static final class SynchronizedSubscriber extends Subscriber {

    private SynchronizedSubscriber(EventBus bus, Object target, Method method) {
        super(bus, target, method);
    }

    @Override
    void invokeSubscriberMethod(Object event) throws InvocationTargetException {
        synchronized (this) {
            super.invokeSubscriberMethod(event);
        }
    }
}
上一篇 下一篇

猜你喜欢

热点阅读