基于osgi框架采用mina+springdm+hibernate实现的网络服务器

背景：系统是一个多级联网的安防平台的网络调度管理部分，主要功能是接受客户端视频请求，分配组播资源，命令设备

         网关打开dvr摄像头，或者命令视频转发服务器负责在动态分配的ip，prot上转发视频流，

         并且在客户端关闭的时候查看是否还有相应的客户端连接，

         无客户端则命令网关/转发服务器关闭视频流。客户端类型有b/s,c/s,录像服务器。

         由于需要兼容不同类型的设备及网关，所以希望做成一个动态扩展的模块化的系统。因此采用的了mina作为网络连接框架，

         主要是简单，基于消息事件的。

 

设计方案：系统提供一个核心的模块，负责提供model及dao服务。其中dao导出为osgi服务，供各个子模块使用。

               系统采用二进制通讯协议，各个decoder&encoder注册动态安装部署为服务。message的消费采用一个

               注册的command服务消费。因此可以做到不同的协议或者协议有变化可以动态添加修改

               主要类与配置如下：

public class DemuxingProtocolDecoder extends CumulativeProtocolDecoder {
                         ：
                         ：
 @Override
    protected boolean doDecode(IoSession session, IoBuffer in,
            ProtocolDecoderOutput out) throws Exception {
            int undecodables = 0;
            MessageDecoder currentDecoder =null;
            for (MessageDecoder decoder:decoders) {
                int limit = in.limit();
                int pos = in.position();

                MessageDecoderResult result;
                try {
                    result = decoder.decodable(session, in);
                } finally {
                    in.position(pos);
                    in.limit(limit);
                }

                if (result == MessageDecoder.OK) {
                    currentDecoder = decoder;
                    break;
                } else if (result == MessageDecoder.NOT_OK) {
                    undecodables++;
                } else if (result != MessageDecoder.NEED_DATA) {
                    throw new IllegalStateException(
                            "Unexpected decode result (see your decodable()): "
                                    + result);
                }
            }

            if (undecodables == decoders.size()) {
                // Throw an exception if all decoders cannot decode data.
                String dump = in.getHexDump();
                in.position(in.limit()); // Skip data
                ProtocolDecoderException e = new ProtocolDecoderException(
                        "No appropriate message decoder: " + dump);
                e.setHexdump(dump);
                log.error(e.getMessage());
                return false;
            }

            if (currentDecoder == null) {
                // Decoder is not determined yet (i.e. we need more data)
                return false;
            }

        MessageDecoderResult result = currentDecoder.decode(session, in,
                out);
        if (result == MessageDecoder.OK) {
            currentDecoder = null;
            return true;
        } else if (result == MessageDecoder.NEED_DATA) {
            return false;
        } else if (result == MessageDecoder.NOT_OK) {
            currentDecoder = null;
            throw new ProtocolDecoderException(
                    "Message decoder returned NOT_OK.");
        } else {
            currentDecoder = null;
            throw new IllegalStateException(
                    "Unexpected decode result (see your decode()): "
                            + result);
        }
    }



                                   ：
                                   ：
}
上面这个类是一个改造后的解码器，可以通过动态寻找注册的decoder服务，在这主要是一个通过spring注入的list来实现的。
-----------------------------------------
系统中的encoder类
public class DemuxingProtocolEncoder  implements ProtocolEncoder,BundleContextAware {
	
	private final Logger logger = LoggerFactory.getLogger(getClass());
	
	private BundleContext bundleContext;
	
    public BundleContext getBundleContext() {
		return bundleContext;
	}

	public void setBundleContext(BundleContext bundleContext) {
		this.bundleContext = bundleContext;
	}

	public DemuxingProtocolEncoder() {
    	
    }

    @SuppressWarnings("unchecked")
	public void encode(IoSession session, Object message,
            ProtocolEncoderOutput out) throws Exception {
        MessageEncoder<Object> encoder = (MessageEncoder<Object>)findCommandServiceByBeanName(message.getClass().getName());
        if (encoder != null) {
            encoder.encode(session, message, out);
        } else {
            throw new UnknownMessageTypeException(
                    "No message encoder found for message: " + message);
        }
    }

    public void dispose(IoSession session) throws Exception {    
    }
    
