用guava实现简单的事件驱动

Guava的EventBus可以简化生产/消费模型。EventBus通过非常简单的方式，实现了观察者模式中的监听注册，事件分发。有了这个玩意，真的可以替换Observer模式了

 

Observer模式是比较常用的设计模式之一，虽然有时候在具体代码里，它不一定叫这个名字，比如改头换面叫个Listener，但模式就是这个模式。手工实现一个Observer也不是多复杂的一件事，只是因为这个设计模式实在太常用了，Java就把它放到了JDK里面：Observable和Observer，从JDK 1.0里，它们就一直在那里。从某种程度上说，它简化了Observer模式的开发，至少我们不用再手工维护自己的Observer列表了。不过，如前所述，JDK里的Observer从1.0就在那里了，直到Java 7，它都没有什么改变，就连通知的参数还是Object类型。要知道，Java 5就已经泛型了。Java 5是一次大规模的语法调整，许多程序库从那开始重新设计了API，使其更简洁易用。当然，那些不做应对的程序库，多半也就过时了。这也就是这里要讨论知识更新的原因所在。今天，对于普通的应用，如果要使用Observer模式该如何做呢？答案是Guava的EventBus。

 

import com.google.common.eventbus.DeadEvent;

import com.google.common.eventbus.Subscribe;

 

EventBus基本用法：

 

　　使用Guava之后, 如果要订阅消息, 就不用再继承指定的接口, 只需要在指定的方法上加上@Subscribe注解即可。代码如下：

 

　　消息封装类：

public class TestEvent {
private final int message;
public TestEvent(int message) {
this.message = message;
System.out.println("event message:"+message);
}
public int getMessage() {
return message;
}
}

 

　　消息接受类：

public class EventListener {
public int lastMessage = 0;

@Subscribe
public void listen(TestEvent event) {
lastMessage = event.getMessage();
System.out.println("Message:"+lastMessage);
}

public int getLastMessage() {
return lastMessage;
}
}

 

　　测试类及输出结果：

public class TestEventBus {
@Test
public void testReceiveEvent() throws Exception {

EventBus eventBus = new EventBus("test");
EventListener listener = new EventListener();

eventBus.register(listener);

eventBus.post(new TestEvent(200));
eventBus.post(new TestEvent(300));
eventBus.post(new TestEvent(400));

System.out.println("LastMessage:"+listener.getLastMessage());
;
}
}

 

//输出信息

event message:200

Message:200

event message:300

Message:300

event message:400

Message:400

LastMessage:400

 

 　　MultiListener的使用：

　　只需要在要订阅消息的方法上加上@Subscribe注解即可实现对多个消息的订阅，代码如下：

public class MultipleListener {
public Integer lastInteger;
public Long lastLong;

@Subscribe
public void listenInteger(Integer event) {
lastInteger = event;
System.out.println("event Integer:"+lastInteger);
}

@Subscribe
public void listenLong(Long event) {
lastLong = event;
System.out.println("event Long:"+lastLong);
}

public Integer getLastInteger() {
return lastInteger;
}

public Long getLastLong() {
return lastLong;
}
}

 

　　测试类：

public class TestMultipleEvents {
@Test
public void testMultipleEvents() throws Exception {

EventBus eventBus = new EventBus("test");
MultipleListener multiListener = new MultipleListener();

eventBus.register(multiListener);

eventBus.post(new Integer(100));
eventBus.post(new Integer(200));
eventBus.post(new Integer(300));
eventBus.post(new Long(800));
eventBus.post(new Long(800990));
eventBus.post(new Long(800882934));

System.out.println("LastInteger:"+multiListener.getLastInteger());
System.out.println("LastLong:"+multiListener.getLastLong());
}
}

 

//输出信息

event Integer:100

event Integer:200

event Integer:300

event Long:800

event Long:800990

event Long:800882934

LastInteger:300

LastLong:800882934

 

　　Dead Event：

　　如果EventBus发送的消息都不是订阅者关心的称之为Dead Event。实例如下：

