ITKeyword,专注技术干货聚合推荐

注册 | 登录

架构设计:系统间通信(42)——自己动手设计ESB(3)

yinwenjie 2016-07-28

====================================================
(接上文《架构设计:系统间通信(41)——自己动手设计ESB(2)》)

4-2-3、业务层与Class文件管理

zookeeper控制模块的业务相关层(封装层),负责根据Apache Curator层分析好的事件类型、Path Node名称信息、Path Node内容信息,进行Class文件的生成。这里的关键点是,Apache Camel有一个动态加载的限制:它只能在被创建的线程中完成Service(例如Route路由定义)的动态加载过程。Apache Curator事件监听规则基于zookeeper原生客户端,当发生ZK事件时Apache Curator会使用一个独立线程处理这个事件而不是在启动Apache Curator组件的线程。所以Apache Camel的Context上下文对象的动态加载操作不能在事件线程中进行,我们需要一个同步队列将Class File文件生成之后的,对Apache Camel相关的操作过程通知到其他线程去继续执行。请看以下业务相关层的实现类的代码片段(对ESBCamelServiceListener接口的实现):

public class ESBCamelServiceListenerImpl implements ESBCamelServiceListener {

    ......

    public void onProcessorAdded(String systemId, String packageName, String className, byte[] contents) {
        /* * 当一个处理器进行添加时,要做以下处理 * 1、首先根据systemId的信息,在ESB-Broker Server的映射容器中寻 * 找个业务系统定义的各种Server Class 文件在ESB-Broker Server节点本地存储的根路径 * 2、然后按照packageName、className、contents的信息将这个class文件写入到正确的位置 * * 注意:由于此时只是完成了class了文件的写入,所以这个class文件还没有被classloader进行初始化。 * 另外由于,CamelContext并没有提供单独进行processor处理器加载的功能,而是随着routes实例的加载而加载 * 而这个工作将在onRouteDefineChanged事件中完成,所以在完成processor-class文件的写入后,就不需要再做其它事情了。 * */
        if(StringUtils.isEmpty(systemId) || StringUtils.isEmpty(packageName)
            || StringUtils.isEmpty(className) || contents == null) {
            ESBCamelServiceListenerImpl.LOGGER.error("必要的入参数据不正确,请检查!");
            return;
        }
        this.createClassfile(systemId, packageName, className, contents);
    }

    ......

    public void onRouteDefineAdded(String systemId, String packageName, String className, byte[] contents) {
        /* * 当一个新的路由定义事件发生时,要做以下几件事情: * * 1、根据systemId的信息,在ESB-Broker Server的映射容器中寻 * 找个业务系统定义的各种Server Class 文件在ESB-Broker Server节点本地存储的根路径 * 2、然后按照packageName、className、contents的信息将这个class文件写入到正确的位置 * 3、不能在本线程操作Apache Camel,只能通过一个同步队列通知Apache Camel主线程 * */
        if(StringUtils.isEmpty(systemId) || StringUtils.isEmpty(packageName)
            || StringUtils.isEmpty(className) || contents == null) {
            ESBCamelServiceListenerImpl.LOGGER.error("必要的入参数据不正确,请检查!");
            return;
        }
        // 第1、2两步处理过程,都是在这里完成
        this.createClassfile(systemId, packageName, className, contents);

        // 3、===============加载到CamelContext中
        SynchronousQueue<String> camelContextOperateQueue = CamelContextOperateQueueUtils.getCamelContextOperateQueue();
        try {
            camelContextOperateQueue.put(packageName + "." + className);
        } catch (InterruptedException e) {
            ESBCamelServiceListenerImpl.LOGGER.error(e.getMessage() , e);
        }
    }

    ......

    /** * 这个私有方法用于生成class文件 * @param systemId * @param packageName * @param className * @param contents */
    private void createClassfile(String systemId, String packageName, String className, byte[] contents) {
        // 1、============
        Map<String, URL> systemPathMapping = SystemRootPathUtils.getSystemPathMapping();
        URL systemRootURL = systemPathMapping.get(systemId);
        if(systemRootURL == null) {
            return;
        }

        // 2、============开始写入class文件
        ClassFileUtils.createClassfile(systemRootURL, packageName, className, contents);

        // 完成
        ESBCamelServiceListenerImpl.LOGGER.info("===================" + packageName + "." + className + ".class 生成过程结束");
    }
}

