mina整合spring ，服务端反向发送消息到客户端完整实例

拼搏现实的明天。 2022-08-13 14:49 281阅读 0赞

之前的项目需要用到mina，实现的功能主要是：服务端主动发送消息到客户端，这个的服务端为外网的tomcat，客户端为内网的tomcat，由于无法知道内网tomcat 的地址，也就不能直接通过http的方式发送信息回来，最后想来想去用mina实现了这个功能。

当然，我这里的服务端是整合的了spring 的，也可以直接把服务端独立出来，不整合spring，这个都一样，区别不大。

代码和配置如下：

\---------------------------

1，jar包，我这里使用的是spring4.0.5，mina2.0.7

maven部分文件如下，这个包会自动也依赖进来mina-filter-ssl-1.1.7.jar

<dependency>
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-context-support</artifactId>
 			<version>${springframework-version}</version>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-jdbc</artifactId>
 			<version>${springframework-version}</version>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-orm</artifactId>
 			<version>${springframework-version}</version>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-aop</artifactId>
 			<version>${springframework-version}</version>
 		</dependency>
 		
 		<dependency>
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-web</artifactId>
 			<version>${springframework-version}</version>
 			<exclusions>
 				<exclusion>
 					<groupId>commons-logging</groupId>
 					<artifactId>commons-logging</artifactId>
 				</exclusion>
 			</exclusions>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-webmvc</artifactId>
 			<version>${springframework-version}</version>
 		</dependency>
 		
 		
 		<dependency>
 			<groupId>org.aspectj</groupId>
 			<artifactId>aspectjrt</artifactId>
 			<version>${aspectj-version}</version>
 		</dependency>
 		<dependency>
 			<groupId>org.aspectj</groupId>
 			<artifactId>aspectjweaver</artifactId>
 			<version>${aspectj-version}</version>
 		</dependency>
 
 
 		
 		
 		
 		<dependency>
 			<groupId>org.apache.mina</groupId>
 			<artifactId>mina-core</artifactId>
 			<version>2.0.7</version>
 		</dependency>
 		<dependency>
 			<groupId>org.apache.mina</groupId>
 			<artifactId>mina-integration-spring</artifactId>
 			<version>1.1.7</version>
 		</dependency>
 		<dependency>
 			<groupId>org.apache.mina</groupId>
 			<artifactId>mina-integration-beans</artifactId>
 			<version>2.0.8</version>
 		</dependency>

2，spring-ztc\_app-mina.xml （spring与mina 的配置文件）

spring与mina 的配置文件，需要导入到spring 的总配置文件中，或者加入到web.xml的spring监听扫描中

<?xml version="1.0" encoding="UTF-8"?>
 <beans xmlns="http://www.springframework.org/schema/beans"
 	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	xsi:schemaLocation="http://www.springframework.org/schema/beans 
 http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
 
 
 
 
 
 	
 	<bean class="org.springframework.beans.factory.config.CustomEditorConfigurer">
 		<property name="customEditors">
 			<map>
 				
 				
 				<entry key="java.net.SocketAddress"
 					value="org.apache.mina.integration.beans.InetSocketAddressEditor">
 				</entry>
 			</map>
 		</property>
 	</bean>
 
 	
 	<bean id="serviceHandler" class="cn.hydom.ztc.ztc_app.controller.mina.ServerHandler" />
 
 	
 	<bean id="ioAcceptor" class="org.apache.mina.transport.socket.nio.NioSocketAcceptor"
 		init-method="bind" destroy-method="unbind">
 		<property name="defaultLocalAddress" value=":6007" />
 		<property name="handler" ref="serviceHandler" />
 		<property name="filterChainBuilder" ref="filterChainBuilder" />
 	</bean>
 
 
 
 
 	
 	<bean id="codec" class="org.apache.mina.filter.codec.ProtocolCodecFilter">
 		<constructor-arg>
 			
 			
 			<bean class="cn.hydom.ztc.ztc_app.controller.mina.HCoderFactory" />
 		</constructor-arg>
 	</bean>
 
 	
 	<bean id="logger" class="org.apache.mina.filter.logging.LoggingFilter"></bean>
 
 	
 	<bean id="filterChainBuilder"
 		class="org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder">
 		<property name="filters">
 			<map>
 				<entry key="codec" value-ref="codec" />
 				<entry key="logger" value-ref="logger" />
 			</map>
 		</property>
 	</bean>
 	
 	
 	
 	
 </beans>

