Guava中的EventBus

其实代码中经常会遇到跟主流程分支出去的异步逻辑，比如说：

爬虫处理逻辑中，进行心跳打点，订单处理中，需要触发用户的个人信息变更等。

这个时候就应该使用观察者模式。

EventBus是Guava的事件处理机制，是设计模式中的观察者模式（生产/消费者编程模型）的优雅实现。对于事件监听和发布订阅模式，EventBus是一个非常优雅和简单解决方案，我们不用创建复杂的类和接口层次结构。

1import com.google.common.eventbus.Subscribe;
2
3public class MultipleListener {
4	public Integer lastInteger;
5	public Long lastLong;
6
7	@Subscribe
8	public void listenInteger(Integer event) {
9		lastInteger = event;
10		System.out.println("event Integer:" + lastInteger);
11	}
12
13	@Subscribe
14	public void listenLong(Long event) {
15		lastLong = event;
16		System.out.println("event Long:" + lastLong);
17	}
18
19	public Integer getLastInteger() {
20		return lastInteger;
21	}
22
23	public Long getLastLong() {
24		return lastLong;
25	}
26}
27
28import com.google.common.eventbus.Subscribe;
29
30public class EventListener {
31	
32	public int lastMessage = 0;
33	
34	@Subscribe
35    public void listen(TestEvent event) {
36        lastMessage = event.getMessage();
37        System.out.println("Message:"+lastMessage);
38    }
39
40    public int getLastMessage() {      
41        return lastMessage;
42    }
43    
44}
45
46import com.google.common.eventbus.EventBus;
47
48public class TestEventBusMain {
49	
50	
51	public static void testMultipleEvents() throws Exception {  
52	       
53        EventBus eventBus = new EventBus("test");  
54        MultipleListener multiListener = new MultipleListener();  
55       
56        eventBus.register(multiListener);  
57       
58        eventBus.post(new Integer(100));
59        eventBus.post(new Integer(200));  
60        eventBus.post(new Integer(300));  
61        eventBus.post(new Long(800)); 
62        eventBus.post(new Long(800990));  
63        eventBus.post(new Long(800882934));  
64       
65        System.out.println("LastInteger:"+multiListener.getLastInteger());
66        System.out.println("LastLong:"+multiListener.getLastLong());
67    }   
68	
69	public static void testReceiveEvent() {
70		EventBus eventBus = new EventBus("test");
71		EventListener listener = new EventListener();
72
73		eventBus.register(listener);
74
75		eventBus.post(new TestEvent(200));
76		eventBus.post(new TestEvent(300));
77		eventBus.post(new TestEvent(400));
78
79		System.out.println("LastMessage:" + listener.getLastMessage());
80	}
81	
82	
83	public static void main(String[] args) throws Exception {
84		testMultipleEvents() ;
85	}
86
87}

实际上EventBus要表达的意图很简单,就是将post(Object arg)这里的arg当做参数传入到已注册的方法(被@Subscribe)的方法里,并调用该方法,所以当post(String)的时候,调用的参数类型为String的注册方法,当post(int)的时候,调用则是参数类型为Integer的注册方法。

在多线程的结构下，可以使用AsyncEventBus

1public void testAysncEventBus() {  
2    AsyncEventBus eventBus = new AsyncEventBus(Executors.newFixedThreadPool(3));  
3    eventBus.register(new EventListener());  
4    eventBus.post("ssdf");  
5    System.out.println("==");  
6}  

但是作为异步调用，EventBus的封装显得比较单薄。

parseq对于异步的包装则非常令人称赞。

https://github.com/linkedin/parseq