websocket 与 jmx和jms 协作工作来管理web项目

客官°小女子只卖身不卖艺 2022-08-21 06:47 52阅读 0赞

MX4J是一个开源JMX管理框架，支持JSR3 (JMX) 和JSR160 (JMX Remote API)。通过[spring][]将MX4J集成到我们的应用系统中，可以通过HTTP协议适配，能够基于Web的方式来实现对应用系统的监控和管理。这里，使用的版本分别为：

*  Spring 2.5
 *  MX4J 3.0.2

MX4J对应的几个JAR文件，加入到CLASSPATH：

*  mx4j.jar
 *  mx4j-impl.jar
 *  mx4j-jmx.jar
 *  mx4j-remote.jar
 *  mx4j-rimpl.jar
 *  mx4j-rjmx.jar
 *  mx4j-tools.jar

首先，给出Spring的完整配置，server.xml 配置文件如下所示：

<?xml version="1.0" encoding="UTF-8" ?>
    
    <beans xmlns="http://www.springframework.org/schema/beans"
    	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:util="http://www.springframework.org/schema/util"
    	xsi:schemaLocation="http://www.springframework.org/schema/beans
      http://www.springframework.org/schema/beans/spring-beans.xsd
      http://www.springframework.org/schema/util
      http://www.springframework.org/schema/util/spring-util-2.5.xsd">
    
    
    <bean id="carrierConfig" class="entity.Carrier" />
    
     <bean id="exporter" class="org.springframework.jmx.export.MBeanExporter">
    	<property name="assembler" ref="assembler" />
    	<property name="beans">
    		<map>
    			<entry key="mx4jServer:name=HttpAdaptor" value-ref="httpAdaptor"/>
    			<entry key="beanEntity:name=carrierMBeanManager" value-ref="carrierConfig"/>
    		</map>
    	</property>
    	<property name="server" ref="mbeanServer" />
    </bean>
    
    <bean id="jmxAttributeSource" class="org.springframework.jmx.export.annotation.AnnotationJmxAttributeSource" />
    
    <bean id="assembler" class="org.springframework.jmx.export.assembler.MetadataMBeanInfoAssembler">
    	<property name="attributeSource" ref="jmxAttributeSource" />
    </bean>
    
    
    
    
    
    
    
    
    
    <bean id="mbeanServer" class="org.springframework.jmx.support.MBeanServerFactoryBean"/>
    
    <bean id="httpAdaptor" class="mx4j.tools.adaptor.http.HttpAdaptor">
    	<property name="processor">
    		<bean id="xsltProcessor" class="mx4j.tools.adaptor.http.XSLTProcessor" />
    	</property>
    	<property name="port" value="8089"/> 
    	<property name="host" value="localhost"/> 
    </bean>
    
    </beans>

上面配置中，比较核心的是MX4J的HTTP适配器类：mx4j.tools.adaptor.http.HttpAdaptor，它有一个processor属性，使用MX4J已经实现的mx4j.tools.adaptor.http.XSLTProcessor来对其注入。XSLTProcessor实现了MX4J定义的XML转换成为浏览器可以解析的格式的文档，就Web网页。

Spring配置中通过实例化一个HttpAdaptor，注册到MBean Server中，实现基于HTTP协议远程访问管理和监控。org.shirdrn.jmx.mx4j.MyConfiguration我们定义的一个MBean，也注册到MBean Server中，实现JMX的管理与监控，代码如下所示：