3，mina服务端业务处理类

package cn.hydom.ztc.ztc_app.controller.mina;
    
    import org.apache.commons.logging.Log;
    import org.apache.commons.logging.LogFactory;
    import org.apache.mina.core.service.IoHandlerAdapter;
    import org.apache.mina.core.session.IdleStatus;
    import org.apache.mina.core.session.IoSession;
    
    
    /**
     * @Description: mina服务端业务处理类
     * @author whl
     * @date 2014-9-30 下午12:36:28
     *
     */
    public class ServerHandler extends IoHandlerAdapter {
    	
    	
    	private final static Log log = LogFactory.getLog(ServerHandler.class);
    	
    	
    	 public ServerHandler() {
    	  // TODO Auto-generated constructor stub
    	 }
    	 @Override
    	 public void exceptionCaught(IoSession session, Throwable cause)
    	   throws Exception {
    	 }
    	 
    	 
    	 @Override
    	 public void messageReceived(IoSession session, Object message)
    	   throws Exception {
    		 log.debug("服务端收到信息-------------");
    		
    		  //获取客户端发过来的key
    		  String key = message.toString();
    		  System.out.println("message :"+message.toString());
    		  String carPark_id = key.substring(key.indexOf("=") + 1);
    		  System.out.println("carPark_id :"+carPark_id);
    		  
    
    		  //保存客户端的会话session
    		  SessionMap sessionMap = SessionMap.newInstance();
    		  sessionMap.addSession(carPark_id, session);
    	  
    	 }
    	 @Override
    	 public void messageSent(IoSession session, Object message) throws Exception {
    		 log.debug("------------服务端发消息到客户端---");
    	 }
    	 @Override
    	 public void sessionClosed(IoSession session) throws Exception {
    	  // TODO Auto-generated method stub
    		 log.debug("远程session关闭了一个..." + session.getRemoteAddress().toString());
    	 }
    	 @Override
    	 public void sessionCreated(IoSession session) throws Exception {
    		 log.debug(session.getRemoteAddress().toString() +"----------------------create");
    	 }
    	 @Override
    	 public void sessionIdle(IoSession session, IdleStatus status)
    	   throws Exception {
    		 log.debug(session.getServiceAddress() +"IDS");
    	 }
    	 @Override
    	 public void sessionOpened(IoSession session) throws Exception {
    		 log.debug("连接打开："+session.getLocalAddress());
    	 }
    	 
    	 
    	}

4,服务端缓存客户端socket连接的单例类

package cn.hydom.ztc.ztc_app.controller.mina;
 
 import java.util.HashMap;
 import java.util.Map;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.mina.core.session.IoSession;
 
 
 /**
 * @Description: 单例工具类，保存所有mina客户端连接
 * @author whl
 * @date 2014-9-29 上午10:09:15
 *
 */
 public class SessionMap {
 	
 	private final static Log log = LogFactory.getLog(SessionMap.class);
 	
 	private static SessionMap sessionMap = null;
 			
 	private Map<String, IoSession>map = new HashMap<String, IoSession>();
 	
 	
 	//构造私有化 单例
 	private SessionMap(){}
 	
 	
 	/**
 	 * @Description: 获取唯一实例
 	 * @author whl
 	 * @date 2014-9-29 下午1:29:33
 	 */
 	public static SessionMap newInstance(){
 		log.debug("SessionMap单例获取---");
 		if(sessionMap == null){
 			sessionMap = new SessionMap();
 		}
 		return sessionMap;
 	}
 	
 	
 	/**
 	 * @Description: 保存session会话
 	 * @author whl
 	 * @date 2014-9-29 下午1:31:05
 	 */
 	public void addSession(String key, IoSession session){
 		log.debug("保存会话到SessionMap单例---key=" + key);
 		this.map.put(key, session);
 	}
 	
 	/**
 	 * @Description: 根据key查找缓存的session
 	 * @author whl
 	 * @date 2014-9-29 下午1:31:55
 	 */
 	public IoSession getSession(String key){
 		log.debug("获取会话从SessionMap单例---key=" + key);
 		return this.map.get(key);
 	}
 	
 	/**
 	 * @Description: 发送消息到客户端
 	 * @author whl
 	 * @date 2014-9-29 下午1:57:51
 	 */
 	public void sendMessage(String[] keys, Object message){
 		for(String key : keys){
 			IoSession session = getSession(key);
 			
 			log.debug("反向发送消息到客户端Session---key=" + key + "----------消息=" + message);
 			if(session == null){
 				return;
 			}
 			session.write(message);
 			
 		}
 	}
 	
 	
 	
 
 }