以上代码片段只列举对ESBCamelServiceListener接口的两个实现方法:onProcessorAdded(String, String, String, byte[])和onRouteDefineAdded(String, String, String, byte[]),其他的实现方法处理过程都是相似的,为了节约篇幅就不再赘述了。请注意createClassfile(String, String, String, byte[])这个私有方法,这个私有方法是进行文件写入的操作,它将在指定的业务系统存储Class File的目录下,写入最新的Class文件。

而具体的文件处理过程放在“ClassFileUtils”这个工具类中,这个工具类是笔者自定义的工具类,主要是保证文件处理过程能够在各处进行代码重用。其中无非就是File Stream的写入过程,因为过于简单就不再占用文章篇幅了。这里截取一些“ClassFileUtils”工具类的代码片段:

/** * class文件的处理工具。工具类不允许继承 * @author yinwenjie */
public final class ClassFileUtils {
    ......
    /** * 工具类不允许进行实例化 */
    private ClassFileUtils() {

    }

    ......

    /** * 通过这个工具方法,可以进行class文件的创建 * @param systemRootURL 本次进行系统创建的system业务系统存储class文件的根目录 * @param packageName class文件的包信息 * @param className class文件的类名信息 * @param contents class文件的内容信息 */
    public static void createClassfile(URL systemRootURL , String packageName, String className, byte[] contents) {
        ......
        // 开始输出文件内容
        try {
            ByteArrayInputStream byteInputStream = new ByteArrayInputStream(contents);
            FileOutputStream fileOut = new FileOutputStream(classFile);
            int realLen;
            byte[] fileContexts = new byte[4096];
            while((realLen = byteInputStream.read(fileContexts, 0, 4096)) != -1) {
                fileOut.write(fileContexts, 0, realLen);
            }
            fileOut.close();
            byteInputStream.close();
        } catch (Exception e) {
            ClassFileUtils.LOGGER.error(e.getMessage() , e);
            return;
        }
        ......
    }

    ......
}

在ESBCamelServiceListenerImpl类的代码片段中,需要注意的另一个细节是:onRouteDefineAdded方法中使用的“SynchronousQueue”同步队列。官方对于“SynchronousQueue”同步队列的特性解释是:“是这样 一种阻塞队列,其中每个 put 必须等待一个 take,反之亦然。同步队列没有任何内部容量,甚至连一个队列的容量都没有。”这个队列也在笔者以前的文中介绍过,它是一个没有任何存储空间的队列,当某一个线程使用“SynchronousQueue”同步队列的put()方法进行推入操作时,会一直进行等待。一直等待到另一线程使用“SynchronousQueue”同步队列的take()方法,试图取得一个值为止。这样保证了对Apache Camel进行操作的主线程能够依次处理其它线程传来的对Apache Camel进行动态Routes加载的请求。

4-3、ESB-Broker Server启动过程

通过本文第4-2小节的介绍,读者实际上应该可以理解到一些涉及ESB-Broker Server节点启动过程的关键信息点。由于Apache Camel服务模块对Service的加载依赖于Apache Curator组件对zookeeper的操作,因为主程序需要从后者中读取有哪些处理器、哪些路由定义需要加载到Camel Context上下文中;另一方面各种Service加载组件又必须等待Camel Context上下文加载组件启动成功后才能进行进行,否则就没有加载的目的地。

所以整个ESB-Broker Server节点的启动过程就需要倒着进行:首先通过Curator连接到zookeeper服务器,并完成事件监听的注册。待前面所有准备工作完成,就需要从zookeeper服务端读取需要加载的处理器、原子服务和编排好的路由配置的数据信息(每种业务系统都需要分别进行读取和处理)。最后启动Camel Context上下文,并一直试图从“SynchronousQueue”同步队列中取出其它线程传来的对Camel Context上下文进行动态加载的请求。以上文字描述的启动顺序可以抽象为以下示意图:

这里写图片描述

注意,当新启动的ESB-Broker Server节点从zookeeper服务端查询需要加载的各种Service时,并不代表这些Service的class信息都需要从zookeeper上读取Path中的data内容进行加载。主程序会检查这些Service所对应的class是否已经存在于本地的某个路径上,如果已经存在,则直接使用classloader进行加载。以下的代码片段大致说明了ESB-Broker Server的启动过程:

