zoukankan      html  css  js  c++  java

  • 【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的个数会减少到当前打开的界面个数。
    这个还没有研究出为什么，需要继续研究。






    附件列表
    • 相关阅读:
      Celery框架
      Tensorflow安装记录
      OpenFace的一些了解
      CentOS+Uwsgi+Nginx发布Flask开发的WebAPI
      Always On 集群监听创建失败问题
      SQL Server 安装好后 Always On群组配置
      Sql server 2016 Always On 搭建Windows集群配置
      汕头市队赛 SRM13 T3
      bzoj 1314: River过河 树套树+单调队列
      hdu 6119 …&&百度之星 T6
    • 原文地址：https://www.cnblogs.com/ssslinppp/p/4480889.html
    Copyright © 2011-2022 走看看