【pushlet学习】具体实战

业务需求：
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；
  1. package com.guoguo;
  3. import java.io.BufferedInputStream;
  4. import java.io.IOException;
  5. import java.io.InputStream;
  6. import java.util.ArrayList;
  8. import javax.servlet.ServletException;
  9. import javax.servlet.http.HttpServlet;
  10. import javax.servlet.http.HttpServletRequest;
  11. import javax.servlet.http.HttpServletResponse;
  13. import org.stringtree.json.JSONReader;
  14. import org.stringtree.json.JSONValidatingReader;
  16. public class TestServlet extends HttpServlet {
  17. 	private static final long serialVersionUID = 1L;
  19. 	public TestServlet() {
  20. 		super();
  21. 	}
  23. 	/**
  24. 	 * get/post方法的处理函数
  25. 	 */
  26. 	protected void service(HttpServletRequest request,
  27. 			HttpServletResponse response) throws ServletException, IOException {
  28. 		
  29. 		// 读取请求报文数据
  30. 		request.setCharacterEncoding("UTF-8");
  32. 		// 获取请求的数据
  33. 		StringBuffer reqData = new StringBuffer();
  34. 		InputStream in = request.getInputStream();
  35. 		BufferedInputStream buf = new BufferedInputStream(in);
  36. 		byte[] buffer = new byte[1024];
  37. 		int iRead;
  38. 		while ((iRead = buf.read(buffer)) != -1) {
  39. 			reqData.append(new String(buffer, 0, iRead, "UTF-8"));
  40. 		}
  42. 		// 获取请求的测点名称数组
  43. 		JSONReader r = new JSONValidatingReader();
  44. 		@SuppressWarnings("unchecked")
  45. 		ArrayList<Object> keyList = (ArrayList<Object>) r.read(reqData
  46. 				.toString());
  48. 		// 获取订阅主题名称
  49. 		String aSubject = request.getParameter("subject");
  50. 		System.out.println("请求的测点：" + reqData.toString() + " , 主题名：" + aSubject);
  52. 		// 启动一个线程，实现创建Pushlet事件、做业务、向前台推送数据等功能
  53. 		PushThread pushThread = new PushThread(aSubject, keyList);
  54. 		pushThread.start();
  55. 	}
  56. }
PushThread.java
主要功能：
1. 这是一个Thread类，该类有两个属性：主题String aSubject和关键字列表 ArrayList<Object> keyList；
2. 线程首先根据主题名创建事件源：Event event = Event.createDataEvent(aSubject);
3. 线程运行的时候，会监测会话状态以及事件订阅情况，并基于此判断线程是否需要结束。
若session关闭或浏览器关闭，则线程退出；
若有会话订阅该事件源，线程则进行业务处理，处理过程如下：
  1. 获取各测点的值；
  2. 将各测点的值组装成字符串；
  3. 将该字符串设置为事件源的属性。