public class BootStartup {

    ......

    /** * 这是一个固定的存储class文件的根路径 * 正式系统中,这个值可来自于系统的配置文件 */
    private static final String CLASSROOTPATH = "/testCamelRoot";

    /** * 主执行函数 * @param args * @throws Exception */
    public static void main(String[] args) throws Exception {
        new BootStartup().start();
    }

    ......
    /** * 主要的启动过程在这里 */
    protected void start() throws Exception {
        /* * 启动顺序为: * * 1、首先使用Apache Curator组件进行zookeeper服务端连接, * * 2、通过Apache Curator组件查询目前已有的业务系统ID信息, * 以便依据这些业务系统ID,建立本地ClassLoader的根路径(当然,由就不需要创建了) * 并且注册Apache Curator封装层的监听。 * * 3、检查第2步得到的业务系统下,已经存在的processor处理器定义和已有的route定义 * 以便生成class信息。 * * 4、启动Apache Camel服务,将第三步那些route定义进行动态加载 * * 5、经过以上步骤,整个启动过程完成了。现在主线程需要一直等待从SynchronousQueue队列发来的动态加载请求了。 * */
        // 1、===============
        // 连接到zookeeper
        CuratorFramework zkClient = CuratorFrameworkFactory.newClient("192.168.61.140:2181", 30000, 30000, new RetryNTimes(50, 3000));
        zkClient.start();
        BootStartup.LOGGER.info("完成zookeeper服务端的连接工作!");

        // 2、===============
        // 取得已存在的系统
        List<String> systemNames = zkClient.getChildren().forPath("/");
        // 开始构造classLoader路径、建立针对zookeeper路径下的监听、
        List<File> systemClassRootPaths = new ArrayList<File>();
        List<RouteBuilder> alreadyRouteBuilders = new ArrayList<RouteBuilder>();
        for (String systemName : systemNames) {
            // 这是system业务系统在本地存储class的根目录
            File systemClassRootPath = new File(CLASSROOTPATH + "/" + systemName);
            if(!systemClassRootPath.exists()) {
                systemClassRootPath.mkdirs();
            }
            Map<String, URL> systemRootPathMapping =  SystemRootPathUtils.getSystemPathMapping();
            // 记录到工具类中,以便其它线程需要时进行取用
            systemRootPathMapping.put(systemName, systemClassRootPath.toURI().toURL());
            systemClassRootPaths.add(systemClassRootPath);

            // 建立ZK目录监听
            PathChildrenCache watcherProcesses = null;
            PathChildrenCache watcherRoutes = null;
            try {
                // 建立System业务系统下的processes目录和监听
                if(zkClient.checkExists().forPath("/" + systemName + "/processes") == null) {
                    zkClient.create().creatingParentsIfNeeded().forPath("/" + systemName + "/processes");
                }
                watcherProcesses = new PathChildrenCache(zkClient, "/" + systemName + "/processes", true);
                watcherProcesses.getListenable().addListener(new ProcessesPathChildrenCacheListener());
                watcherProcesses.start();

                // 建立System业务系统下的routes目录和监听
                if(zkClient.checkExists().forPath("/" + systemName + "/routes") == null) {
                    zkClient.create().creatingParentsIfNeeded().forPath("/" + systemName + "/routes");
                }
                watcherRoutes = new PathChildrenCache(zkClient, "/" + systemName + "/routes", true);
                watcherRoutes.getListenable().addListener(new RoutesPathChildrenCacheListener());
                watcherRoutes.start();
            } catch(Exception e) {
                if(watcherProcesses != null) {
                    watcherProcesses.close();
                }
                if(watcherRoutes != null) {
                    watcherRoutes.close();
                }
            }

            // 3、===============
            // 建立已经存在的class文件
            List<String> processesPaths = zkClient.getChildren().forPath("/" + systemName + "/processes");
            List<String> routesPaths = zkClient.getChildren().forPath("/" + systemName + "/routes");
            // 创建processor文件
            for (String processesPath : processesPaths) {
                String fullClassName = processesPath;
                // 获得class的包名和类名
                int lastIndex = fullClassName.lastIndexOf(".");
                String className = fullClassName.substring(lastIndex + 1);
                String packageName = fullClassName.substring(0 , lastIndex);
                byte[] fileContexts = zkClient.getData().forPath("/" + systemName + "/processes" + "/" + processesPath);

                // 创建文件
                ClassFileUtils.createClassfile(systemClassRootPath.toURI().toURL(), packageName, className, fileContexts);
            }

            // 创建route文件
            for (String routesPath : routesPaths) {
                String fullClassName = routesPath;
                // 获得class的包名和类名
                int lastIndex = fullClassName.lastIndexOf(".");
                String className = fullClassName.substring(lastIndex + 1);
                String packageName = fullClassName.substring(0 , lastIndex);
                byte[] fileContexts = zkClient.getData().forPath("/" + systemName + "/routes" + "/" + routesPath);

                // 创建文件
                ClassFileUtils.createClassfile(systemClassRootPath.toURI().toURL(), packageName, className, fileContexts);
                RouteBuilder routeBuilder = (RouteBuilder)Class.forName(packageName + "." + className).newInstance();
                alreadyRouteBuilders.add(routeBuilder);
            }
        }

        // 4、=============
        // 首先启动Apache Camel服务
        CamelContext camelContext = new DefaultCamelContext();
        camelContext.start();
        BootStartup.LOGGER.info("Apache Camel Context 启动完成......");
        // 加载和设置ClassLoader
        List<URL> URLs = new ArrayList<URL>();
        for (File systemClassRootPath : systemClassRootPaths) {
            URLs.add(systemClassRootPath.toURI().toURL());
        }
        ClassLoader currentClassLoader = Thread.currentThread().getContextClassLoader();
        ClassLoader camelESBClassLoader = new URLClassLoader(URLs.toArray(new URL[]{}) , currentClassLoader);
        Thread.currentThread().setContextClassLoader(camelESBClassLoader);
        camelContext.setApplicationContextClassLoader(camelESBClassLoader);
        // 然后就可以进行RouteBuilder的加载
        for (RouteBuilder routeBuilder : alreadyRouteBuilders) {
            try {
                camelContext.addRoutes(routeBuilder);
            } catch(Exception e) {
                BootStartup.LOGGER.warn(e.getMessage() , e);
            }
        }

        // 5、=============
        // 开始监控CamelContext动态操作队列
        SynchronousQueue<String> camelContextOperateQueue = CamelContextOperateQueueUtils.getCamelContextOperateQueue();
        String mustLoadedClassName = null;
        // 如果没有收到其它线程的加载请求,主线程将停止在这里
        while((mustLoadedClassName = camelContextOperateQueue.take()) != null) {
            Class<RouteBuilder> routeBuilderClass = (Class<RouteBuilder>)camelESBClassLoader.loadClass(mustLoadedClassName);
            if(routeBuilderClass != null) {
                RouteBuilder routeBuilder = routeBuilderClass.newInstance();
                camelContext.addRoutes(routeBuilder);
            }
        }
    }
    ......
}

