业务需求：
1. 前端界面需要实时显示空调、照明等设备的状态， 如：空调电压、空调电流、光照强度等，这些量每一个称作一个测点；
2. 不同的用户登录系统后，用户只能看到自己设备的运行状态，而看不到其他人设备的运行状态；
3. 由于每个用户的设备类型、种类、个数等都不相同，因此每个用户需要查询测点也不相同；
4. 当多个用户同时登陆系统时，其实就是在多个浏览器上打开多个浏览界面，去查看自己设备运行状态，
即：多个浏览器上的多个界面对后台请求的测点是不同的，例如：
用户1：<测点1，测点2，测点3，测点4，.....>；
用户2：<测点21，测点22，测点23，测点24，.....>；
用户3：<测点31，测点32，测点33，测点34，.....>；
用户n：<测点n1，测点n2，测点n3，测点n4，.....>；
功能需求：
采用传统的“请求/响应”方式，很难达到前台界面实时显示最新数据，为达到实时显示最新数据，我们采用一种“服务器推”的技术comet，而pushlet是“服务器推”技术的一种实现，这里我们采用pushlet技术来实现上述业务需求。
1. 假设后台的测点数据是实时变化的，且保存在HashMap中，即：HashMap<测点名，测点值>；当测点值发生变化时，后台就会实时更新Hashmap中对应的测点值，即Hashmap中保存的始终是实时数据；
2. 每隔固定的分钟数（如：5分钟），后台就会向前台推送最新数据，前台界面实时更新显示；
具体实现：
1. 前台界面打开时，会将测点名称集合以及主题名传递到后台，格式形如：{[测点1，测点2，测点3，....]，subject}；并在前台开启对此主题的监听;（注：主题名是动态随机生成的，每个界面的主题名都保证不相同）
2. 后台解析从前台传递来的{测点名称集合+主题名}，并根据主题名创建响应的事件源（EventPollSource）；需要说明的是：每个事件源（主题名）对应一个Thread，即前端打开m个界面，后台就会有m个对应的Thread，每个Thread处理一个事件源。
3. 当前端浏览器界面关闭时，后台能检测到关闭的界面对应的事件源（Thread），并将对应的Thread释放掉。
示例程序
环境如下：
TestServlet.java
主要功能：
1. 获取前台传递过来的测点名称数组ArrayList<Object> keyList和主题名称aSubject，
2. 根据测点名称数组和主题名称开启一个新Thread，在Thread中处理业务；
PushThread pushThread = new PushThread(aSubject, keyList);
3. 每个session（或界面）对应一个Thread；
package com.guoguo;import java.io.BufferedInputStream;import java.io.IOException;import java.io.InputStream;import java.util.ArrayList;import javax.servlet.ServletException;import javax.servlet.http.HttpServlet;import javax.servlet.http.HttpServletRequest;import javax.servlet.http.HttpServletResponse;import org.stringtree.json.JSONReader;import org.stringtree.json.JSONValidatingReader;public class TestServlet extends HttpServlet {private static final long serialVersionUID = 1L;public TestServlet() {super();}/*** get/post方法的处理函数*/protected void service(HttpServletRequest request,HttpServletResponse response) throws ServletException, IOException {// 读取请求报文数据request.setCharacterEncoding("UTF-8");// 获取请求的数据StringBuffer reqData = new StringBuffer();InputStream in = request.getInputStream();BufferedInputStream buf = new BufferedInputStream(in);byte[] buffer = new byte[1024];int iRead;while ((iRead = buf.read(buffer)) != -1) {reqData.append(new String(buffer, 0, iRead, "UTF-8"));}// 获取请求的测点名称数组JSONReader r = new JSONValidatingReader();@SuppressWarnings("unchecked")ArrayList<Object> keyList = (ArrayList<Object>) r.read(reqData.toString());// 获取订阅主题名称String aSubject = request.getParameter("subject");System.out.println("请求的测点：" + reqData.toString() + " , 主题名：" + aSubject);// 启动一个线程，实现创建Pushlet事件、做业务、向前台推送数据等功能PushThread pushThread = new PushThread(aSubject, keyList);pushThread.start();}}
PushThread.java
主要功能：
1. 这是一个Thread类，该类有两个属性：主题String aSubject和关键字列表 ArrayList<Object> keyList；
2. 线程首先根据主题名创建事件源：Event event = Event.createDataEvent(aSubject);
3. 线程运行的时候，会监测会话状态以及事件订阅情况，并基于此判断线程是否需要结束。
若session关闭或浏览器关闭，则线程退出；
若有会话订阅该事件源，线程则进行业务处理，处理过程如下：
- 获取各测点的值；
- 将各测点的值组装成字符串；
- 将该字符串设置为事件源的属性。
4.然后以广播的形式将事件发送出去，Dispatcher.getInstance().multicast(event);
5. 经测试：Thread的run()函数执行结束后，Thread就自动退出了，系统会释放该Thread的资源。
package com.guoguo;import java.text.SimpleDateFormat;import java.util.ArrayList;import java.util.Date;import java.util.HashMap;import java.util.Random;import org.stringtree.json.JSONValidatingWriter;import nl.justobjects.pushlet.core.Dispatcher;import nl.justobjects.pushlet.core.Event;import nl.justobjects.pushlet.core.Session;import nl.justobjects.pushlet.core.SessionManager;public class PushThread extends Thread {// 主题public String aSubject; // 客户端传递过来// 关键字列表public ArrayList<Object> keyList; // 客户端传递过来/*** 构造函数** @param aSubject* @param keyList*/public PushThread(String aSubject, ArrayList<Object> keyList) {this.aSubject = aSubject;this.keyList = keyList;}@Overridepublic void run() {Event event = Event.createDataEvent(aSubject);int i = 0;while (true) {try {Thread.sleep(5000);} catch (InterruptedException e) {// 线程阻塞，结束线程System.out.println("=========>sleep异常 --->" + "线程"+ Thread.currentThread().getId() + "关闭");break;}System.out.println("\n-----Thread ID: "+ Thread.currentThread().getId());// 判断当前连接的会话个数，没有会话，则线程退出Session[] sessions = SessionManager.getInstance().getSessions();// 当前无会话，结束线程if (0 == sessions.length) {System.out.println("=========>无sessions --->" + "线程"+ Thread.currentThread().getId() + "关闭");break;}// 判断当前会话中是否存在订阅该主题的订阅者，不存在则结束线程boolean if_exist_subscriber = true;// 遍历所有sessionfor (int j = 0; j < sessions.length; j++) {System.out.println(sessions[j].getSubscriber().match(event) == null ? "Session"+ j + ": 未订阅该事件 ": "Session" + j + "：订阅了该事件 ");if (null != sessions[j].getSubscriber().match(event)) {if_exist_subscriber = false;}}if (if_exist_subscriber) {System.out.println("=========>无"+aSubject+"订阅者 --->" + "线程" + Thread.currentThread().getId() + "关闭");break;}// 模拟业务处理：获取各测点的值HashMap<Object, Object> ret_value = new HashMap<Object, Object>();for (Object keyStr : keyList) {SimpleDateFormat df = new SimpleDateFormat("HH:mm:ss");// 设置日期格式String currTm = df.format(new Date());ret_value.put(keyStr, currTm);// ret_value.put(keyStr, (10*(new Random().nextFloat())));}// 将返回值封装为json数据形式String ret_string = "[";ret_string += new JSONValidatingWriter().write(ret_value);ret_string += "]";event.setField("key1", ret_string);// 推送消息Dispatcher.getInstance().multicast(event); // 向所有和event名称匹配的事件推送}}}
sources.properties 本测试程序不需要在该文件中配置任何东西
## Properties file for EventSource objects to be instantiated.## Place this file in the CLASSPATH (e.g. WEB-INF/classes) or directly under WEB-INF.## $Id: sources.properties,v 1.2 2007/11/10 14:12:16 justb Exp $## Each EventSource is defined as <key>=<classname># 1. <key> should be unique within this file but may be any name# 2. <classname> is the full class name### Define Pull Sources here. These classes must be derived from# nl.justobjects.pushlet.core.EventPullSource# Inner classes are separated with a $ sign from the outer class.source1=nl.justobjects.pushlet.test.TestEventPullSources$TemperatureEventPullSourcesource2=nl.justobjects.pushlet.test.TestEventPullSources$SystemStatusEventPullSourcesource3=nl.justobjects.pushlet.test.TestEventPullSources$PushletStatusEventPullSourcesource4=nl.justobjects.pushlet.test.TestEventPullSources$AEXStocksEventPullSourcesource5=nl.justobjects.pushlet.test.TestEventPullSources$WebPresentationEventPullSourcesource6=nl.justobjects.pushlet.test.TestEventPullSources$PingEventPullSourcesource7=nl.justobjects.pushlet.test.TestEventPullSources$MyEventPullSourcesource8=nl.justobjects.pushlet.test.TestEventPullSources$SpEventPullSource# TO BE DONE IN NEXT VERSION# define Push Sources here. These must implement the interface# nl.justobjects.pushlet.core.EventSource
web.xml
这里主要是配置servlet：TestServlet，其他servlet用不到
<?xml version="1.0" encoding="UTF-8"?><web-app><!-- Define the pushlet servlet --><servlet><servlet-name>pushlet</servlet-name><servlet-class>nl.justobjects.pushlet.servlet.Pushlet</servlet-class><load-on-startup>1</load-on-startup></servlet><servlet-mapping><servlet-name>pushlet</servlet-name><url-pattern>/pushlet.srv</url-pattern></servlet-mapping><servlet><display-name>ChatServlet</display-name><servlet-name>ChatServlet</servlet-name><servlet-class>com.guoguo.ChatServlet</servlet-class></servlet><servlet-mapping><servlet-name>ChatServlet</servlet-name><url-pattern>/ChatServlet</url-pattern></servlet-mapping><servlet><display-name>TestServlet</display-name><servlet-name>TestServlet</servlet-name><servlet-class>com.guoguo.TestServlet</servlet-class></servlet><servlet-mapping><servlet-name>TestServlet</servlet-name><url-pattern>/TestServlet</url-pattern></servlet-mapping></web-app>
receive.jsp
前台界面，主要功能如下：
1. 负责生成随机主题名和测点集合；
2. 页面初始化时，将主题名和测点集合发送到后台，并开起pushlet监听；
PL._init();
PL.joinListen(aSubject);
3. 编写onData()函数，用于处理“服务器推”送到前台的数据，并显示在页面上；
4. 同时编写了“取消订阅”按钮，点击时，前台主动“取消订阅”，后台服务器接收到后，就会释放与此对应的Thread。
<%@ page language="java" import="java.util.*" pageEncoding="UTF-8"%><%String path = request.getContextPath();String basePath = request.getScheme() + "://"+ request.getServerName() + ":" + request.getServerPort()+ path + "/";%><!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN"><html><head><base href="<%=basePath%>"><meta http-equiv="pragma" content="no-cache"><meta http-equiv="cache-control" content="no-cache"><meta http-equiv="expires" content="0"><meta http-equiv="keywords" content="keyword1,keyword2,keyword3"><meta http-equiv="description" content="This is my page"><script type="text/javascript"src="<%=basePath%>js/pushlet_js/ajax-pushlet-client.js"></script><script type="text/javascript">var subscriptionId = null;window.onload = onInit;window.onbeforeunload = onUnsubscribe;// 监听后台返回的数据信息，更新页面function onData(event) {// 保存订阅编号，用于页面关闭时进行退订subscriptionId = event.get(‘p_sid‘);// 更新页面document.dataEventDisplay.event.value = event.get("key1");// ------实际案例处理（test）---------/* var respData = decodeURIComponent(event.get("key1"));var respObj = eval(respData);//var respActionNum = respObj.length;var obj = respObj[0];var str = "";for(var p in obj){if(typeof(obj[p]) != "function"){str += p + "=" + obj[p] + ", " ;}}alert(str); */}// 页面关闭时,取消订阅function onUnsubscribe() {if (subscriptionId != null) {PL.unsubscribe(subscriptionId);}}// 页面加载完，初始化请求、监听function onInit() {var aSubject = _getRandomString(6); //主题名var httpRequest = getXMLHttpRequest();if (httpRequest) {var reqData = getData();httpRequest.onreadystatechange = function() {if (httpRequest.readyState == 4) {if (httpRequest.status == 200) {// 请求成功，起pushlet监听PL._init();PL.joinListen(aSubject);} else {alert("实时请求失败!\n" + httpRequest.statusText);}}}url = ‘<%=request.getContextPath()%>‘ + ‘/TestServlet‘+ ‘?subject=‘ + aSubject;httpRequest.open("POST", url, true);httpRequest.send(reqData);}}// 请求关键字function getData() {var reqData = "[30200000001010, 30200000001012, 30800000003009, 30800000006009]";return reqData;}// 获取http请求function getXMLHttpRequest() {req = false;//本地XMLHttpRequest对象if (window.XMLHttpRequest) {try {req = new XMLHttpRequest();} catch (e) {req = false;}//IE/Windows ActiveX版本} else if (window.ActiveXObject) {try {req = new ActiveXObject("Msxml2.XMLHTTP");} catch (e) {try {req = new ActiveXObject("Microsoft.XMLHTTP");} catch (e) {req = false;}}}return req;}// 获取长度为len的随机字符串function _getRandomString(len) {var len = len || 32;var chars = ‘ABCDEFGHJKMNPQRSTWXYZabcdefhijkmnprstwxyz‘;var maxPos = chars.length;var pwd = ‘‘;for (i = 0; i < len; i++) {pwd += chars.charAt(Math.floor(Math.random() * maxPos));}return pwd;}</script></head><body><form name="dataEventDisplay"><table border="2" bordercolor="white" cellpadding="0" cellspacing="0"><tr><td><textarea cols="60" rows="10" name="event">没有消息 </textarea></td></tr></table></form><button onclick="onUnsubscribe()">取消订阅</button></body></html>
测试结果分析：
运行程序，浏览器中输入：http://localhost:8080/pushletTest/receive.jsp
如下是运行界面：（只开启了一个session，即一个界面）
这是前台显示的界面：这些数据是服务器实时推送过来的。
当在前端打开多个不同的界面，如3个：
前台打开了3个界面，后台会自动开启3个对应的Thread，在各个Thread中分别处理每个session对应的业务。
我们来看一下后台的线程数：
我们再开启一个界面，现在总共有4个界面开启，见下图：
看一下对应的后台：
从上图可以看到，总共有4个线程在运行，对应4个session；在看一下线程数：
与前一个对比：
能看到线程由47个变为了48个。
接下来我们关掉其中的3个界面，只留下一个界面，按照之前的分析，应该只剩下1个Thread、1个session，
且后台的线程数应该由48个变为45个，下面我们看下截图：（现在只有一个界面留下，其他3个都关闭了）：
后台的javaw.exe的线程数还是48个；
此时前台浏览器关闭了，但是Thread并没有立即关闭；
等待一会，大约半分钟（等待时间不会太长），再次出现下面的界面：
看一下后台：
从上面的分析可以看到，前台界面关闭后，稍后延时一会，后台会将其对应的Thread释放掉。
（多次测试后，发现关闭浏览器界面后，很快（大约2~3）后台线程j就释放掉了）
如果不关闭界面，而是直接点击“取消订阅”按键，会发现后台会立即将其对应的Thread释放掉。（这个已测试过）；
所以，为了在关闭浏览器界面时，立即释放掉后台对应的Thread，我们可以在页面关闭时，自动执行“取消订阅”函数，
这样，在每次关闭浏览器界面时，后台检测到就会立即释放对应的线程。
还有一点需要注意，Thread释放了，但是我们发现session的个数并没有立即减少；
继续等待，大约几分钟过后，session的个数会减少到当前打开的界面个数。
这个还没有研究出为什么，需要继续研究。
附件列表
时间： 2024-11-02 12:50:56