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

时间：2016-04-22 19:41:43 阅读：240 评论：0 收藏：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="assembler"  class="org.springframework.jmx.export.assembler.InterfaceBasedMBeanInfoAssembler">   -->
<!--         <property name="managedInterfaces">   -->
<!--             <list>   -->
<!--                 <value>entity.SelectedMethodsInterface</value>   -->
<!--             </list>   -->
<!--        </property>   -->
<!--  </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 onload="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" onclick="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 服务器可用

jetty 7.0.1 包含了一个初步的实现

resin 包含有websocket 实现

pywebsocket, apache http server 扩展

apache tomcat 7.0.27 版本

Nginx 1.3.13 版本

jWebSocket java实现版

现在开启服务，以下是效果图
技术分享