NIO系列5：事件模型

前文讲述了NIO数据读写处理，那么这些数据最终如何被递交给上层业务程序进行处理的呢？

NIO框架一般都采用了事件派发模型来与业务处理器交互，它与原生NIO的事件机制是模型匹配的，缺点是带来了业务处理的碎片化。需要业务程序开发者对事件的生命周期有一个清晰的了解，不像传统方式那么直观。

事件派发器（EventDispatcher）就成为了NIO框架中IO处理线程和业务处理回调接口（Handler）之间的桥梁。

由于业务处理的时间长短是难以确定的，所以一般事件处理器都会分离IO处理线程，使用新的业务处理线程池来进行事件派发，回调业务接口实现。

下面通过一段示例代码来说明事件的派发过程：

这是processor从网络中读取到一段字节后发起的MESSAGE_RECEIVED事件，调用了eventDispatcher.dispatch(Event e)方法。

private void fireMessageReceived(AbstractSession session, ByteBuffer buf, int length) {
		// fire message received event, here we copy buffer bytes to a new byte array to avoid handler expose <code>ByteBuffer</code> to end user.
		byte[] barr = new byte[length];
		System.arraycopy(buf.array(), 0, barr, 0, length);
		eventDispatcher.dispatch(new Event(EventType.MESSAGE_RECEIVED, session, barr, handler));
	}

dispatch的方法实现有以下关键点需要考虑：

1. 事件派发是多线程的，派发线程最终会调用业务回调接口来进行事件处理，回调接口由业务方实现自身去保证线程并发性和安全性。

2. 对于TCP应用来说，由同一session（这里可代表同一个连接）收到的数据必须保证有序派发，不同的session可无序。

3. 不同session的事件派发要尽可能保证公平性，例如：session1有大量事件产生导致派发线程繁忙时，session2产生一个事件不会因为派发线程都在忙于处理session1的事件而被积压，session2的事件也能尽快得到及时派发。

下面是一个实现思路的代码示例：

public void dispatch(Event event) {
		AbstractSession s = (AbstractSession) event.getSession();
		s.add(event);
		if (!s.isEventProcessing()) {
			squeue.offer(s);
		}
	}

为了保证每个session的事件有序，我们将事件存放在每个session自身包含的队列中，然后再将session放入一个公共的阻塞队列中。

有一组worker线程在监听阻塞队列，一旦有session进入队列，它们被激活对session进行事件派发，如下：

public void run() {
			try {
				for (AbstractSession s = squeue.take(); s != null; s = squeue.take()) {					
					// first check any worker is processing this session? if any other worker thread is processing this event with same session, just ignore it.
					synchronized (s) {
						if (!s.isEventProcessing()) {
							s.setEventProcessing(true);
						} else {
							continue;
						}
					}
					
					// fire events with same session
					fire(s);
					
					// last reset processing flag and quit current thread processing
					s.setEventProcessing(false);
					
					// if remaining events, so re-insert to session queue
					if (s.getEventQueue().size() > 0 && !s.isEventProcessing()) {
						squeue.offer(s);
					}
				}
			} catch (InterruptedException e) {
				LOG.warn(e.getMessage(), e);
			}
		}

这里的要点在于，worker不止一个，但为了保证每个session的事件有序我们只能让唯一一个线程对session进行处理，因此可以看到上面的代码中一开始对session进行了加锁，并改变了session的状态（置为事件处理中）。

退出临界区后，进入事件派发处理方法fire()，在fire()方法退出前其他线程都没有机会对该session进行处理，保证了同一时刻只有一个线程进行处理的约束。

如果某个session一直不断有数据进入，则派发线程可能在fire()方法中停留很长时间，具体看fire()的实现如下：

private void fire(Session s) {
			int count = 0;
			Queue<Event> q = s.getEventQueue();
			for (Event event = q.poll(); event != null; event = q.poll()) {
				event.fire();
				count++;
				if (count > SPIN_COUNT) {
					// quit loop to avoid stick same worker thread by same session
					break;
				}
			}
		}

从上面代码可以看出，每次fire()的循环数被设置了一个上限，若事件太多时每次达到上限会退出循环释放线程，等下一次再处理。

当前线程释放对session的控制权只需简单置事件处理状态为false，其他线程就有机会重新获取该session的控制权。

在最后退出前为了避免事件遗漏，因为可能当前线程因为处理事件达到上限数被退出循环而又没有新的事件进入阻塞队列触发新的线程激活，则由当前线程主动去重新将该session放入阻塞队列中激活新线程。