Flex与java进行socket通信完成以断点续传方式的文件上传

简单的文件上传是将整个文件在一起此请求中将文件上传至服务器中，而对户外作业平台而言，网络的稳定性是一个问题，加之平台的大文件性，故而在该平台中采用断点续传方式来上传文件较为合适。

断点续传简单来讲，客户端与服务器端通信，了解到已传输文件的大小后，而后在按照一定的文件块大小对剩余未传输的文件进行分块传输，服务器端则将这些块文件一点点写入到同一个文件中，从而形成一个完成文件，达到断点续传的目的。客户端逻辑示意如下所示：



Flex的文件上传下载会使用到FileReference，而在FlashPlayer10以后提供了Load方法和data属性。在FileReference执行完load方法后，data中便存储了FileReference所代表的文件的数据，其为ByteArray对象，即我们可以以字节的形式访问Flex加载的文件，而这也为在断点续传文件时对文件进行分块处理提供了可能性。

Flex与java通信上传文件的实现方式可以有多种，本文采用socket通信方式来传输文件。Flex端代码如下所示：

[html] view
plaincopyprint?

<span style="font-size:12px;">//利用socket

socket = new Socket;

socket.connect("12.156.53.29",1234); //连接服务器

//监听是否连接

socket.addEventListener(Event.CONNECT,function conn(){

//发送名称

socket.writeUTF(fileReference.name);

socket.flush();

//文件大小

socket.writeUTF(String(fileReference.data.length));

socket.flush();

});

//监听接受数据

socket.addEventListener(ProgressEvent.SOCKET_DATA,function receiveData(){

//已上传大小

var len:String = socket.readUTF();

if(len == "0"){

if(fileReference.data.length < 1024){

socket.writeBytes(fileReference.data);

}else{

socket.writeBytes(fileReference.data,0,1024);

}

socket.flush();

}else{

if((fileReference.data.length - uint(len)) > 1024){

socket.writeBytes(fileReference.data,uint(len),1024);

}else{

socket.writeBytes(fileReference.data,uint(len), fileReference.data.length - uint(len));

}

socket.flush();

}

});

</span><pre name="code" class="html">//监听连接关闭

socket.addEventListener(Event.CLOSE,functioncloseConn(){ });

</pre>

在flash.net包中存在Socket类，在文档的描述中，我们可以了解到socket通信需要使用套接字策略文件。在实际的socket通信过程中，我们在客户端不管发送什么信息，在服务端的socket第一次接收的信息都会是<policy-file-request/>。这个信息是在要求服务端提供给客户端socket通信的策略文件，在该文件中指定通信的端口号等信息。这个过程涉及到Flash Player安全机制，不过多讲述，了解到flex进行跨域socket通信时默认必须要在843端口上接收Flash
Player的策略文件请求。此处需要注意的是对策略文件的请求和断点续传过程主动发起的请求同服务端的socket连接是两个独立的连接，在处理完策略文件请求连接后，我们要关闭策略文件请求的连接，这样Flash Player会自动重新连接，从而可实现断点续传的socket连接，否则我们的主动请求将无法连接上。

针对上面术的内容，我做了这样处理：建立两个两个listener，一个监听对策略文件的请求，一个监听对断点续传socket连接的请求，后者是我们的主请求。在每个监听器使用多线程处理，每次接受到socket连接请求，就会创建一个线程用于处理策略文件请求或者断点续传请求。

Web.xml中配置listener

[html] view
plaincopyprint?

<span style="font-size:12px;"><!-- 大文件上传端口监听器 -->

<listener>

<display-name>myListener</display-name>

<listener-class>com. fileoperation.LargeFileUploadListener</listener-class>

</listener>

<!-- 大文件上传端口安全策略监听器 -->

<listener>

<display-name>policyListener</display-name>

<listener-class>com. fileoperation.LargeFileUploadPolicyListener</listener-class>

</listener></span>

LargeFileUploadPolicyListener.java:

[java] view
plaincopyprint?

package com.fileoperation;

import java.net.ServerSocket;

import java.net.Socket;

import javax.servlet.ServletContextEvent;

import org.apache.log4j.Logger;