4.然后以广播的形式将事件发送出去，Dispatcher.getInstance().multicast(event);
5. 经测试：Thread的run()函数执行结束后，Thread就自动退出了，系统会释放该Thread的资源。
  1. package com.guoguo;
  3. import java.text.SimpleDateFormat;
  4. import java.util.ArrayList;
  5. import java.util.Date;
  6. import java.util.HashMap;
  7. import java.util.Random;
  9. import org.stringtree.json.JSONValidatingWriter;
  11. import nl.justobjects.pushlet.core.Dispatcher;
  12. import nl.justobjects.pushlet.core.Event;
  13. import nl.justobjects.pushlet.core.Session;
  14. import nl.justobjects.pushlet.core.SessionManager;
  16. public class PushThread extends Thread {
  17. 	// 主题
  18. 	public String aSubject; // 客户端传递过来
  19. 	// 关键字列表
  20. 	public ArrayList<Object> keyList; // 客户端传递过来
  22. 	/**
  23. 	 * 构造函数
  24. 	 * 
  25. 	 * @param aSubject
  26. 	 * @param keyList
  27. 	 */
  28. 	public PushThread(String aSubject, ArrayList<Object> keyList) {
  29. 		this.aSubject = aSubject;
  30. 		this.keyList = keyList;
  31. 	}
  33. 	@Override
  34. 	public void run() {
  35. 		Event event = Event.createDataEvent(aSubject);
  36. 		int i = 0;
  37. 		while (true) {
  38. 			try {
  39. 				Thread.sleep(5000);
  40. 			} catch (InterruptedException e) {
  41. 				// 线程阻塞，结束线程
  42. 				System.out.println("=========>sleep异常 --->" + "线程"
  43. 						+ Thread.currentThread().getId() + "关闭");
  44. 				break;
  45. 			}
  46. 			System.out.println("
-----Thread ID: "
  47. 					+ Thread.currentThread().getId());
  48. 			// 判断当前连接的会话个数，没有会话，则线程退出
  49. 			Session[] sessions = SessionManager.getInstance().getSessions();
  50. 			// 当前无会话，结束线程
  51. 			if (0 == sessions.length) {
  52. 				System.out.println("=========>无sessions --->" + "线程"
  53. 						+ Thread.currentThread().getId() + "关闭");
  54. 				break;
  55. 			}
  57. 			// 判断当前会话中是否存在订阅该主题的订阅者，不存在则结束线程
  58. 			boolean if_exist_subscriber = true;
  59. 			// 遍历所有session
  60. 			for (int j = 0; j < sessions.length; j++) {
  61. 				System.out
  62. 						.println(sessions[j].getSubscriber().match(event) == null ? "Session"
  63. 								+ j + ": 未订阅该事件 "
  64. 								: "Session" + j + "：订阅了该事件 ");
  66. 				if (null != sessions[j].getSubscriber().match(event)) {
  67. 					if_exist_subscriber = false;
  68. 				}
  69. 			}
  70. 			if (if_exist_subscriber) {
  71. 				System.out.println("=========>无"+aSubject+"订阅者 --->" + "线程"	+ Thread.currentThread().getId() + "关闭");
  72. 				break;
  73. 			}
  75. 			// 模拟业务处理：获取各测点的值
  76. 			HashMap<Object, Object> ret_value = new HashMap<Object, Object>();
  77. 			for (Object keyStr : keyList) {
  78. 				SimpleDateFormat df = new SimpleDateFormat("HH:mm:ss");// 设置日期格式
  79. 				String currTm = df.format(new Date());
  80. 				ret_value.put(keyStr, currTm);
  81. 				// ret_value.put(keyStr, (10*(new Random().nextFloat())));
  83. 			}
  85. 			// 将返回值封装为json数据形式
  86. 			String ret_string = "[";
  87. 			ret_string += new JSONValidatingWriter().write(ret_value);
  88. 			ret_string += "]";
  89. 			event.setField("key1", ret_string);
  91. 			// 推送消息
  92. 			Dispatcher.getInstance().multicast(event); // 向所有和event名称匹配的事件推送
  93. 		}
  95. 	}
  97. }

sources.properties 本测试程序不需要在该文件中配置任何东西
  1. # 
  2. # Properties file for EventSource objects to be instantiated.
  3. #
  4. # Place this file in the CLASSPATH (e.g. WEB-INF/classes) or directly under WEB-INF.
  5. #
  6. # $Id: sources.properties,v 1.2 2007/11/10 14:12:16 justb Exp $
  7. #
  8. # Each EventSource is defined as <key>=<classname>
  9. # 1. <key> should be unique within this file but may be any name
  10. # 2. <classname> is the full class name
  11. #
  12. #
  13. # Define Pull Sources here. These classes must be derived from
  14. # nl.justobjects.pushlet.core.EventPullSource
  15. # Inner classes are separated with a $ sign from the outer class. 
  16. source1=nl.justobjects.pushlet.test.TestEventPullSources$TemperatureEventPullSource
  17. source2=nl.justobjects.pushlet.test.TestEventPullSources$SystemStatusEventPullSource
  18. source3=nl.justobjects.pushlet.test.TestEventPullSources$PushletStatusEventPullSource
  19. source4=nl.justobjects.pushlet.test.TestEventPullSources$AEXStocksEventPullSource
  20. source5=nl.justobjects.pushlet.test.TestEventPullSources$WebPresentationEventPullSource
  21. source6=nl.justobjects.pushlet.test.TestEventPullSources$PingEventPullSource
  22. source7=nl.justobjects.pushlet.test.TestEventPullSources$MyEventPullSource
  23. source8=nl.justobjects.pushlet.test.TestEventPullSources$SpEventPullSource
  24. # TO BE DONE IN NEXT VERSION
  25. # define Push Sources here. These must implement the interface
  26. # nl.justobjects.pushlet.core.EventSource