    public Object findCommandServiceByBeanName(String messageTypeClass){
		try {
			ServiceReference[] sefs = bundleContext.getAllServiceReferences("org.apache.mina.filter.codec.demux.MessageEncoder", "(messageTypeClass="+messageTypeCl


ass+")");
			if(sefs!=null&&sefs.length>0){
				return bundleContext.getService(sefs[0]);
			}
		} catch (InvalidSyntaxException e) {
			logger.error(e.getMessage());
		}		
		return null;
	}
}
这个类主要是实现了在osgi注册表中寻找相应的message的encoder，通过osgi服务的fliter查找的。上面标红的部分。
以上就是主要的改造mian的类。其中spring-dm的主要配置如下：
<osgi:list id="decoders" interface="org.apache.mina.filter.codec.demux.MessageDecoder" cardinality="0..N"></osgi:list> 
这个decoders会动态注入到上面的DemuxingProtocolDecoder对象中。每个模块注册decoder与encoder，
command的方式如下：
<?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:osgi="http://www.springframework.org/schema/osgi"
  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
                      http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd"
                      default-autowire="byName" default-lazy-init="false" >

	<bean id="registerExDecoder" class="com.bct.exchange.au.exregister.RegisterExDecoder"></bean>
    <bean id="registerExMessageCommand" class="com.bct.exchange.au.exregister.RegisterExMessageCommand"></bean>
    
    <bean id="registerExEncoder" class="com.bct.exchange.au.exregister.RegisterExEncoder"></bean>
    <bean id="unitAddDeleteEncoder3" class="com.bct.exchange.au.exregister.UnitAddDeleteEncoder3"></bean>
    
    <osgi:service id="registerExDecoderService" 
    	interface="org.apache.mina.filter.codec.demux.MessageDecoder"
    	ref="registerExDecoder">
    </osgi:service>
    
    <osgi:service id="registerExEncoderService" 
    	interface="org.apache.mina.filter.codec.demux.MessageEncoder"
    	ref="registerExEncoder">
    	<osgi:service-properties>
			<entry key="messageTypeClass">
				<value>com.bct.exchange.au.exregister.RegisterExMessage</value>
			</entry>    	
    	</osgi:service-properties>
    </osgi:service>
    
    <osgi:service id="unitAddDeleteEncoder3Service" 
    	interface="org.apache.mina.filter.codec.demux.MessageEncoder"
    	ref="unitAddDeleteEncoder3">
    	<osgi:service-properties>
			<entry key="messageTypeClass">
				<value>com.bct.exchange.au.exregister.UnitAddDeleteMessage3</value>
			</entry>    	
    	</osgi:service-properties>
    </osgi:service>
    
    <osgi:service id="registerExMessageCommandService" 
    	interface="com.bct.commandHandle.Command"
    	ref="registerExMessageCommand">
    	<osgi:service-properties>
			<entry key="messageTypeClass">
				<value>com.bct.exchange.au.exregister.RegisterExMessage</value>
			</entry>    	
    	</osgi:service-properties>
    </osgi:service>
    
</beans> 

目前这种构架基本实现成功，主要功能已经在这个上面实现成功，没有什么大的问题。现在想考虑的问题是，
这个东西怎么做分布式，也就是怎么在一个jvm上的osgi调用另一个jvm上面的osgi服务，第二，这个系统怎么
做成热备或集群的方式。现在主要考虑的是这两个方面，等有好的事项方式在和各位探讨

 

 

评论

6 楼 kebo 2008-12-03  

呵呵，等我搞完了这段，在写点想法

5 楼 boyingking 2008-11-27  

楼主主要在哪些方面应用了OSGI呢？能否进行一下更深点的交流？楼主在实际项目中用过了OSGI，能否从宏观上总结一下经验，写给大家分享一下呢？

4 楼 kebo 2008-11-20  

呵，没有来讨论的。那就自己来。这套方式，基本融合了osgi的面向服务和事件驱动的编程方式。
首先。系统是通过一个一个message来驱动的。io产生一个message。注册的command service去消费
。而且灵活就在于一个command service可以注册对多个message感兴趣。所以command service的的执行可以被多种方式驱动。第二一个消息可以被多个command service同时消费处理。所以一个消息可以在系统中产生一连串的反应。
这样可以实现一些细粒度的command service，在系统中组合执行。达到最大的重用与解耦。

3 楼 kebo 2008-11-19  

需要那块的代码？

2 楼 javierhui 2008-11-19  

楼主可否给出更详尽一点的源代码?

1 楼 liuchan7448 2008-11-18  

楼主对OSGi的理解很深，写的很详细,正在学习osgi,向楼主学习。