5，编码解码器，

HCoderFactory.java

package cn.hydom.ztc.ztc_app.controller.mina;
    
    import java.nio.charset.Charset;
    
    import org.apache.mina.core.session.IoSession;
    import org.apache.mina.filter.codec.ProtocolCodecFactory;
    import org.apache.mina.filter.codec.ProtocolDecoder;
    import org.apache.mina.filter.codec.ProtocolEncoder;
    
    /**
     * @Description: 编码和解码器工厂类.
     * @author whl
     * @date 2014-9-30 下午12:34:59
     *
     */
    public class HCoderFactory implements ProtocolCodecFactory {
    	private final HEncoder encoder;
    	private final HDecoder decoder;
    
    	public HCoderFactory() {
    		//this(Charset.defaultCharset());
    		this(Charset.forName("UTF-8"));
    
    	}
    
    	public HCoderFactory(Charset charSet) {
    		this.encoder = new HEncoder(charSet);
    		this.decoder = new HDecoder(charSet);
    	}
    
    	@Override
    	public ProtocolDecoder getDecoder(IoSession arg0) throws Exception {
    		return decoder;
    	}
    
    	@Override
    	public ProtocolEncoder getEncoder(IoSession arg0) throws Exception {
    		return encoder;
    	}
    }

HDecoder.java

package cn.hydom.ztc.ztc_app.controller.mina;
    
    import java.nio.charset.Charset;
    import java.nio.charset.CharsetDecoder;
    
    import org.apache.mina.core.buffer.IoBuffer;
    import org.apache.mina.core.session.IoSession;
    import org.apache.mina.filter.codec.CumulativeProtocolDecoder;
    import org.apache.mina.filter.codec.ProtocolDecoderOutput;
    
    /**
     * @Description: 解码工具类
     * @author whl
     * @date 2014-9-30 下午12:35:22
     *
     */
    public class HDecoder extends CumulativeProtocolDecoder {
    	private final Charset charset;
    
    	public HDecoder(Charset charset) {
    		this.charset = charset;
    	}
    
    	public boolean doDecode(IoSession session, IoBuffer in,
    			ProtocolDecoderOutput out) throws Exception {
    		
    		//System.out.println("-------doDecode----------");
    		
    		CharsetDecoder cd = charset.newDecoder();
    		String mes = in.getString(cd);
    		out.write(mes);
    		return true;
    		
    		
    		/*
    		if (in.remaining() > 4) {// 有数据时，读取字节判断消息长度
    			in.mark();// 标记当前位置，以便reset
    			int size = in.getInt();
    
    			// 如果消息内容不够，则重置，相当于不读取size
    			if (size > in.remaining()) {
    				in.reset();
    				return false;// 接收新数据，以拼凑成完整数据
    			} else if (size != 0 && (size - 4 >= 0)) {
    				byte[] bytes = new byte[size - 4];
    				//int protocol = in.getInt();
    				// 拿到客户端发过来的数据组装成基础包写出去
    				in.get(bytes, 0, size - 4);
    				//in.get(bytes, size - 4, size);
    				
    				PackageBeanFactory beanFactory = (PackageBeanFactory) session
    						.getAttribute(ServerHandler.BEAN_FACTORY);
    				//out.write(beanFactory.getPackage(protocol, size, bytes));
    				
    				String mes = in.getString(cd);
    				
    				out.write(mes);
    				// 如果读取内容后还粘了包，就让父类再给读取进行下次解析
    				if (in.remaining() > 0) {
    					return true;
    				}
    			}
    		}
    		
    		return false;// 处理成功，让父类进行接收下个包
    */	
    		
    	
    	}
    }

HEncoder.java

