欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

pushlet源码学习

程序员文章站 2024-02-01 16:30:34
...

 

Protocol: 一些常量的定义

Config: 加载和维护配置信息,如加载Pushlet.properties配置文件

SessionManager: 管理session的整个生命周期

EventSourceManager: 事件源管理类

Controller: 对来自客户端的请求的处理类,包含请求的事件的各种处理方法(封装响应事件ResponseEvent)

Subscriber: 订阅信息的添加删除,及事件推送等。

Dispatcher: 分配事件给订阅者

首先加载Pushlet类,它为一个servlet实现Protocol接口。

Pushlet中的init方法:

1. 取得路径 2. 加载配置文件 3.日志初始化 4. 初始化SessionManager 5. 初始化Dispatcher 6. 加载事件源

 

public void init() throws ServletException {
		try {
			// Load configuration (from classpath or WEB-INF root path)
			String webInfPath = getServletContext().getRealPath("/") + "/WEB-INF";
			Config.load(webInfPath);

			Log.init();
// Start session manager  通过单例模式获取sessionManager实例
			SessionManager.getInstance().start();

			// Start event Dispatcher
			Dispatcher.getInstance().start();
            if (Config.getBoolProperty(Config.SOURCES_ACTIVATE)) {
             //判断配置文件中是否设定为使用事件源,是则加载sources.properties
             //并对其中的所有事件源创建单独的守护进程并激活。
				EventSourceManager.start(webInfPath);
			} else {
				Log.info("Not starting local event sources");
			}
}

 SessionManager中的start方法通过JDK中的Timer类,重复执行任务。

 

public void start() throws PushletException {
		if (timer != null) {
			stop();
		}
		timer = new Timer(false);
		timer.schedule(new AgingTimerTask(), TIMER_INTERVAL_MILLIS, TIMER_INTERVAL_MILLIS);
		info("started; interval=" + TIMER_INTERVAL_MILLIS + "ms");
	}

 AgingTimerTask类即定义的任务,在其中不断判断sessionCache中的所有session是否超时,若超时则停止session

在定时器的run方法中调用apply方法,在这里应用了Visitor模式,apply方法即visitor提供的对所有元素访问的接口,Session即被访问的元素。在SessionManager中有Session[] sessionCacheMap sessions变量,前者用于缓存可修改的session,后者保存活跃的session。在apply中访问所有sessionCache中的session,通过反射访问AgingTimerTask中的回调方法visitVisist方法则用于判断此session是否已超时。

 

public AgingTimerTask() throws PushletException {
			try {
				// Setup Visitor Methods for callback from SessionManager
				Class[] argsClasses = {Session.class};
				visitMethod = this.getClass().getMethod("visit", argsClasses);//通过反射取得此类中的visit方法
			} catch (NoSuchMethodException e) {
				throw new PushletException("Failed to setup AgingTimerTask", e);
			}
		}
		/**
		 * Clock tick callback from Timer.
		 */
		public void run() {
			long now = Sys.now(); //获取当前时间
			delta = now - lastRun; //两次时间差
			lastRun = now; //将当前时间设为上一次是时间
			debug("AgingTimerTask: tick");
			// Use Visitor pattern to loop through Session objects (see visit() below)
			getInstance().apply(this, visitMethod, new Object[1]);
		}
		/**
		 * Callback from SessionManager during apply()
		 */
		public void visit(Session aSession) {
			try {
				// Age the lease
				aSession.age(delta); //减少session的生存时间
				debug("AgingTimerTask: visit: " + aSession);
				// Stop session if lease expired 超时则停止session
				if (aSession.isExpired()) {
					info("AgingTimerTask: Session expired: " + aSession);
					aSession.stop();
				}
			} catch (Throwable t) {
				warn("AgingTimerTask: Error in timer task : " + t);
			}
		}

 Dispatcher初始化很简单,创建一个SessionManagerVisitor实例,其为visitor模式中的visitor。其构造方法中将visitMulticastvisitBroadcast方法以Method存入一个Map中。这两个方法即在visitor中单独访问元素的visit方法。

       最后若使用事件源(sources.activatetrue),将加载事件源配置文件,并对其中的所有事件创建守护线程并激活线程。在EventSourceManager中创建一个全局的事件源集合,现调用activate()激活事件源,此方法应用templates模式其实现由其子类EventPullSource实现,此类中便是对线程进行管理。在线程的run()中产生事件并多点发布。线程每睡眠一段时间便发布一次时间。

 

public void run() {
		Log.debug(getClass().getName() + ": starting...");
		alive = true;
		while (alive) {
			try {//getSleepTime()由继承EventPullSource的子类实现
				Thread.sleep(getSleepTime());
				// Stopped during sleep: end loop.
				if (!alive) {
					break;
				}
				// If passivated wait until we get
				// get notify()-ied. If there are no subscribers
				// it wasts CPU to remain producing events...
				synchronized (this) {
					while (!active) {
						Log.debug(getClass().getName() + ": waiting...");
						wait();
					}
				}
			} catch (InterruptedException e) {
				break;
			}
			try {
				//pullEvent()由继承EventPullSource的子类实现
				Event event = pullEvent();
				// 多点发布事件
				Dispatcher.getInstance().multicast(event);
			} catch (Throwable t) {
				Log.warn("EventPullSource exception while multicasting ", t);
				t.printStackTrace();
			}
		}
		Log.debug(getClass().getName() + ": stopped");
	}

 至此Pushlet的初始化即完成。

当收到http请求时无论doGet还是doPost都是创建一个事件,事件Event通过一个HashMap来保存”p_event”事件类型及其属性,最后都调用doRequest进行处理。其中doGet针对普通请求,doPost针对xml处理。

doRequest处理中,当事件类开为”join”时说明还未创建session,此时创建一个新的session,否则通过Event中保存的SessionID获取session。在session在创建过程中,包括随机生成sessionid并依据session生成ControllerSubscriber实例赋给session

然后创建一个Command对象,调用Controller中的doCommand执行。Command为对request,response,event等信息的一个封装。

doCommand中执行如下:

更新Session的存活时间。重新设置为默认值。

设置 Session中的address为客户端请求的IP地址。

根据事件类型执行相应的方法。

Refresh:  Command的响应事件中加入一个”refresh-ack”刷新确认事件。

Subscribe: 订阅

调用Controller中的doSubscribe(),当事件中存在主题Subject时,首先获取初始化时创建的订阅者Subscriber对象,然后调用Subscriber中的addSubscriber创建一订阅信息,添加到一HashMap的全局变量subscribers中,同时返回此订阅。订阅创建即为创建一个subscription对象,在其中设置其属性subject,subjects,label其中subjects为一数组变量,存储subject中包含的所有主题。最后设置相关信息到responseEvent响应事件中,将相应事件封装到Command中。

Unsubscrib: 退订主题,将Map类型的subscribers中的订阅移除。

Join: 将请求的此session加入到SessionManager中的Map类型的sessions中,并设置sessionformate等信息。

Listen: 主要是确定订阅信息的转送方式mode,包含三种stream\pull\pollStream是发送一连串的事件(http长链接),即记录是不停止的。Pullpoll(客户端定时刷新服务端产即推送)是一个完整的记录返回的。配置文件中的listen.force.pull.all设置是否全部使用pull

Join-listen: joinlisten的集合。

Leave: 停止session,也即移除所有的订阅信息。

Hb: heartbear心跳,只是设置心跳事件给客户端。

Publish: 发布事件,若请求中有p_to(sessionid),即发送目标,将进行单播,否则为多播。单播时,将请求事件进行浅克隆,将克隆事件发送给目标事件的订阅者。onEvent()中若请求事件已激活、未超时且事件队列未满则加入事件队列,否则停止session。广播时调用上面介绍的apply()

 

// Send Event to subscriber.
		session.getSubscriber().onEvent((Event) event.clone());

 无论是哪种事件类型,最后都是将信息封装在responseEvent中,然后封装在Command中。

若请求事件为listenrefresh则只要存在连接就不断循环从队列中获取事件推送到客户端,不同的mode下进行不同的处理,添加心跳包到队列中等操作。在推送时应用到adapter模式,在command中有方法createClientAdapter,内容为根据开始设置的session中的format值来实例化不同的适配器类。包含jsjava序列化对象、xml等。其它的事件类型直接通过适配器推送。

 

pushlet服务端的代码其实就是一个中间过渡的过程,客户端传入请求事件及参数,服务端根据请求事件做一定的处理,再返回一个事件及传入的参数,客户端再根据返回的事件选择相应的回调函数获取参数进行操作,回调函数就由自己去实现。

相关资料:

http://blog.csdn.net/yxw246/article/details/2418255这个分析的很透彻

http://blog.csdn.net/anghlq/article/details/5869233