public class LargeFileUploadPolicyListener extends javax.servlet.http.HttpServlet implements javax.servlet.ServletContextListener{

private static final long serialVersionUID = 1L;

private static final Logger log = Logger.getLogger(LargeFileUploadListener.class);

private static Thread thread = null;

@SuppressWarnings("deprecation")

public void contextDestroyed(ServletContextEvent arg0) {

if(thread != null){

thread = null;

}

}

public void contextInitialized(ServletContextEvent arg0) {

try {

thread = new Thread() {

public void run() {

log.info("大文件上传侦听开始。。。。");

try{

ServerSocket policyServerSocket= new ServerSocket(Integer.parseInt("843"));//服务器套接字

Socket policyClientSocket = null;

Socket clientSocket=null;

while(true){

policyClientSocket = policyServerSocket.accept(); //获得客户端的请求的Socket

log.info("已侦听到了客户端的请求。。。。。");

new MyPolicyServerThread(policyClientSocket);

}

}catch (Exception e) {

log.error("接收大文件异常：",e);

}

}

};

thread.start();

} catch (Exception e) {

log.error("启动监听器异常：",e);

}

}

}

MyPolicyServerThread.java:

[java] view
plaincopyprint?

package com. fileoperation;

import java.io.BufferedReader;

import java.io.IOException;

import java.io.InputStreamReader;

import java.io.PrintStream;

import java.io.UnsupportedEncodingException;

import java.net.Socket;

import org.apache.log4j.Logger;

public class MyPolicyServerThread extends Thread {

private static final Logger log = Logger.getLogger(MyServerThread.class);

private Socket socket;

private final String policy_xml = "<policy-file-request/>";

private final String cross_xml = "<?xml version=\"1.0\"?>" +

"<cross-domain-policy>" +

"<site-control permitted-cross-domain-policies=\"all\"/>" +

"<allow-access-from domain=\"*\" to-ports=\"1234\"/>" +

"</cross-domain-policy>\0";

public MyPolicyServerThread(Socket socket) {

this.socket = socket;

this.start();

}

@Override

public void run() {

try {

BufferedReader readerIn = new BufferedReader(new InputStreamReader(socket.getInputStream()));

PrintStream printOut = new PrintStream(socket.getOutputStream());

char[] by = new char[22];

readerIn.read(by, 0, 22);

String s = new String(by);

if (s.equals(policy_xml)) {

System.out.println("接收policy-file-request认证");

printOut.print(cross_xml);

printOut.flush();

readerIn.close();

printOut.close();

socket.close();

System.out.println("完成policy-file-request认证");

}

} catch (UnsupportedEncodingException e) {

e.printStackTrace();

} catch (IOException e) {

e.printStackTrace();

}

}

}

上面是策略文件socket连接的处理过程，此过程是Flash Player默认发起的，在客户端无需我们人工干预，只需要在服务端进行处理即可，在上述策略文件中我们发现，其指定了to-ports=\"1234\"，即指明了socket主动请求的socket端口号为1234，而在关闭此策略socket连接后，FlashPlayer重新连接的端口号即为1234了。下面的服务端处理过程是与我们的主动请求（断点续传）相关联的，在上文断点续传逻辑图中，只写明了客户端的逻辑，而服务端的逻辑实际上是与之相对对应的，而核心是java.io.RandomAccessFile将文件分批写入相应文件下。下面代码中发起的线程执行情况是与前文所讲Flex端的socket相互通信共同实现断点续传上传文件功能的。

LargeFileUploadListener.java

[java] view
plaincopyprint?

package com. fileoperation;

import java.net.ServerSocket;

import java.net.Socket;

import javax.servlet.ServletContextEvent;

import org.apache.log4j.Logger;

public class LargeFileUploadListener extends javax.servlet.http.HttpServlet implements javax.servlet.ServletContextListener{

private static final long serialVersionUID = 1L;

private static final Logger log = Logger.getLogger(LargeFileUploadListener.class);

private static Thread thread = null;

@SuppressWarnings("deprecation")

public void contextDestroyed(ServletContextEvent arg0) {

if(thread != null){

thread = null;

}

}

public void contextInitialized(ServletContextEvent arg0) {

try {

thread = new Thread() {

public void run() {

log.info("大文件上传侦听开始。。。。");

try{

ServerSocket serverSocket= new ServerSocket(Integer.parseInt("1234"));//服务器套接字

Socket clientSocket=null;

while(true){

clientSocket= serverSocket.accept();//获得客户端的请求的Socket

log.info("已侦听到了客户端的请求。。。。。");

new MyServerThread(clientSocket);

}

}catch (Exception e) {

log.error("接收大文件异常：",e);

}

}

};

thread.start();

} catch (Exception e) {

log.error("启动监听器异常：",e);

}

}

}