package cn.hydom.ztc.ztc_app.controller.mina;
    
    import java.nio.charset.Charset;
    import java.nio.charset.CharsetDecoder;
    import java.nio.charset.CharsetEncoder;
    
    import org.apache.commons.logging.Log;
    import org.apache.commons.logging.LogFactory;
    import org.apache.mina.core.buffer.IoBuffer;
    import org.apache.mina.core.session.IoSession;
    import org.apache.mina.filter.codec.ProtocolEncoder;
    import org.apache.mina.filter.codec.ProtocolEncoderOutput;
    
    
    /**
     * @Description: 编码工具类
     * @author whl
     * @date 2014-9-30 下午12:35:35
     *
     */
    public class HEncoder implements ProtocolEncoder {
    	
    	private final static Log log = LogFactory.getLog(HEncoder.class);
    	
    	private final Charset charset;
    
    	public HEncoder(Charset charset) {
    		this.charset = charset;
    	}
    
    	@Override
    	public void encode(IoSession session, Object message,
    			ProtocolEncoderOutput out) throws Exception {
    		
    		CharsetEncoder ce = charset.newEncoder();
    		String mes = (String) message;
    		IoBuffer buffer = IoBuffer.allocate(100).setAutoExpand(true);
            buffer.putString(mes,ce);
            buffer.flip();
            out.write(buffer);
    		
    		/*System.out.println("---------encode-------------");
    		String mes = (String) message;
    		byte[] data = mes.getBytes("UTF-8");
    		IoBuffer buffer = IoBuffer.allocate(data.length + 4);
    		buffer.putInt(data.length);
    		buffer.put(data);
    		buffer.flip();
    		out.write(buffer);
    		out.flush();*/
    	}
    
    	@Override
    	public void dispose(IoSession session) throws Exception {
    		log.info("Dispose called,session is " + session);
    	}
    }

6,客户端程序  
ClentMain.java