web.xml
这里主要是配置servlet：TestServlet，其他servlet用不到
  1. <?xml version="1.0" encoding="UTF-8"?>
  3. <web-app>
  4. 	<!-- Define the pushlet servlet -->
  5. 	<servlet>
  6. 		<servlet-name>pushlet</servlet-name>
  7. 		<servlet-class>nl.justobjects.pushlet.servlet.Pushlet</servlet-class>
  8. 		<load-on-startup>1</load-on-startup>
  9. 	</servlet>
  10. 	<servlet-mapping>
  11. 		<servlet-name>pushlet</servlet-name>
  12. 		<url-pattern>/pushlet.srv</url-pattern>
  13. 	</servlet-mapping>
  16. 	<servlet>
  17. 		<display-name>ChatServlet</display-name>
  18. 		<servlet-name>ChatServlet</servlet-name>
  19. 		<servlet-class>com.guoguo.ChatServlet</servlet-class>
  20. 	</servlet>
  21. 	<servlet-mapping>
  22. 		<servlet-name>ChatServlet</servlet-name>
  23. 		<url-pattern>/ChatServlet</url-pattern>
  24. 	</servlet-mapping>
  26. 	<servlet>
  27. 		<display-name>TestServlet</display-name>
  28. 		<servlet-name>TestServlet</servlet-name>
  29. 		<servlet-class>com.guoguo.TestServlet</servlet-class>
  30. 	</servlet>
  31. 	<servlet-mapping>
  32. 		<servlet-name>TestServlet</servlet-name>
  33. 		<url-pattern>/TestServlet</url-pattern>
  34. 	</servlet-mapping>
  36. </web-app>