以上代码关于“SynchronousQueue”同步队列的使用中,主线程将一直试图使用同步队列的take()方法取得数据,如果没有数据主线程就将会等待在这里。这里主要进行Route路由定义的动态加载,但是在实际应用中,还应该根据“SynchronousQueue”同步队列中数据的状态判断是进行Route路由定义的加载还是卸载。

我们还需要对运行BootStartup实例的主线程进行ClassLoader的重设置。这是因为每一个业务系统在本地都有一个独立的class文件的存储位置,但是这个位置并不在JVM默认的加载路径中。JVM的ClassLoader组织层次最顶层的Bootstrap Loader位置和各上层的Extension Loader都不会加载以上代码中的“/testCamelRoot/XXXXX”这个位置。所以我们需要编写代码进行干预,如下图所示:

这里写图片描述

====================================================
(接下文)

==================================================== (接上文《架构设计:系统间通信(41)——自己动手设计ESB(2)》)4-2-3、业务层与Class文件管理zookeeper控制模块的业务相关层

相关阅读排行


用户评论

游客

相关内容推荐

最新文章

×

×

请激活账号

为了能正常使用评论、编辑功能及以后陆续为用户提供的其他产品,请激活账号。

您的注册邮箱: 修改

重新发送激活邮件 进入我的邮箱

如果您没有收到激活邮件,请注意检查垃圾箱。