public class DeadEventListener {
boolean notDelivered = false;

@Subscribe
public void listen(DeadEvent event) {

notDelivered = true;
}

public boolean isNotDelivered() {
return notDelivered;
}
}

 

　　测试类：

public class TestDeadEventListeners {
@Test
public void testDeadEventListeners() throws Exception {

EventBus eventBus = new EventBus("test");
DeadEventListener deadEventListener = new DeadEventListener();
eventBus.register(deadEventListener);

eventBus.post(new TestEvent(200));
eventBus.post(new TestEvent(300));

System.out.println("deadEvent:"+deadEventListener.isNotDelivered());

}
}

 

//输出信息

event message:200

event message:300

deadEvent:true

 

　　说明：如果没有消息订阅者监听消息， EventBus将发送DeadEvent消息，这时我们可以通过log的方式来记录这种状态。

 

　　Event的继承：

　　如果Listener A监听Event A, 而Event A有一个子类Event B, 此时Listener A将同时接收Event A和B消息，实例如下：

 

　　Listener 类：

public class NumberListener {

private Number lastMessage;

@Subscribe
public void listen(Number integer) {
lastMessage = integer;
System.out.println("Message:"+lastMessage);
}

public Number getLastMessage() {
return lastMessage;
}
}

public class IntegerListener {

private Integer lastMessage;

@Subscribe
public void listen(Integer integer) {
lastMessage = integer;
System.out.println("Message:"+lastMessage);
}

public Integer getLastMessage() {
return lastMessage;
}
}

 

　　测试类：

public class TestEventsFromSubclass {
@Test
public void testEventsFromSubclass() throws Exception {

EventBus eventBus = new EventBus("test");
IntegerListener integerListener = new IntegerListener();
NumberListener numberListener = new NumberListener();
eventBus.register(integerListener);
eventBus.register(numberListener);

eventBus.post(new Integer(100));

System.out.println("integerListener message:"+integerListener.getLastMessage());
System.out.println("numberListener message:"+numberListener.getLastMessage());

eventBus.post(new Long(200L));

System.out.println("integerListener message:"+integerListener.getLastMessage());
System.out.println("numberListener message:"+numberListener.getLastMessage());
}
}

 //输出类

Message:100

Message:100

integerListener message:100

numberListener message:100

Message:200

integerListener message:100

numberListener message:200

 

　　说明：在这个方法中,我们看到第一个事件(新的整数(100))是收到两个听众,但第二个(新长(200 l))只能到达NumberListener作为整数一不是创建这种类型的事件。可以使用此功能来创建更通用的监听器监听一个广泛的事件和更详细的具体的特殊的事件。

 

 　　一个综合实例：

public class UserThread extends Thread {
private Socket connection;
private EventBus channel;
private BufferedReader in;
private PrintWriter out;

public UserThread(Socket connection, EventBus channel) {
this.connection = connection;
this.channel = channel;
try {
in = new BufferedReader(new InputStreamReader(connection.getInputStream()));
out = new PrintWriter(connection.getOutputStream(), true);
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
}

@Subscribe
public void recieveMessage(String message) {
if (out != null) {
out.println(message);
System.out.println("recieveMessage:"+message);
}
}

@Override
public void run() {
try {
String input;
while ((input = in.readLine()) != null) {
channel.post(input);
}
} catch (IOException e) {
e.printStackTrace();
}

//reached eof
channel.unregister(this);
try {
connection.close();
} catch (IOException e) {
e.printStackTrace();
}
in = null;
out = null;
}
}

 

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

import com.google.common.eventbus.EventBus;

public class EventBusChat {
public static void main(String[] args) {
EventBus channel = new EventBus();
ServerSocket socket;
try {
socket = new ServerSocket(4444);
while (true) {
Socket connection = socket.accept();
UserThread newUser = new UserThread(connection, channel);
channel.register(newUser);
newUser.start();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}

 

 

　　说明：用telnet命令登录：telnet 127.0.0.1 4444 ，如果你连接多个实例你会看到任何消息发送被传送到其他实例。

 

另参考：Guava的生产/消费模型 —— EventBus