如何利用Websphere MQ实现大文件交换

本文介绍了如何利用Websphere MQ提供的机制，实现大文件断点续传，从而可以在低速网络的情况下，不会因为网络的故障而使整个大文件重新传输。本文介绍的这个实现可以大大的提高业务处理效率，并且可以对本文提供的思路进行扩展以完成更加复杂的功能。1 引言在低速的网络环境中，上传或者下载一个大文件需要花费很长的时间并且网络出现故障的机率很高。如果网络出现故障的话，这个传输将会从头开始；如果经常出现网络故障的话，大文件的传输可能永远不会成功。为了解决这个问题，实现断点续传功能的软件产品由此产生，比较著名的有网络蚂蚁，迅雷，BT，电驴。Websphere MQ是一种消息中间件，Websphere MQ Server与Websphere MQ Server之间的数据传输实现了断点续传的功能，从而保障了传输的可靠性。但是Websphere MQ 客户端应用与Websphere MQ Server之间没有提供断点续传的功能。并且在很多的应用场合，Websphere MQ 客户端应用是通过低速网络与Websphere MQ Server通讯以完成数据的传输；所以我们需要实现一个机制来实现Websphere MQ客户端应用与Websphere MQ Server之间的断点续传，并且提供简单的应用编程接口来给第三方使用。采用Websphere MQ来实现断点续传的功能的话，还可以利用Websphere MQ提供的加密机制来保障传输的安全性，以及在HTTP协议上传输以穿透防火墙等功能,从而使我们的断点续传功能更加具有竞争性。2 术语定义Websphere MQ数据交换网络：表示多个Websphere MQ队列管理器进行配置形成一个完整的数据交换环境。Websphere MQ 客户端应用：表示与Websphere MQ进行交互的应用程序。MQ Client App: Websphere MQ 客户端应用的英文简称。IBM Content Manager V8:IBM提供的一个内容管理平台。3 Websphere MQ大文件交换应用架构在利用Websphere MQ实现大文件传输的应用系统中，存在以下几种架构。3.1 单个Websphere MQ Server在只有一个Websphere MQ Server的也存在两种架构，分别如下所示：图3.1.1 断点传输应用架构一这张架构图描述了Websphere MQ 客户端断点上传应用、Websphere MQ Server、MQ Websphere MQ 客户端断点下载应用分别部署在不同的机器上进行文件的交换工作。图3.1.1 断点传输应用架构二这张架构图描述了Websphere MQ 客户端断点上传应用部署在一台机器上而Websphere MQ Server和Websphere MQ 客户端断点下载应用在同一台机器上进行文件的交换工作。3.2 多个Websphere MQ Server图3.2.1 断点传输应用架构三这张架构图描述了在整个系统中存在多个Websphere MQ队列管理器,这些队列管理器形成一个完整的Websphere MQ数据交换网络，而Websphere MQ 客户端断点上传应用、Websphere MQ 客户端断点下载应用分别部署于网络中任意一台机器上。图3.2.2 断点传输应用架构四这张架构图描述了在整个系统中存在多个Websphere MQ队列管理器,这些队列管理器形成一个完整的Websphere MQ数据交换网络，而Websphere MQ 客户端断点上传应用、Websphere MQ 客户端断点下载应用分别部署于网络中任意一台机器上。4 功能对于大文件传输的应用，我们主要实现以下表格所示的功能。功能点功能子点大文件传输没有确定的接收者有确定的一个接收者有确定的多个接收者 目录传输没有确定的接收者有确定的一个接收者有确定的多个接收者 分块传输多线程并发分快传输 系统扩展调用用户自定义处理程序 系统纠错机制网络中断纠错机制客户端应用进程突然死掉纠错机制MQ Server进程突然死掉纠错机制表格4.1 功能列表1、 单个大文件传输Websphere MQ 客户端断点上传应用把本机存在的一个大文件断点上传到Websphere MQ Server中。Websphere MQ 客户端断点下载应用把存储在Websphere MQ中的大文件断点下载到本地，并且可以根据需求把这个大文件存储到IBM Content Manager V8中。而传输大文件的这个功能点又可以细分为两个小的功能点：l Websphere MQ 客户端断点上传应用上传的大文件没有确定的接收者，这样的话任何的Websphere MQ 客户端断点下载应用都可以把存放在Websphere MQ Server中的这个大文件下载到本地。l Websphere MQ 客户端断点上传应用上传的大文件有确定的接收者，这样的话只有按照某个用户登陆进入的Websphere MQ 客户端断点下载应用才可以从Websphere MQ Server断点下载此大文件到本地。2、 目录传输Websphere MQ 客户端断点上传应用把本机存在的一个目录断点上传到Websphere MQ Server中。Websphere MQ 客户端断点下载应用把存储在Websphere MQ中的目录断点下载到本地，并且可以根据需求把这个目录存储到IBM Content Manager V8中。这个功能其实是在单个大文件传输这个功能点的基础上进行更高一层的抽象。这个功能点如同单个大文件传输一样可以细分成两个功能点：l Websphere MQ 客户端断点上传应用上传的目录没有确定的接收者，这样的话任何的Websphere MQ 客户端断点下载应用都可以把存放在Websphere MQ Server中的这个目录下载到本地。l Websphere MQ 客户端断点上传应用上传的目录有确定的接收者，这样的话只有按照某个用户登陆进入的Websphere MQ 客户端断点下载应用才可以从Websphere MQ Server断点下载此目录到本地3、分块传输这个功能点主要是针对在传输很大文件（如2G的文件）的时候，如果仅仅建立一个网络连接的话，相应的环境会限制这个网络连接最大的传输速度。所以为了尽快地提高传输速度，需要启动多个线程以建立多个网络连接，每个线程传输大文件中的某段数据来实现并行上传和下载，这个功能点如同网络蚂蚁中多线程下载的功能。4、调用用户自定义处理程序这个功能点主要对应的是这样的一种情况， Websphere MQ 客户端断点下载应用从Websphere MQ 中下载下来相应的大文件和目录后，需要对这些文件数据进行处理，然而对这些文件数据的处理逻辑程序是和用户的需求密切相关的，如处理逻辑程序就是完成把下载下来的大文件和目录装载到IBM Content Manager V8中。这样的话就需要系统提供一种能够调用用户自定义开发的处理逻辑程序的能力，从而可以扩展此系统的功能。5、系统纠错机制在现实情况下传输中会出现以下几种情况：1、 网络中断：出现这种情况，传输将中断。这样的话就需要系统提供一种机制从失败的那一点开始传输。也就是说应用如果重新调用这个传输接口，传输将会从失败的那一点继续传输，而不是从开始传输。2、 客户端应用进程突然死掉：出现这种情况，传输将中断。并且在应用进程中保存的一些信息将丢失，这样的话就需要传输接口提供一种信息持久化机制把相应的信息持久化到本地硬盘中（如果本地硬盘损坏的话，系统将还是从头开始传输），然后应用进程重新启动，传输将从失败的那一点继续传输。当然这一点的实现可能会导致一些性能的损耗。3、 Websphere MQ Server进程突然死掉：Websphere MQ Server提供了持久话机制，如果利用了MQ Server的持久化机制，Websphere MQ Server重新启动后；传输将从失败的那一点继续传输。如果没有,则传输将还是从头开始。我们必需实现相应地机制，来对这些情况进行相应地纠错处理。5 实现5.1 消息设计为了实现断点续传功能，在上传的时候我们需要把一个大文件进行拆分，形成大量的消息并把这些消息放入Websphere MQ中从而实现断点上传，在下载的时候我们从Websphere MQ中取出相应的消息并进行组装形成相应的大文件从而实现断点下载。我们需要对文件进行拆分成多个消息，这样我们就需要对这个消息所包含的内容进行设计，程序根据这些消息格式和程序代码本身的控制逻辑来实现可靠地传输。5.1.1 文件传输控制消息(FileTransControlMsg)在传输大文件的时候，我们采用把文件进行分段（如每段的大小为4K，当然这个是一个可以进行配置的选项）的机制，为了实现这种机制我们需要一个控制消息来进行相应地控制。这个控制消息将存放在FileTrans.Control 这个Websphere MQ队列中。每传送一个文件只会在这个队列中产生一条消息。在java/c+的实现中的类名为：FileTransControlMsg.字段名概述 大小 补充说明Msgid字段的值为(通过对FileDir+ FileName+ HostName+Ipaddress的值进行SHA-1数字摘要算法得到)+ ChunkNum，这个值用来唯一表示这个文件的ID号（因为只要FileDir、FileName、 HostName、Ipaddress的值不同则生成的SHA值肯定不同），这个值将用来在Websphere MQ中进行相应地检索。我们将利用MQMD中的Msgid字段SegSize文件分段传输的大小，表示发送大文件以这个字段的值为数据传输单位。4字节用字符串来描述单位为KFileDir要发送的文件所属的目录512字节FileName要发送的文件名256字节ChunkFileName文件分块传输后某块的临时文件名256字节ChunkCount表示要传输的原始文件分成了多少块来进行并发传输2个字节用字符串来描述.如果这个值为1的话将说明大文件的传输将不会分块并发执行传输。并且ChunkFileName的值和FileName的值将一样ChunkNum是第几个分块2个字节描述这个分块文件是第几个分块FileDataSize要发送分块大文件的数据总大小32字节即描述ChunkFileName对应的大小Offset偏移量32字节如果要对文件分块传输的时候，这个分块文件对应的文件的开始偏移HostName发送者的主机名称128字节IpAddress发送者的主机IP地址32字节接收者接收完此文件后将在它的接收目录建立HOSTNAMEIPADDRESS这样的一个目录Receiver接收者32字节这个是对接收用户名进行SHA-1算法运算后的值QueueName存放大文件数据的Websphere MQ队列名64字节TradeCode交易码8个字节如果此字段的值不为”的话，则接收者将调用相应的处理程序进行处理 表格文件传输控制消息格式定义5.1.2 文件传输数据消息(FileTransDataMsg)这个消息将存放在按照FileTransControlMsg这个消息结构中定义的QueueName字段对应的队列中，通过这种方式可以实现按照用户的需要把相应的消息放在所设定的队列中。在java/c+的实现中的类名为：FileTransDataMsg.格式定义：字段名概述 大小 补充说明Msgid这个字段的值等同于FileTransControlMsg消息中的Msgid字段我们将利用MQMD中的Msgid字段SegNumber表示此消息对应的数据区中的数据是文件中第几块第几段的数据（从1开始增长）我们将利用MQMD中的CorrelId字段MsgDataSize这个消息中容纳具体文件数据的大小8字节用字符串来描述Offset这个字段对应于FileTransControlMsg消息中的ChunkFileName字段描述开始要发送的文件的偏移量32字节用字符串来描述DataSection数据区等于MsgDataSize所描述的大小,并且小于等于FileTransControlMsg 消息中SegSize中所描述的大小表格文件传输数据消息格式定义5.1.3 上传完成控制消息(SendFinishControlMsg)当上传应用完成消息的断点上传的时候，将发送一个上传完成控制消息(SendFinishControlMsg)到队列FileTrans.Control.Finish中，下载应用可以从Websphere MQ队列中获取能够下载的文件列表，从而完成组装成一个大文件以供后续处理（如装载到IBM Content Manager V8中等）。在java/c+的实现中的类名为：SendFinishControlMsg.字段名概述 大小 补充说明Msgid这个字段的值等同于FileTransControlMsg消息中的Msgid字段我们将利用MQMD中的Msgid字段Receiver这个字段的值等同于FileTransControlMsg消息中的Receiver字段我们将利用MQMD中的CorrelId字段ChunkCount总共有几个分块2个字节描述文件被分成几个文件传输了ChunkNum是第几个分块2个字节描述这个分块文件是第几个分块Desc描述信息512字节存放一些基本的描述信息TradeCode交易码8个字节如果此字段的值不为”的话，则接收者将调用相应的处理程序进行处理表格上传完成控制消息格式定义5.2 数据结构设计此数据结构设计的目地在于让程序能够完成相应的纠错机制（如网络纠错、程序异常终止等情况的纠错）。这些数据结构对应的信息根据相应地配置文件的设定，以决定是否持久化到硬盘中。5.2.1 上传文件控制结构(SendFileContorl)此结构描述了上传应用在发送文件的时候相应的进度信息。在java/c+的实现中的类名为：SendFileContorl.字段名概述 大小 补充说明ID字段的值为(通过对FileDir+ FileName+ HostName+Ipaddress的值进行SHA-1数字摘要算法得到)+ ChunkNum，这个值用来唯一表示这个文件的ID号（因为只要FileDir、FileName、 HostName、Ipaddress的值不同则生成的SHA值肯定不同），这个值将用来在Websphere MQ中进行相应地检索。这个做为这个结构持久化到文件系统中的文件名的一部份SegSize文件分段传输的大小，表示发送大文件以这个字段的值为数据传输单位。4字节用字符串来描述单位为KChunkFileName文件分块传输后某块的临时文件名256字节ChunkCount表示要传输的原始文件分成了多少块来进行并发传输2个字节用字符串来描述.如果这个值为1的话将说明大文件的传输将不会分块并发执行传输。并且ChunkFileName的值和FileName的值将一样ChunkNum是第几个分块描述这个分块文件是第几个分块DataSize要发送分块大文件的数据总大小32字节即描述ChunkFileName对应的大小Offset偏移量如果要对文件分块传输的时候，这个分块文件对应的文件的开始偏移SendedDataSize已经发送数据的大小32字节用字符串来描述.SendedSegNum已经发送的数据段的数目16字节用字符串来描述.Finished是否已经完成1个字节1:完成0:没有完成 表格上传文件控制结构定义这个结构将按照配置文件的需要被持久化到文件系统中（持久化的文件名称:ID字段的值.IBMMQ）5.2.2 下载文件控制结构(RecvFileControl)在java/c+的实现中的类名为：RecvFileControl.字段名概述 大小 补充说明SegSize文件分段传输的大小，表示发送大文件以这个字段的值为数据传输单位。4字节用字符串来描述单位为KFileDir要发送的文件所属的目录512字节FileName要发送的文件名256字节ChunkFileName文件分块传输后某块的临时文件名256字节ChunkCount表示要传输的原始文件分成了多少块来进行并发传输2个字节用字符串来描述.如果这个值为1的话将说明大文件的传输将不会分块并发执行传输。并且ChunkFileName的值和FileName的值将一样ChunkNum是第几个分块2个字节描述这个分块文件是第几个分块FileDataSize要发送这个文件的数据的总大小32字节即描述ChunkFileName对应的大小Offset偏移量32字节如果要对文件分块传输的时候，这个分块文件对应的文件的开始偏移HostName发送的主机名称128字节IpAddress发送的主机IP地址32字节接收者接收完以后将在它的接收目录建立HOSTNAMEIPADDRESS这样的目录Receiver接收者32字节这个是对接收的用户名的SHA-1算法运算后的值QueueName要传送数据存放的队列64字节TradeCode交易码8个字节如果此字段的值不为”的话，则接收者将调用相应的处理程序进行处理Desc描述信息512字节存放一些基本的描述信息RecvedDataSize已经接收的数据大小32字节用字符串来描述.RecedSegNum已经接收的数据段的数目16字节用字符串来描述.Finished是否已经完成1个字节1:完成0:没有完成表格上传文件控制结构定义5.3控制流程逻辑设计 这个主要阐述了此程序内部实现的大致流程。 图5.3.1 系统大致流程图5.3.1初始化流程对应的方法：Init。Step1、读取相应地配置文件，初始化相应地信息，完成如下之类的工作，l 是否对多大的文件启动分块传输（并且分成多少块并发的开始传输）l 是否启用MQ 消息持久化机制l 是否启用程序控制信息（SendFileContorl信息）的持久化机制。Step2、初始化SHA-1算法引擎，从而为能够更好更快地进行SHA值运算做准备。Step3、获得用户机器的IP地址（IpAddress）和机器名(HostName)。5.3.2.上传文件流程对应的方法：SendFile(FileName,RecevierName).Step1、对FileName、IPAddress、HostName的值进行组合,并且通过SHA-1运算获得一个唯一标志符（ID）。Step2、根据这个ID判断在内存中是否存在SendFileContorl对应的结构信息，如果不存在的话，转Step2.1 ；如果存在转Step3。Step2.1、在文件系统中找是否存在一个/或者多个(ID表示的值).IBMMQ对应的文件，如果存在的话（表明应用进程重新启动过）.然后读取这个文件的内容初始化到SendFileContorl结构体中并在内存中缓冲，然后转Step3。 如果不存在的转step2.2。 Step2.2、重新生成此要传输文件要对应的SendFileContorl信息（根据系统配置信息判断是否需要生成多个SendFileContorl信息。Step3、一个SendFileContorl结构信息由一个线程进行处理，如果存在多个SendFileContorl结构信息则启动多个线程进行处理。Step4、开始启动相应的一个或者多个发送线程，把SendFileContorl结构信息传给发送线程。Step5、同步等待发送线程返回。 发送者线程流程Step1、根据SendFileContorl结构中的msgid到FileTrans.Control队列中查询是否存在相应地消息，如果不存在的话，则需要把SendFileContorl结构中相应的值全部初始化成缺省值（这个表示对应的数据将从0开始传输）,并且往FileTrans.Control队列中打入相应地消息。然后转Step2。如果存在的话则转Step1.1.Step1.1、把1与msgid信息组合，到相应地消息队列中判断消息是否存在，如果存在的话，则转Step1.2，否则表示对应的数据要从0开始传输，把SendFileContorl结构中的SendedSegNum的值更新为0。Step1.2、从SendFileContorl结构中取得SendedSegNum字段信息，并且与msgid信息组合到相应地消息队列中判断消息是否存在，如果不存在则把SendedSegNum字段的值减1。转Step2Step2、从SendFileContorl结构中取得SendedSegNum字段、SegSize字段、Offset字段的信息，如果传输完成的话则转Step3。 否则根据这些信息从文件中读取内容，把这个内容组织成FileTransDataMsg消息，打入到相应地队列中。如果打入成功，则更新SendFileContorl结构中内容并且按照配置信息决定是否刷新到文件系统中。转step2 .如果失败的话，则退出。Step3、组织SendFinishControlMsg消息放入到FileTrans.Control.Finish队列中5.3.3 下载文件流程 对应的方法：RecevieFile(RecevierName);这个流程基本上同发送文件流程，详见源代码. 接收者线程流程这个流程基本上同发送者线程流程，详见源代码. 5.4 对外接口设计对外接口的设计尽量简单和数目较少，并且这个程序现在只实现了JAVA版本，如果要被C/C+程序调用的话，可以采用JNI的方式。相应的对外可以被其它程序调用的接口定义如下。MQTransInterface 是这个接口的实现类名。l 方法：initFromConfig(String configFileName)此方法是静态方法，在调用MQ断点续传接口时仅仅需要调用此接口一次configFileName：是所需要的配置文件的绝对路径名l 方法：int sendFile(String absolutFileName,String receiverName,TransInfo transInfo)发送文件到WEBSPHERE MQ中。参数：absolutFileName：要发送的文件名receiverName：此文件的接收者，现在使用时请置NULLtransInfo:存储了发送的字节数目等信息，使用此方法时，每使用一次请创建此对象一次。返回值：/* * -6:没有足够的可用线程来启动分块传输，请过段时间重试 * -5:没有空闲的服务线程服务 * -4:要发送的文件不能够读取，不能够继续 * -3:要发送的不是文件 * -2:要发送的文件不存在 * -1:失败 * 0:成功 * 1:无法取得与MQ队列管理器的连接 * 2:与MQ队列管理器的连接无效，即出现网络失效错误 */l 方法：List getReceiveFileList(String receiverName,int count) 从WEBSPHERE MQ中取得可以下载的文件列表参数： receiverName：接收者，使用时请置NULL count：需要此接口的返回数目，如置上50返回值： 返回能够接收的文件列表对象，此列表包含ReceveFileData对象ReceveFileData对象的定义：Msgid：唯一标志能够接收的文件IDChunkcount：此文件被分成几段上传的，也就意味着将分成几段同时下载fileLength：此文件的长度desc：此文件的描述信息,此信息的格式为：主机名+:+IP地址+:+文件名;tradeCode：需要调用外部类的交易ID，现在不支持l 方法:String recvFile(ReceveFileData receveFileData,String dirName,String receiverName,TransInfo transInfo,List errorList)接收文件，调用此方法前 需要先调用getReceiveFileList参数：receveFileData:调用getReceiveFileList方法返回值中对应的某个对象，指明需要下载的文件。dirName：指定下载下来的文件需要存放的路径transInfo: 存储了接收的字节数目等信息。使用此方法时，每使用一次请创建此对象一次。errorList：错误信息的一些列表，主要是对应分段下载，有多个信息对象。6 如何编译和修改此系统此系统是采用JAVA语言，eclipse IDE环境实现，在相应的工程目录中，src目录是源代码的目录，lib 目录是系统所需要的外部库的目录（需要把这个目录中相应的JAR文件加入到ECLIPSE环境中）。建议使用JDK1.5。7 如何使用此系统l 导入com.ibm.mqtrans. MQTransInterface类、com.ibm.mqtrans.model.TransInfo类、com.ibm.mqtrans.model.TransInfo，com.ibm.mqtrans.model .ReceveFileData类.在你的java程序中调用5.4 对外接口设计所列出来的接口就可以了。l 编辑系统提供的配置文件，此配置文件的定义查看: sysparamconfig.xml 文件。l 系统初始化1、 安装WEBSPHERE MQ，并建立相应地队列管理器（如TESTQM），并且建立一个侦听器（如侦听1414端口）。2、 接口的日志信息记录是采用LOG4J库，所以如果更改相应的日志记录信息请修改log4j的配置文件3、 打开配置文件sysparamconfg.xml文件，找到queuename、filetranscontorlqueuename、filetranscontorlfinishqueuename三个节点修改相应的值或者不进行修改。并在队列管理器中建立此三个节点对应的值的队列名，（特别是需要把filetranscontorlqueuenam