receive.jsp
前台界面，主要功能如下：
1. 负责生成随机主题名和测点集合；
2. 页面初始化时，将主题名和测点集合发送到后台，并开起pushlet监听；
PL._init();
PL.joinListen(aSubject);
3. 编写onData()函数，用于处理“服务器推”送到前台的数据，并显示在页面上；
4. 同时编写了“取消订阅”按钮，点击时，前台主动“取消订阅”，后台服务器接收到后，就会释放与此对应的Thread。
  1. <%@ page language="java" import="java.util.*" pageEncoding="UTF-8"%>
  2. <%
  3. 	String path = request.getContextPath();
  4. 	String basePath = request.getScheme() + "://"
  5. 			+ request.getServerName() + ":" + request.getServerPort()
  6. 			+ path + "/";
  7. %>
  9. <!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN">
  10. <html>
  11. <head>
  12. <base href="<%=basePath%>">
  14. <meta http-equiv="pragma" content="no-cache">
  15. <meta http-equiv="cache-control" content="no-cache">
  16. <meta http-equiv="expires" content="0">
  17. <meta http-equiv="keywords" content="keyword1,keyword2,keyword3">
  18. <meta http-equiv="description" content="This is my page">
  19. <script type="text/javascript"
  20. 	src="<%=basePath%>js/pushlet_js/ajax-pushlet-client.js"></script>
  21. <script type="text/javascript">	   
  22. 		   var subscriptionId = null;
  23. 		   window.onload = onInit;
  24. 		   window.onbeforeunload = onUnsubscribe;
  25. 		   
  26. 		   // 监听后台返回的数据信息，更新页面
  27. 		   function onData(event) {
  28. 		   		// 保存订阅编号，用于页面关闭时进行退订
  29.            		subscriptionId = event.get('p_sid');
  30.            		
  31.            		// 更新页面
  32.                 document.dataEventDisplay.event.value = event.get("key1");
  33.                 // ------实际案例处理（test）---------
  34.                /*  var respData = decodeURIComponent(event.get("key1"));
  35.                 var respObj = eval(respData);
  36.                 //var respActionNum = respObj.length;
  37.                 var obj = respObj[0];
  38.                 var str = "";  
  39. 			    for(var p in obj){  
  40. 			      if(typeof(obj[p]) != "function"){  
  41. 			        str += p + "=" + obj[p] + ", " ;
  42. 		          }  
  43. 			    }  
  44. 			    alert(str);     */             
  45.            }
  46.            
  47.            // 页面关闭时,取消订阅
  48. 		   function onUnsubscribe() {
  49. 		   		if (subscriptionId != null) {
  50.            			PL.unsubscribe(subscriptionId);
  51. 			    }
  52. 		   }
  53. 		   
  54. 	   // 页面加载完，初始化请求、监听
  55. 		function onInit() {
  56. 		  var aSubject = _getRandomString(6);          //主题名
  57. 		  var httpRequest   = getXMLHttpRequest();
  58. 		  if (httpRequest) {
  59. 			var reqData = getData();
  60. 			httpRequest.onreadystatechange = function() {
  61. 			    if (httpRequest.readyState == 4) {
  62. 			    	if (httpRequest.status == 200) {
  63. 			    		// 请求成功，起pushlet监听
  64. 			    		PL._init(); 
  65. 			    		PL.joinListen(aSubject);
  66. 				    } else {
  67. 				    	alert("实时请求失败!
" + httpRequest.statusText);
  68. 				    }
  69. 			    }
  70. 		    }
  71. 			url = '<%=request.getContextPath()%>' + '/TestServlet'
  72. 				+ '?subject=' + aSubject;
  73. 		    httpRequest.open("POST", url, true);
  74. 		    httpRequest.send(reqData);
  75. 	     }
  76.    	   }
  78. 	// 请求关键字
  79. 	function getData() {
  80. 		var reqData = "[30200000001010, 30200000001012, 30800000003009, 30800000006009]";
  81. 		return reqData;
  82. 	}
  84. 	// 获取http请求
  85. 	function getXMLHttpRequest() {
  86. 		req = false;
  87. 		//本地XMLHttpRequest对象
  88. 		if (window.XMLHttpRequest) {
  89. 			try {
  90. 				req = new XMLHttpRequest();
  91. 			} catch (e) {
  92. 				req = false;
  93. 			}
  94. 			//IE/Windows ActiveX版本
  95. 		} else if (window.ActiveXObject) {
  96. 			try {
  97. 				req = new ActiveXObject("Msxml2.XMLHTTP");
  98. 			} catch (e) {
  99. 				try {
  100. 					req = new ActiveXObject("Microsoft.XMLHTTP");
  101. 				} catch (e) {
  102. 					req = false;
  103. 				}
  104. 			}
  105. 		}
  106. 		return req;
  107. 	}
  109. 	// 获取长度为len的随机字符串
  110. 	function _getRandomString(len) {
  111. 		var len = len || 32;
  112. 		var chars = 'ABCDEFGHJKMNPQRSTWXYZabcdefhijkmnprstwxyz';
  113. 		var maxPos = chars.length;
  114. 		var pwd = '';
  115. 		for (i = 0; i < len; i++) {
  116. 			pwd += chars.charAt(Math.floor(Math.random() * maxPos));
  117. 		}
  118. 		return pwd;
  119. 	}
  120. </script>
  121. </head>
  123. <body>
  124. 	<form name="dataEventDisplay">
  125. 		<table border="2" bordercolor="white" cellpadding="0" cellspacing="0">
  126. 			<tr>
  127. 				<td><textarea cols="60" rows="10" name="event">没有消息   </textarea></td>
  128. 			</tr>
  129. 		</table>
  130. 	</form>
  131. 	<button onclick="onUnsubscribe()">取消订阅</button>
  132. </body>
  133. </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的个数会减少到当前打开的界面个数。
这个还没有研究出为什么，需要继续研究。






附件列表

    【推广】 免费学中医，健康全家人
    原文地址：https://www.cnblogs.com/ssslinppp/p/4480889.html