package entity;
    
    import java.io.IOException;
    import java.io.Serializable;
    import java.nio.CharBuffer;
    import java.util.HashMap;
    import java.util.Map;
    
    import org.springframework.jmx.export.annotation.ManagedOperation;
    import org.springframework.jmx.export.annotation.ManagedOperationParameter;
    import org.springframework.jmx.export.annotation.ManagedOperationParameters;
    import org.springframework.jmx.export.annotation.ManagedResource;
    
    import utils.StartServerService;
    import webSocket.MyWebSocketBound;
    
    
    
    @ManagedResource(objectName = "bean:name=CarrierMBean")
    public class Carrier implements Serializable{
    	
    	private static final long serialVersionUID = -1490071251860434130L;
    	
    	
    	private Long carrierId;
     	private String carrierName;
     	private String serialNumber;
     	private String status;
     	private static Map<Integer,Carrier> map = new HashMap<Integer,Carrier>();
     	
     	
     	public Carrier(){
     		super();
     	}
     	
     	 public Carrier(long carrierId,String carrierName,String serialNumber,String status){
     		 this.carrierId = carrierId;
     		 this.carrierName = carrierName;
     		 this.serialNumber = serialNumber;
     		 this.status = status;
     	 }
     	
     	
     	/**
    	 * 
    	 * @param carrierName
    	 * @param serialNumber
    	 * @param status
    	 */
         @ManagedOperation(description = "To set carrierInfor ")
     	 @ManagedOperationParameters({
     	 @ManagedOperationParameter(name = "carrierId", description = "The carrier carrierId"),
     	 @ManagedOperationParameter(name = "carrierName", description = "The carrier carrierName"),
     	 @ManagedOperationParameter(name = "serialNumber", description = "The carrier serialNumber"),
     	 @ManagedOperationParameter(name = "status", description = "The carrier status")})
        	public void setCarrierInfor(long carrierId,String carrierName,String serialNumber,String status) {
        		this.carrierId = carrierId;
        		this.carrierName = carrierName;
        		this.serialNumber = serialNumber;
        		this.status = status;
        	}
     	 
     	 /**
     	  * 显示信息
     	  * @return
     	  */
        @ManagedOperation(description = "To show carrier ")
    	public String show() {
    	 StringBuffer sb = new StringBuffer().append("carrierId=").append(carrierId).append(  
                    ", carrierName=").append(carrierName).append(",serialNumber=").append(serialNumber)
                    .append(",status=").append(status);  
            return sb.toString();  
    	}
        
        
        /**
      	  * 发送信息，通过webSocket中间件
      	  */
      	 @ManagedOperation(description = "To send carrierInfor ")
    	 @ManagedOperationParameters({
    	 @ManagedOperationParameter(name = "carrierId", description = "The carrier carrierId"),
    	 @ManagedOperationParameter(name = "carrierName", description = "The carrier carrierName"),
    	 @ManagedOperationParameter(name = "serialNumber", description = "The carrier serialNumber"),
    	 @ManagedOperationParameter(name = "status", description = "The carrier status")})
      	 public void sendMessage(long carrierId,String carrierName,String serialNumber,String status){
    //  	String str = URLDecoder.decode(serialNumber, "UTF-8");
      		 
      		StringBuffer sb = new StringBuffer();
      		MyWebSocketBound test = new MyWebSocketBound();
      		map =  StartServerService.getCarrierMap();
      		for(int i = 1;i<= map.size();i++){
      			Carrier entity = map.get(i);
      			if(entity.getCarrierId() == carrierId){
      				entity.setCarrierId(carrierId);
      				if(!"".equals(carrierName) && null != carrierName){
      					entity.setCarrierName(carrierName);
      				}
      				if(!"".equals(serialNumber) && null != serialNumber){
      					entity.setSerialNumber(serialNumber);
      				}
      				if(!"".equals(status) && null != status){
      					entity.setStatus(status);
      				}
      				System.out.println("服务端修改信息"+carrierId+"的信息为："+entity);
       			}
      			sb.append(entity.toString());
      		}
      		 String sbTest = sb.toString().substring(0, sb.toString().length() - 1);
      	     CharBuffer buffer = CharBuffer.wrap(sbTest); 
      		try {
      			test.onTextMessage(buffer);
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
      	 }
      	 
      	 /**
      	  * 构造实体
      	  * @return
      	  */
    	public Map<Integer, Carrier> createEntity() {
    		for(int i=1;i<=10;i++){
    			Carrier entity = new Carrier(i, "车牌号"+i, "00"+i, "在线"+i);
    			map.put(i, entity);
    		}
    		return map;
    	}
     	/**
     	 * 按要求显示
     	 */
    	@Override
       	public String toString() {
       		 String test = carrierId + "," + carrierName + "," + serialNumber + "," + status + ";";
       		return test;
       }
    	
    	public Long getCarrierId() {
    		return carrierId;
    	}
    	public void setCarrierId(Long carrierId) {
    		this.carrierId = carrierId;
    	}
    	public String getCarrierName() {
    		return carrierName;
    	}
    	public void setCarrierName(String carrierName) {
    		this.carrierName = carrierName;
    	}
    	public String getSerialNumber() {
    		return serialNumber;
    	}
    	public void setSerialNumber(String serialNumber) {
    		this.serialNumber = serialNumber;
    	}
    	public String getStatus() {
    		return status;
    	}
    	public void setStatus(String status) {
    		this.status = status;
    	}
    
    	public static Map<Integer, Carrier> getMap() {
    		return map;
    	}
    
    	public static void setMap(Map<Integer, Carrier> map) {
    		Carrier.map = map;
    	}
    
    	
    	
    }

但是，Spring配置中的org.springframework.jmx.export.assembler.InterfaceBasedMBeanInfoAssembler对MBean暴露的操作进行了限制，主要是通过我们定义的接口org.shirdrn.jmx.mx4j.SelectedMethodsInterface来指定需要将MBean的哪些内容暴露给JMX管理，接口如下所示：（必须配置文件中  对接口有相应的配置  ）

package entity;
    
    import org.springframework.jmx.export.annotation.ManagedOperation;
    import org.springframework.jmx.export.annotation.ManagedOperationParameter;
    import org.springframework.jmx.export.annotation.ManagedOperationParameters;
    
    
    public interface SelectedMethodsInterface {
    	    public long getId();    
    	    public void setName(String name);  
    	    public String getName();  
    	    public void show();    
    }

最后，我们可以初始化Spring的IOC容器，启动MX4J的HttpAdaptor服务，实现通过Web控制台的JMX管理。

/\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*websock协作工作建立长连接实时改变对象信息\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*/

websocket  页面的调用test.jsp  代码如下：

<%@ page language="java" import="java.util.*" pageEncoding="utf-8"%>
    <%
    String path = request.getContextPath();
    String basePath = request.getScheme()+"://"+request.getServerName()+":"+request.getServerPort()+path+"/";
    %>
    
    <!DOCTYPE html>
    <html>
    <head>
     <base href="<%=basePath%>">
    <meta http-equiv="Content-Type" content="text/html; charset=utf-8">
    <title>WebSocket Test</title>
    
    <script type="text/javascript" src="js/jquery-1.8.3.min.js"></script>  
    <script type="text/javascript">
    var ws = null;
    function initWebSocket() {
    	if ('WebSocket' in window){
    		ws = new WebSocket("ws://127.0.0.1:8080/webSocket/mywebsocket.do");
    	}else{
    		alert("not support");
    	}
    	
    	ws.onmessage = function(evt) {
    		var mess = "";
    		var mss = "";
    		var ms = [];
    		var message = [];
     		var msg = evt.data;
    		message = msg.split(";");
    		for(var mege in message){
    			 mss = message[mege];
    			 ms = mss.split(",");
    			var result ="carrierId"+mege+":"+ "<input type='text' id=carrierId"+mege+" value='"+ms[0]+"' readonly='readonly'/>"
    			+ "carrierName"+mege+":"+ "<input type='text' id=carrierName"+mege+" value='"+ms[1]+"' readonly='readonly'/>"
    			+ "serialNumber"+mege+":"+ "<input type='text' id=serialNumber"+mege+" value='"+ms[2]+"' readonly='readonly'/>"
    			+ "status"+mege+":"+ "<input type='text' id=status"+mege+" value='"+ms[3]+"' readonly='readonly'/>";
    				
    	 		$("#receiveMessage").html(result +"</br>"+ mess);
    	 		mess = $("#receiveMessage").html();
    		}
    	};
    	
    	ws.onclose = function(evt) {
    		alert("close");
    	};
    	
    	ws.onopen = function(evt) {
    		alert("open");
    	};
    }
    
    function sendMsg() {
    	var carrierId = document.getElementById('carrierId').value;
    	var carrierName = document.getElementById('carrierName').value;
    	var serialNumber = document.getElementById('serialNumber').value;
    	var status = document.getElementById('status').value;
    	var sendmsg = carrierId +"," + carrierName +"," + serialNumber +"," + status;
    	ws.send(sendmsg);
    };
    </script>
    </head>
    <body οnlοad="initWebSocket();">
    	carrierId:<input type="text" id="carrierId"/>
    	carrierName:<input type="text" id="carrierName"/>
    	serialNumber:<input type="text" id="serialNumber"/>
    	status:<input type="text" id="status"/>
    <input type="button" value="send" οnclick="sendMsg()"></input>
    
    <div id="receiveMessage">
    	carrierId:<input type="text" id="carrierIds" readonly="readonly"/>
    	carrierName:<input type="text" id="carrierNames" readonly="readonly"/>
    	serialNumber:<input type="text" id="serialNumbers" readonly="readonly"/>
    	status:<input type="text" id="statuss" readonly="readonly"/>
    </div>
    </body>
    </html>

以下是websock工作所需要的类：

ApplicationContextListener.java  具体代码  如下：

package webSocket;
    
    import javax.servlet.ServletContext;
    import javax.servlet.ServletContextEvent;
    import javax.servlet.ServletContextListener;
    
    import utils.SpringBeanHolder;
    import utils.StartServerService;
    
    public class ApplicationContextListener implements ServletContextListener {
    
    	@Override
    	public void contextDestroyed(ServletContextEvent sce) {
    	}
    
    	@Override
    	public void contextInitialized(ServletContextEvent event) {
    
    		ServletContext servlet = event.getServletContext();
    		SpringBeanHolder.setContextOfWeb(servlet);
    		StartServerService.startHttpAdaptor();
    		StartServerService.getCarriers();
    	}
    
    }

InitServlet.java  具体代码  如下：

package webSocket;
    
    import java.util.ArrayList;
    import java.util.List;
    
    import javax.servlet.ServletConfig;
    import javax.servlet.ServletException;
    import javax.servlet.http.HttpServlet;
    
    import org.apache.catalina.websocket.MessageInbound;
    
    
    
    public class InitServlet extends HttpServlet {
    
    	private static final long serialVersionUID = -2895964351950418467L;
    
    	private static List<MessageInbound> socketList;
    
    	public void init(ServletConfig config) throws ServletException {
    		InitServlet.socketList = new ArrayList<MessageInbound>();
    		super.init(config);
    		System.out.println("socket init success..");
    	}
      
    	public static synchronized List<MessageInbound> getSocketList() {
    		return InitServlet.socketList;
    	}
    
    }

MyWebSocket.java  具体代码  如下：

package webSocket;
    
    import javax.servlet.http.HttpServletRequest;
    
    import org.apache.catalina.websocket.StreamInbound;
    import org.apache.catalina.websocket.WebSocketServlet;
    
    public class MyWebSocket extends WebSocketServlet {
    
    	private static final long serialVersionUID = 7780410787770617234L;
    
    	/**
    	 * 在这里初始化自定义的WebSocket连接对象
    	 */
    	@Override
    	protected StreamInbound createWebSocketInbound(String subProtocol, HttpServletRequest request) {
    		return new MyWebSocketBound();
    	}
    
    }

<pre name="code" class="java">MyWebSocketBound .java  具体代码  如下:

package webSocket;
    
    import java.io.IOException;
    import java.nio.ByteBuffer;
    import java.nio.CharBuffer;
    import java.util.HashMap;
    import java.util.Map;
    
    import org.apache.catalina.websocket.MessageInbound;
    import org.apache.catalina.websocket.WsOutbound;
    
    import utils.StartServerService;
    import entity.Carrier;
    
    public class MyWebSocketBound extends MessageInbound {
    
    	private StringBuffer buf = new StringBuffer();
    	private static Map<Integer, Carrier> entityMap = new HashMap<Integer, Carrier>();
    
    	public MyWebSocketBound() {
    		super();
    	}
    
    	/**
    	 * 打开连接
    	 */
    	@Override
    	public void onOpen(WsOutbound outbound) {
    		System.out.println("连接已打开");
    		super.onOpen(outbound);
    		InitServlet.getSocketList().add(this);
    		try {
    			entityMap = StartServerService.getCarrierMap();
    			for (int i = 1; i <= entityMap.size(); i++) {
    				Carrier entity = entityMap.get(i);
    				buf.append(entity.toString());
    			}
    			String buff = buf.toString().substring(0, buf.toString().length() - 1);
    			CharBuffer buffer = CharBuffer.wrap(buff);
    			this.getWsOutbound().writeTextMessage(buffer);
    
    		} catch (IOException e) {
    			System.out.println("连接异常");
    		}
    	}
    
    	/**
    	 * 关闭连接
    	 */
    	@Override
    	public void onClose(int status) {
    		System.out.println("连接已关闭");
    		InitServlet.getSocketList().remove(this);
    		super.onClose(status);
    	}
    
    	/**
    	 * 发送信息（发送字符串类型的信息）
    	 */
    	@Override
    	public void onTextMessage(CharBuffer msg) throws IOException {
    		String receiveMessage = msg.toString();
    		for (MessageInbound messageInbound : InitServlet.getSocketList()) {
    			CharBuffer buffer = CharBuffer.wrap(receiveMessage);
    			WsOutbound outbound = messageInbound.getWsOutbound();
    			outbound.writeTextMessage(buffer);
    			outbound.flush();
    		}
    	}
    
    	/**
    	 * 发送信息（发送二进制类型的文件）
    	 */
    	@Override
    	public void onBinaryMessage(ByteBuffer message) throws IOException {
    	}
    
    	public StringBuffer getBuf() {
    		return buf;
    	}
    
    	public void setBuf(StringBuffer buf) {
    		this.buf = buf;
    	}
    
    	public static Map<Integer, Carrier> getEntityMap() {
    		return entityMap;
    	}
    
    	public static void setEntityMap(Map<Integer, Carrier> entityMap) {
    		MyWebSocketBound.entityMap = entityMap;
    	}
    
    }

以下是工具类，Spring获得bean的工具类 SpringBeanHolder .java 具体代码如下：

package utils;
    
    import javax.servlet.ServletContext;
    
    import org.springframework.context.ApplicationContext;
    import org.springframework.web.context.support.WebApplicationContextUtils;
    
    public class SpringBeanHolder {
    
    	private static ApplicationContext context;
     	
    	public static void setContextOfWeb(ServletContext servlet) {
    		context = WebApplicationContextUtils
    				.getRequiredWebApplicationContext(servlet);
    	}
    	
    	
    
    	public static Object getBean(String beanName) {
    		return context.getBean(beanName);
    	}
    	
    }

以下是工具类，Spring获得bean的工具类StartServerService .java 具体代码如下：

package utils;
    
    import java.io.IOException;
    import java.util.HashMap;
    import java.util.Map;
    
    import mx4j.tools.adaptor.http.HttpAdaptor;
    import entity.Carrier;
    
    public class StartServerService {
    
    	private static Map<Integer,Carrier> carrierMap = new HashMap<Integer,Carrier>();
    	
    	private static StartServerService serverService = new StartServerService();
    	/**
    	 * 单例模式
    	 */
    	public StartServerService(){
    	}
    	
    	public static StartServerService getStartServerService(){
    		return serverService;
    	}
    	
    	/**
    	 * 开启MX4j服务
    	 */
    	public static void startHttpAdaptor() {
    		HttpAdaptor httpAdaptor = (HttpAdaptor)SpringBeanHolder.getBean("httpAdaptor");  
    		try {
            	httpAdaptor.start();
            	System.out.println("后台   mx4j Server 启动成功");
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    		
    	}
    	/**
    	 * 得到对象列表信息，放入application全局资源中
    	 */
    	public static void getCarriers() {
    		Carrier carrier = (Carrier)SpringBeanHolder.getBean("carrierConfig");  
    		carrierMap = carrier.createEntity();
    	}
    
    	public static Map<Integer, Carrier> getCarrierMap() {
    		return carrierMap;
    	}
    
    	public static void setCarrierMap(Map<Integer, Carrier> carrierMap) {
    		StartServerService.carrierMap = carrierMap;
    	}
    }

注意：要使用websocket  服务器可用[  
][Link 1]

[jetty][Link 1] 7.0.1 包含了一个初步的实现

resin 包含有websocket 实现

pywebsocket, apache http server 扩展

apache tomcat 7.0.27 版本

Nginx 1.3.13 版本

jWebSocket java实现版  
  
现在开启服务，以下是效果图  
![Center][]

![Center 1][]

![Center 2][]

用浏览器打开  两个分开页面  websocket 建立长连接  实时改变 对象的属性信息：

![Center 3][]

[spring]: http://lib.csdn.net/base/17
[Link 1]: http://baike.baidu.com/view/1425008.htm
[Center]: /images/20220724/b3bfffc14da0455db2e21ba4936c5a45.png
[Center 1]: /images/20220724/68fe4c40313b478c85b0c563470cfaf0.png
[Center 2]: /images/20220724/9858e52f4112463488e9370d3de5e4c7.png
[Center 3]: /images/20220724/93b59b1076e643fe9f036ab2db872480.png