package cn.hydom.ztc.ztc_app.controller.mina;
    
    import java.net.InetSocketAddress;
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    import org.apache.commons.logging.Log;
    import org.apache.commons.logging.LogFactory;
    import org.apache.mina.core.filterchain.IoFilterAdapter;
    import org.apache.mina.core.future.ConnectFuture;
    import org.apache.mina.core.session.IoSession;
    import org.apache.mina.filter.codec.ProtocolCodecFilter;
    import org.apache.mina.filter.logging.LoggingFilter;
    import org.apache.mina.transport.socket.nio.NioSocketConnector;
    
    /**
     * @Description: mina客户端，包含断线重连机制，空闲重连机制
     * @author whl
     * @date 2014-11-2
     */
    public class ClentMain extends Thread{
    
    	private final static Log log = LogFactory.getLog(ClentMain.class);
    	
    	@Override
    	public void run() {
    		
    		//ip
    		String host = "192.168.0.38";
    		//端口
    		int port = 6007;
    		//停车场id
    		final String carPark_id = "1";
    		
    		
    		// 创建客户端连接器.
    		final NioSocketConnector connector = new NioSocketConnector();
    		//设置连接超时  
    		connector.setConnectTimeoutMillis(30000); 
    		// 设置默认访问地址  
    		connector.setDefaultRemoteAddress(new InetSocketAddress(host, port));
    		//将IoSession的主键属性注入线程映射表MDC中
    	    //connector.getFilterChain().addLast("mdc", new MdcInjectionFilter());  
    	    //日志过滤器
    		connector.getFilterChain().addLast("logger", new LoggingFilter());
    		// 设置编码过滤器
    		connector.getFilterChain().addLast("codec",new ProtocolCodecFilter(new HCoderFactory()));        
    	    //添加处理器  
    	    connector.setHandler(new ClintHandler());
    	    // 设置接收缓冲区的大小  
    	    connector.getSessionConfig().setReceiveBufferSize(10240);
    	    // 设置输出缓冲区的大小  
    	    connector.getSessionConfig().setSendBufferSize(10240);
    	    
    	    /**
    	     * 空闲重连的机制，根据需要选择相应的配置
    	     */
    	    // 读写都空闲时间:30秒  
    	    //connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 30); 
    	    // 读(接收通道)空闲时间:40秒 
    	    //connector.getSessionConfig().setIdleTime(IdleStatus.READER_IDLE, 40);
    	    // 写(发送通道)空闲时间:50秒 
    	    //connector.getSessionConfig().setIdleTime(IdleStatus.WRITER_IDLE, 50); 
    	   
    	    //断线重连回调拦截器  
    	    connector.getFilterChain().addFirst("reconnection", new IoFilterAdapter() {  
    	        @Override  
    	        public void sessionClosed(NextFilter nextFilter, IoSession ioSession) throws Exception {  
    	            for(;;){  
    	                try{  
    	                    Thread.sleep(3000); 
    	                    ConnectFuture future = connector.connect();  
    	                    future.awaitUninterruptibly();// 等待连接创建成功  
    	                    IoSession session = future.getSession();// 获取会话  
    	                    session.write("key="+carPark_id);
    	                    
    	                    if(session.isConnected()){  
    	                    	log.info("断线重连["+ connector.getDefaultRemoteAddress().getHostName() +":"+ connector.getDefaultRemoteAddress().getPort()+"]成功");  
    	                        //System.out.println("断线重连["+ connector.getDefaultRemoteAddress().getHostName() +":"+ connector.getDefaultRemoteAddress().getPort()+"]成功");
    	                    	break;  
    	                    }  
    	                }catch(Exception ex){  
    	                	log.info("重连服务器登录失败,3秒再连接一次:" + ex.getMessage());  
    	                	//System.out.println("重连服务器登录失败,3秒再连接一次:" + ex.getMessage());  
    	                }  
    	            }  
    	        }  
    	    });
    	    
    	    //开始连接
    	    for (;;) {  
    	        try {  
    	            ConnectFuture future = connector.connect();  
    	            // 等待连接创建成功  
    	            future.awaitUninterruptibly();  
    	            // 获取会话  
    	            IoSession session = future.getSession();  
    	            //发送消息
    	            session.write("key=" + carPark_id);
    	            log.error("连接服务端" + host + ":" + port + "[成功]" + ",,时间:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));  
    	            break;  
    	        } catch (Exception e) {  
    	            //System.out.println("连接服务端" + host + ":" + port + "失败" + ",,时间:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + ", 连接MSG异常,请检查MSG端口、IP是否正确,MSG服务是否启动,异常内容:" + e.getMessage());  
    	            log.error("连接服务端" + host + ":" + port + "失败" + ",,时间:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + ", 连接MSG异常,请检查MSG端口、IP是否正确,MSG服务是否启动,异常内容:" + e.getMessage(), e);  
    	            // 连接失败后,重连10次,间隔30s  
    	            try {
    					Thread.sleep(5000);
    				} catch (InterruptedException e1) {
    					e1.printStackTrace();
    					log.error("连接服务端失败后，睡眠5秒发生异常！");
    				}
    	        }  
    	    }
    	    
    	  
    			
    		// cf.getSession().write("quit");//发送消息
    		//cf.getSession().close();
    		//cf.getSession().getCloseFuture().awaitUninterruptibly();// 等待连接断开
    		//connector.dispose();
    
    	}
    
    }

ClintHandler.java

7，测试

这里就不写实际的代码了，

1，首先部署web工程启动tomcat，让服务端先运行起来，

2，然后运行客户端

public class ClintTest1 {
    
    
    	public static void main(String[] args) throws Exception {
    		
    		ClentMain mina = new ClentMain();
    		mina.start();
    
    	}
    }

3，想要服务端主动发回来信息，还得在服务端的web工程的action中写一个htpp访问的方法，方法中去sessionMap类中缓存的map中查找session，然后发送消息。

代码如下，我用的是springmvc

/**
 	 * @Description: 发送消息到客户端
 	 * @author whl
 	 * @date 2014-9-29 下午1:18:54
 	 */
 	@ResponseBody
 	@RequestMapping(value="/sendMessage")
 	public String sendMessage(HttpServletRequest request, String[] carPark_id){
 		try{
 			
 			//
 			//获取链接的参数carPark_id
 			log.debug("carPark_id[].length--------- " + carPark_id.length);
 			for(String id : carPark_id){
 				log.debug("carPark_id --------- " + id);
 			}
 			//这里用的假数据<pre name="code" class="java"> carPark_id = new String[]{"1"};

//发送的信息String jsonstr = "123";//反向发送信息log.debug("开始反向发送消息到客户端-------- ");SessionMap sessionMap = SessionMap.newInstance();sessionMap.sendMessage(carPark\_id, jsonstr);//返回信息return "发送的信息为" + jsonstr ;  
\}catch (Exception e) \{log.error(e);return "出错了。。";\}\}

4，好了，现在重新发布工程，启动服务端和客户端，然后访问这个链接就可以了，当然springmvc需要自己配置，这也不是这里的重点。

注意：客户端发过来的carPark\_id必须与服务端查找的一致，不让就找不到相应的客户端session连接，消息无法发送成功。