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

注册 | 登录

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

==================================================== (接上文《架构设计:系统间通信(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);

}

}

// 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”这个位置。所以我们需要编写代码进行干预,如下图所示:==================================================== (接下文)

$(function () {

$('pre.prettyprint code').each(function () {

var lines = $(this).text().split('\n').length;

var $numbering = $('<ul/>').addClass('pre-numbering').hide();

$(this).addClass('has-numbering').parent().append($numbering);

for (i = 1; i <= lines; i++) {

$numbering.append($('<li/>').text(i));

};

$numbering.fadeIn(1700);

});

});

相关推荐:基于ActiveMQ的消息中间件系统 OneMM逻辑与物理架构设计详解

维基百科对消息中间件的定义是“Message-oriented Middleware is software infrastructure focused on sending and receiving messages between distributed sys

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

相关阅读排行


用户评论

游客

相关内容推荐

最新文章

×

×

请激活账号

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

您的注册邮箱: 修改

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

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