MyServerThread.java:

[java] view
plaincopyprint?

package com.ibm.rise.workplace.fileoperation;

import java.io.DataInputStream;

import java.io.DataOutputStream;

import java.io.File;

import java.io.FileInputStream;

import java.io.IOException;

import java.io.InputStream;

import java.io.RandomAccessFile;

import java.net.Socket;

import java.nio.channels.FileChannel;

import java.nio.channels.FileLock;

import java.util.Map;

import java.util.StringTokenizer;

import org.apache.log4j.Logger;

public class MyServerThread extends Thread {

private static final Logger log = Logger.getLogger(MyServerThread.class);

private Socket socket;

public MyServerThread(Socket socket) {

this.socket = socket;

this.start();

}

@Override

public void run() {

DataInputStream dataInputStream = null;

DataOutputStream dataOutputStream = null;

RandomAccessFile randomAccessFile = null;

boolean isInfoSubmission = false;

Document docInfoSubmission = null;

Double totalSize = 0.0;

DocProperty docProperty = null;

try {

dataInputStream = new DataInputStream(socket.getInputStream());

dataOutputStream = new DataOutputStream(socket.getOutputStream());

//读取名称

String fileName = dataInputStream.readUTF();

String fileSize = dataInputStream.readUTF();

//检测上传文件是否存在

String FilePath = “file path”; //可使用配置文件形式将路径写清楚

StringTokenizer st = new StringTokenizer(FilePath.toString(),"/");

String toAddPath = st.nextToken()+"/";

String toTestPath = toAddPath;

while(st.hasMoreTokens()){

toAddPath = st.nextToken()+"/";

toTestPath += toAddPath;

File inbox = new File(toTestPath);

if(!inbox.exists()) {

inbox.mkdir();

}

}

//检测上传位置

File file = new File( FilePath + "/" + fileName);

long position = 0;

if(file.exists()){

position = file.length();

}

//通知客户端已传大小

dataOutputStream.writeUTF(String.valueOf(position));

dataOutputStream.flush();

byte[] buffer = null;

int read = 0;

while(true){

//检测上传位置

file = new File( FilePath + "/" + fileName);

position = 0;

if(file.exists()){

position = file.length();

}

//rw代表写流(随机读写）

randomAccessFile = new RandomAccessFile(file,"rw");

FileLock fileLock = null;

FileChannel fileChannel = null;

fileChannel = randomAccessFile.getChannel();

fileLock = fileChannel.tryLock();

//拿到了文件锁,写入数据

if(fileLock != null){

randomAccessFile.seek(position);

read = 0;

buffer = new byte[1024];

read = dataInputStream.read(buffer);

randomAccessFile.write(buffer,0,read);

if(fileLock != null){

fileLock.release();

fileLock = null;

}

if(randomAccessFile != null){

randomAccessFile.close();

randomAccessFile = null;

}

}

//检测已上传的大小

file = new File( FilePath + "/" + fileName);

position = 0;

if(file.exists()){

position = file.length();

}

System.out.println("文件 " + fileName + " 已传输 " + String.valueOf(position)+ "字节");

//判断文件是否传输完成

if(position >= Long.parseLong(fileSize)){

//文件传输完成

System.out.println("文件 " + fileName + " 已传输完毕，总大小为" + String.valueOf(position) + "字节");

break ;

}else{

//通知客户端已传大小

dataOutputStream.writeUTF(String.valueOf(position));

dataOutputStream.flush();

}

} // END WHILE

//跳出while循环，即已结束文件上传，则终止socket通信

dataInputStream.close();

dataOutputStream.close();

socket.close();

} catch (IOException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}

}

flex利用socket通信，并不需要像通过remoteObject方式访问java端服务一样配置相关服务，只需要在flex端的mxml或as文件中进行socket连接时写明连接的ip和端口即可进行通信，但需要注意的是需要对FlashPlayer所要求的策略文件进行相应处理。此外断点续传所要做的核心内容是将整个文件在一次请求中传递改为多次请求中分块传递，服务端利用相关类对分块文件进行整合即可。