Netty文件上传断点续传的演示
netty文件上传采用自定义的协议方式实现，断点续传主要是依据RandomAccessFile类的随机读写能力，主要流程是客户端发起请求，将需要上传文件名称、路径、读取文件的数据、以及读取文件的起始位置等等信息，并且缓存在服务端中（以文件路径为key，自定义协议对象为value），服务端拿到客户端发送的上述数据，就会写文件，并且写完文件，也会记录写过数据位置等信息，再次发送信息给客户端下一次需要读取的数据。
假如这个过程中，客户端断开了链接，此时由于服务端缓存了已经写过文件的位置，那么只会从写过文件的位置进行读文件，再传文件给服务端写等循环操作，从而达到了断点续茶传的效果。
* RandomAccessFile的基本api
* .getFilePointer : 获取当前操作的位置
* .seek(index): 将操作位置设置到index
* .read(byte):读文件到byte数组中，返回读取文件的长度
*
* 构造函数
* rf = new RandomAccessFile(new File(filePath),"r"); // 参数2是模式
*
* 模式详解:
* “r” 以只读方式来打开指定文件夹。如果试图对该RandomAccessFile执行写入方法，都将抛出IOException异常。
* “rw” 以读，写方式打开指定文件。如果该文件尚不存在，则试图创建该文件。
* “rws” 以读，写方式打开指定文件。相对于”rw” 模式，还要求对文件内容或元数据的每个更新都同步写入到底层设备。
* “rwd” 以读，写方式打开指定文件。相对于”rw” 模式，还要求对文件内容每个更新都同步写入到底层设备。
* 测试1：下面演示基本的读文件操作。
*/
@Test
public void test1() throws IOException {
String filePath = "src/d00_file/a.txt";
RandomAccessFile rf = null;
try {
rf = new RandomAccessFile(new File(filePath), "r");
System.out.println("输入内容：" + rf.getFilePointer());
// 从10位置操作
rf.seek(10);
byte[] b = new byte[1024];
int len = 0;
// 循环读写 (len是读取的长度)
while ((len = rf.read(b)) > 0) {
System.out.print(new String(b, 0, len));
}
} catch (IOException e) {
e.printStackTrace();
} finally {
rf.close();
}
}
/**
* 测试2： 向文件追加内容
*/
@Test
public void test2() throws IOException{
String filePath="src/d00_file/a.txt";
RandomAccessFile rf=null;
try {
rf = new RandomAccessFile(new File(filePath), "rw");
// 将操作指针移动到文件的最后
rf.seek(rf.length());
// 向文件写出数据
rf.write("这是追加的内容。。".getBytes());
}catch (IOException e){
e.printStackTrace();
}finally {
rf.close();
}
}
/**
* 测试3：修改文件内容
*/
@Test
public void test3() {
RandomAccessFile rf = null;
String oldStr = "www.www.www";
String newStr = "hahahaha";
try {
rf = new RandomAccessFile("src/d00_file/a.txt", "rw");
String line = null;
// 记录上次操作点
long lastpoint = 0;
while ((line = rf.readLine()) != null) {
long ponit = rf.getFilePointer();
// 如果包含替换字符串
if (line.contains(oldStr)) {
// 替换字符串
String str = line.replace(oldStr, newStr);
// 恢复到上次读取位置 （readLine前的位置）
rf.seek(lastpoint);
// 重写行数据
if (oldStr.length() != newStr.length()) {
byte[] newb = Arrays.copyOf(str.getBytes(), line.getBytes().length);
rf.write(newb);
// lastpoint = lastpoint + newb.length;
lastpoint = rf.getFilePointer();
continue;
} else {
rf.write(str.getBytes());
}
}
lastpoint = ponit;
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
rf.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}传输对象分为三种，一种是客户端发送给服务端的起始传输信息类、一种是分片文件数据类、一种是服务端返回给客户端分片文件指令类。
协议对象就是包含二个字段，一个是类型（上面三个对象的类型），一个是obejct字段，表示传输对象。
常量：
/**
* 文件传输常量
*/
public class Constants {
/**
* 文件传输状态的标示
*/
public static class FileStatus{
public static int BEGIN = 0; //开始
public static int CENTER = 1; //中间
public static int END = 2; //结尾
public static int COMPLETE = 3; //完成
}
/**
* 协议对象的传输对象类型
*/
public static class TransferType{
public static int REQUEST = 0; //文件信息类型
public static int INSTRUCT = 1; //文件指令类型
public static int DATA = 2; //文件分片类型
}
}
协议对象：
public class FileTransferProtocol {
private Integer transferType; // 类型
private Object transferObj; // 数据对象；(0)FileDescInfo、(1)FileBurstInstruct、(2)FileBurstData
public Integer getTransferType() {
return transferType;
}
public void setTransferType(Integer transferType) {
this.transferType = transferType;
}
public Object getTransferObj() {
return transferObj;
}
public void setTransferObj(Object transferObj) {
this.transferObj = transferObj;
}
}文件传输相关对象：
/**
* 文件描述信息
*/
public class FileDescInfo {
private String fileUrl; // 文件url
private String fileName; // 文件名称
private Long fileSize; // 文件size
public String getFileUrl() {
return fileUrl;
}
public void setFileUrl(String fileUrl) {
this.fileUrl = fileUrl;
}
public String getFileName() {
return fileName;
}
public void setFileName(String fileName) {
this.fileName = fileName;
}
public Long getFileSize() {
return fileSize;
}
public void setFileSize(Long fileSize) {
this.fileSize = fileSize;
}
}
/**
* 文件分片指令
*/
public class FileBurstInstruct {
private Integer status; // Constants.FileStatus ｛0开始、1中间、2结尾、3完成｝
private String clientFileUrl; // 客户端文件URL
private Integer readPosition; // 读取位置
public FileBurstInstruct() {
}
public FileBurstInstruct(Integer status) {
this.status = status;
}
public Integer getStatus() {
return status;
}
public void setStatus(Integer status) {
this.status = status;
}
public String getClientFileUrl() {
return clientFileUrl;
}
public void setClientFileUrl(String clientFileUrl) {
this.clientFileUrl = clientFileUrl;
}
public Integer getReadPosition() {
return readPosition;
}
public void setReadPosition(Integer readPosition) {
this.readPosition = readPosition;
}
}/**
* 文件分片数据
*/
public class FileBurstData {
private String fileUrl; //客户端文件地址
private String fileName; //文件名称
private Integer beginPos; //开始位置
private Integer endPos; //结束位置
private byte[] bytes; //文件字节；再实际应用中可以使用非对称加密，以保证传输信息安全
private Integer status; //Constants.FileStatus ｛0开始、1中间、2结尾、3完成｝
public FileBurstData(){
}
public String getFileUrl() {
return fileUrl;
}
public void setFileUrl(String fileUrl) {
this.fileUrl = fileUrl;
}
public FileBurstData(Integer status){
this.status = status;
}
public String getFileName() {
return fileName;
}
public void setFileName(String fileName) {
this.fileName = fileName;
}
public Integer getBeginPos() {
return beginPos;
}
public void setBeginPos(Integer beginPos) {
this.beginPos = beginPos;
}
public Integer getEndPos() {
return endPos;
}
public void setEndPos(Integer endPos) {
this.endPos = endPos;
}
public byte[] getBytes() {
return bytes;
}
public void setBytes(byte[] bytes) {
this.bytes = bytes;
}
public Integer getStatus() {
return status;
}
public void setStatus(Integer status) {
this.status = status;
}
}从协议对象的设计我们知道，本次文件上传将采用对象传输的方式，那么就要自定义编码解码器，自定义编码解码器采用的是以protostuff为基础实现，编码时，通过加一个int类型的长度字段，以及序列化的协议对象，解码时，通过解码长度字段后，读取数据，然后反序列化协议对象。
1) 序列化工具
/**
* protostuff序列化工具类
*/
public class SerializingUtil {
/**
* 将目标类序列化为byte数组
*/
public static <T> byte[] serialize(T source) {
Schema<T> schema = RuntimeSchema.getSchema((Class<T>) source.getClass());
LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
final byte[] result;
try {
result = ProtobufIOUtil.toByteArray(source, schema, buffer);
} finally {
buffer.clear();
}
return result;
}
/**
* 将byte数组序列化为目标类
*/
public static <T> T deserialize(byte[] source, Class<T> clazz) {
Schema<T> schema = RuntimeSchema.getSchema(clazz);
T t = schema.newMessage();
ProtobufIOUtil.mergeFrom(source, t, schema);
return t;
}
}2) 对象解码器 (输入向处理--放置在处理器链第一个--后续处理器可直接获得Clazz类型对象)
/**
* 对象解码器
*/
public class ObjectDecode extends ByteToMessageDecoder{
private Class<?> clazz ;
public ObjectDecode(Class<?> clazz) {
super();
this.clazz = clazz;
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
if (in.readableBytes() < 4) {
return;
}
// 标记包头位置
in.markReaderIndex();
// 读取长度
int len = in.readInt();
// 可读数少于长度
if(in.readableBytes() < len) {
// 重置到包头位置
in.resetReaderIndex();
return;
}
// 读取数据并序列化
byte[] bytes = new byte[len];
in.readBytes(bytes);
Object object = SerializingUtil.deserialize(bytes, clazz);
out.add(object);
}
}3) 对象编码器 (输出编码器---编码协议对象---放置在处理器链倒数第二个)
public class ObjectEncode extends MessageToByteEncoder<Object> {
private Class<?> clazz;
public ObjectEncode(Class<?> clazz) {
super();
this.clazz = clazz;
}
@Override
protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
if (clazz.isInstance(msg)) {
byte[] data = SerializingUtil.serialize(msg);
out.writeInt(data.length);
out.writeBytes(data);
}
}
}public class FileUtil {
// 默认一次只能读取10k数据
private static final int DEF_BUFF_SIZE = 1024*10;
private static int buff_size = DEF_BUFF_SIZE;
/**
* 客户端根据文件路径和position读取文件，返回文件分片数据FileBurstData对象。
*/
public static FileBurstData readFile(String fileUrl, Integer readPosition) throws IOException {
File file = new File(fileUrl);
RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r");
randomAccessFile.seek(readPosition);
byte[] bytes = new byte[buff_size];
int readSize = randomAccessFile.read(bytes);
if (readSize <= 0) {
randomAccessFile.close();
return new FileBurstData(Constants.FileStatus.COMPLETE);
}
FileBurstData fileBurstData = new FileBurstData();
fileBurstData.setFileUrl(fileUrl);
fileBurstData.setFileName(file.getName());
fileBurstData.setBeginPos(readPosition);
fileBurstData.setEndPos(readPosition + readSize);
// 不足buff尺寸需要拷贝去掉空字节
if (readSize < buff_size) {
byte[] copy = new byte[readSize];
System.arraycopy(bytes, 0, copy, 0, readSize);
fileBurstData.setBytes(copy);
fileBurstData.setStatus(Constants.FileStatus.END);
} else {
fileBurstData.setBytes(bytes);
fileBurstData.setStatus(Constants.FileStatus.CENTER);
}
randomAccessFile.close();
return fileBurstData;
}
/**
* 服务端根据url和客户端的文件分片数据对象，进行写文件，并且返回文件分片指令（由客户端使用）
*/
public static FileBurstInstruct writeFile(String baseUrl, FileBurstData fileBurstData) throws IOException {
if (Constants.FileStatus.COMPLETE == fileBurstData.getStatus()) {
return new FileBurstInstruct(Constants.FileStatus.COMPLETE); // Constants.FileStatus ｛0开始、1中间、2结尾、3完成｝
}
// 服务端写文件
File file = new File(baseUrl + "/" + fileBurstData.getFileName());
RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw");
randomAccessFile.seek(fileBurstData.getBeginPos());
randomAccessFile.write(fileBurstData.getBytes());
randomAccessFile.close();
if (Constants.FileStatus.END == fileBurstData.getStatus()) {
return new FileBurstInstruct(Constants.FileStatus.COMPLETE);
}
// 构建文件分片指令
FileBurstInstruct fileBurstInstruct = new FileBurstInstruct();
fileBurstInstruct.setStatus(Constants.FileStatus.CENTER); // 字段：读取状态
fileBurstInstruct.setClientFileUrl(fileBurstData.getFileUrl());
fileBurstInstruct.setReadPosition(fileBurstData.getEndPos() + 1); // 字段(核心)：下次读取位置
return fileBurstInstruct;
}另外补充文件协议对象的一个构建工具类
/**
* 消息构建对象
*/
public class MsgUtil {
/**
* 构建对象；文件传输info(客户端)
*/
public static FileTransferProtocol buildRequestTransferFile(String fileUrl, String fileName, Long fileSize) {
FileDescInfo fileDescInfo = new FileDescInfo();
fileDescInfo.setFileUrl(fileUrl);
fileDescInfo.setFileName(fileName);
fileDescInfo.setFileSize(fileSize);
FileTransferProtocol fileTransferProtocol = new FileTransferProtocol();
fileTransferProtocol.setTransferType(0);//0请求传输文件、1文件传输指令、2文件传输数据
fileTransferProtocol.setTransferObj(fileDescInfo);
return fileTransferProtocol;
}
/**
* 构建对象；文件传输指令(服务端)
* @param status 文件读取状态
* @param clientFileUrl 客户端文件地址
* @param readPosition 读取位置
*/
public static FileTransferProtocol buildTransferInstruct(Integer status, String clientFileUrl, Integer readPosition) {
FileBurstInstruct fileBurstInstruct = new FileBurstInstruct();
fileBurstInstruct.setStatus(status);
fileBurstInstruct.setClientFileUrl(clientFileUrl);
fileBurstInstruct.setReadPosition(readPosition);
FileTransferProtocol fileTransferProtocol = new FileTransferProtocol();
fileTransferProtocol.setTransferType(Constants.TransferType.INSTRUCT); // 0 (info) 1 (INSTRUCT) 2 (data)
fileTransferProtocol.setTransferObj(fileBurstInstruct);
return fileTransferProtocol;
}
/**
* 构建对象；文件传输指令(服务端)
*/
public static FileTransferProtocol buildTransferInstruct(FileBurstInstruct fileBurstInstruct) {
FileTransferProtocol fileTransferProtocol = new FileTransferProtocol();
fileTransferProtocol.setTransferType(Constants.TransferType.INSTRUCT); // 0 (info) 1 (INSTRUCT) 2 (data)
fileTransferProtocol.setTransferObj(fileBurstInstruct);
return fileTransferProtocol;
}
/**
* 构建对象；文件传输数据(客户端)
*/
public static FileTransferProtocol buildTransferData(FileBurstData fileBurstData) {
FileTransferProtocol fileTransferProtocol = new FileTransferProtocol();
fileTransferProtocol.setTransferType(Constants.TransferType.DATA); // 0 (info) 1 (INSTRUCT) 2 (data)
fileTransferProtocol.setTransferObj(fileBurstData);
return fileTransferProtocol;
}
}上面已经完成的代码包和类结构如下
public class FileTransferClient {
private EventLoopGroup workerGroup = new NioEventLoopGroup();
private Channel channel;
public ChannelFuture connect(String inetHost, int inetPort) {
ChannelFuture channelFuture = null;
try {
Bootstrap b = new Bootstrap();
b.group(workerGroup);
b.channel(NioSocketChannel.class);
b.option(ChannelOption.AUTO_READ, true);
b.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new ObjectDecode(FileTransferProtocol.class));
ch.pipeline().addLast(new ObjectEncode(FileTransferProtocol.class));
ch.pipeline().addLast(new FileTransferHandler());
}
});
channelFuture = b.connect(inetHost, inetPort).syncUninterruptibly();
this.channel = channelFuture.channel();
} catch (Exception e) {
e.printStackTrace();
} finally {
if (null != channelFuture && channelFuture.isSuccess()) {
System.out.println("client start done");
} else {
System.out.println("client start error");
}
}
return channelFuture;
}
public void destroy() {
if (null == channel)
return;
channel.close();
workerGroup.shutdownGracefully();
}
}我们将客户端的channel返回，通过channel进行初始发起请求给服务端，自定义处理器只负责接收数据，经过解码器的处理，我们知道接收的数据都是协议对象FileTransferProtocol，而这个对象有三个类型，我们客户端接收的都是服务端的指令对象，服务端缓存的也是指令对象，即客户端读取数据是经过初始请求后，由接收到服务端返回的指令对象后才会进行读数据，发送读数据等等循环操作。
客户端发送数据的演示：
public class ClientMain {
public static void main(String[] args) {
FileTransferClient client = new FileTransferClient();
ChannelFuture connect = client.connect("127.0.0.1", 7000);
File file = new File("C:\\test\\src\\测试传输文件.rar");
// 构建传输协议对象 （是包装info对象）
FileTransferProtocol fileTransferProtocol = MsgUtil.buildRequestTransferFile(file.getAbsolutePath(),
file.getName(), file.length());
connect.channel().writeAndFlush(fileTransferProtocol);
}
}
客户端自定义处理器：
/**
* 客户端自定义处理器
*/
public class FileTransferHandler extends ChannelInboundHandlerAdapter{
/**
* 主要逻辑：
* 判断指令是否完成，完成退出，没完成，读数据，构建传输对象，并且写传输对象到服务端。
**/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 客户端接收FileTransferProtocol对象
if (!(msg instanceof FileTransferProtocol)) return;
FileTransferProtocol fileTransferProtocol = (FileTransferProtocol) msg;
switch (fileTransferProtocol.getTransferType()) {
// 客户端只会处理类型为1，即传输对象是FileBurstInstruct(指令)
case 1:
FileBurstInstruct fileBurstInstruct = (FileBurstInstruct) fileTransferProtocol.getTransferObj();
// 服务端返回的instruct的状态已经完成,客户端退出操作
if (Constants.FileStatus.COMPLETE == fileBurstInstruct.getStatus()) {
ctx.flush();
ctx.close();
System.exit(-1);
return;
}
// 客户端读文件数据返回fileBurstData
FileBurstData fileBurstData = FileUtil.readFile(fileBurstInstruct.getClientFileUrl(), fileBurstInstruct.getReadPosition());
System.out.println("客户端读取一次文件，结尾是：" + fileBurstData.getEndPos());
// 构建协议对象传输
ctx.writeAndFlush(MsgUtil.buildTransferData(fileBurstData));
break;
default:
break;
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
System.out.println("异常信息：\r\n" + cause.getMessage());
}
}/**
* 服务端
*/
public class FileTransferServer {
/**
* Main函数
*/
public static void main(String[] args) {
String path = "";
FileTransferServer server = new FileTransferServer(path);
server.bind(7000);
}
private EventLoopGroup parentGroup = new NioEventLoopGroup(1);
private EventLoopGroup childGroup = new NioEventLoopGroup();
private Channel channel;
// 上传文件的服务端存储目标路径
private String dest_path;
public FileTransferServer(String dest_path) {
super();
this.dest_path = dest_path;
}
public ChannelFuture bind(int port) {
ChannelFuture channelFuture = null;
try {
ServerBootstrap b = new ServerBootstrap();
b.group(parentGroup, childGroup).channel(NioServerSocketChannel.class) // 非阻塞模式
.option(ChannelOption.SO_BACKLOG, 128).childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new ObjectDecode(FileTransferProtocol.class));
ch.pipeline().addLast(new ObjectEncode(FileTransferProtocol.class));
ch.pipeline().addLast(new FileTransferServerHandler(dest_path));
}
});
channelFuture = b.bind(port).sync();
this.channel = channelFuture.channel();
} catch (Exception e) {
e.printStackTrace();
} finally {
if (null != channelFuture && channelFuture.isSuccess()) {
System.out.println("server start done");
} else {
System.out.println("server start error");
}
}
return channelFuture;
}
public void close() {
if (channel != null) {
channel.close();
}
parentGroup.shutdownGracefully();
childGroup.shutdownGracefully();
}
public Channel getChannel() {
return channel;
}
public void setChannel(Channel channel) {
this.channel = channel;
}
}我们从客户端的发送请求的方法以及read方法可以知晓，服务端的处理器需要接收二种类型的传输对象，一个是info对象，一个是data对象，info对象是请求打开时处理用途，data对象是在已经传输过程中接收data对象，并且将data数据写入服务端。
在上面这个过程中，我们不管是接收info对象还是data对象，发送给客户端的都是instruct指令对象。
如果接收的是info对象，就要判断是否有instruct缓存？如果有，就是断点续传的，那么直接拿到此缓存对象发送给客户端，并且叫客户端读文件传数据。如果没有，那么就要初始构建instruct指令对象，即指定position位置为0起始。
如果接收的是data对象，就要将data对象的数据写入服务端存储地址，并且构建insturct指令存入缓存（指定position为上次读文件的结尾+1），然后发送此insturct对象给客户端，客户端根据是否完成来决定是否继续读取文件。
服务端自定义处理器实现：
public class FileTransferServerHandler extends ChannelInboundHandlerAdapter{
private String dest_path;
public FileTransferServerHandler(String dest_path) {
this.dest_path = dest_path;
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("客户端链接" + ctx.channel().localAddress().toString());
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
System.out.println("客户端断开链接" + ctx.channel().localAddress().toString());
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (!(msg instanceof FileTransferProtocol))return;
FileTransferProtocol fileTransferProtocol = (FileTransferProtocol) msg;
switch (fileTransferProtocol.getTransferType()) {
case 0:
FileDescInfo info = (FileDescInfo) fileTransferProtocol.getTransferObj();
// 有缓存说明是断点续传，instruct记录的是下次需要读的postion
FileBurstInstruct old = CacheUtil.get(info.getFileName());
if(old!=null) {
if (old.getStatus() == Constants.FileStatus.COMPLETE) {
CacheUtil.remove(info.getFileName());
}
ctx.writeAndFlush(MsgUtil.buildTransferInstruct(old));
return ;
}
// 没缓存就是初始传输，主要是指定instruct的position为0
FileTransferProtocol sendFileTransferProtocol = MsgUtil.buildTransferInstruct(Constants.FileStatus.BEGIN, info.getFileUrl(), 0);
ctx.writeAndFlush(sendFileTransferProtocol);
break;
case 2:
FileBurstData fileBurstData = (FileBurstData) fileTransferProtocol.getTransferObj();
FileBurstInstruct fileBurstInstruct = FileUtil.writeFile(dest_path, fileBurstData);
// 保存断点续传信息
CacheUtil.put(fileBurstData.getFileName(), fileBurstInstruct);
ctx.writeAndFlush(MsgUtil.buildTransferInstruct(fileBurstInstruct));
// 传输完成删除断点信息
if (fileBurstInstruct.getStatus() == Constants.FileStatus.COMPLETE) {
CacheUtil.remove(fileBurstData.getFileName());
}
break;
default:
break;
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("异常信息：\r\n" + cause.getMessage());
}
}
end !!!