java nio socket结合netty传输文件

43 阅读 0 评论 29 点赞

我是靠谱客的博主秀丽高山，最近开发中收集的这篇文章主要介绍java nio socket结合netty传输文件，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

最近才开始研究java nio以及netty方面的知识，作为一个IT菜鸟，对这些东西几乎一无所知，所以有些地方的见解也纯属个人愚见，出现差错也在所难免

基于我的理解，java nio 之间数据的传递主要建立在channel和ByteBuffer之上，将要传递的数据转化成ByteBuffer写入到对应的channel中去，就实现了数据的写入

同理，使用netty实际上是基于java nio的一个框架，只不过对其进行了更好的封装，它通过管道中的Handler接收数据，然后将数据对应转化为对应的Bytebuf，实现数据的读取

以下就是一个文件传输例子的实现

其中客户端参考了网络上的数据重传以及给文件加上首部的一些方法(java nio 文件传输)

客户端代码：NioSocketClient.java

package com.test.nio.netty;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.SocketChannel;
import java.util.logging.Logger;
public class NioSocketClient {
private final static Logger log = Logger.getLogger(NioSocketClient.class.getName());
private final static String START = "START";
private InetSocketAddress inetSocketAddress;
/* 发送数据缓冲区 */
private static ByteBuffer sendBuffer = ByteBuffer.allocate(1024);
public CopyOfNioSocketClient(int port){
try{
inetSocketAddress = new InetSocketAddress("localhost", port);
init();
} catch(Exception e){
e.printStackTrace();
}
}
private void init(){
try {
SocketChannel socketChannel = SocketChannel.open(inetSocketAddress);
socketChannel.configureBlocking(false);
sendFile(socketChannel);
} catch (Exception e) {
e.printStackTrace();
}
}
private void sendFile(SocketChannel client) {
FileInputStream fis = null;
FileChannel channel = null;
try {
File file = new File("F:\test\1.avi");
fis = new FileInputStream(file);
channel = fis.getChannel();
int i = 1;
int sum = 0;
int len = 0;
//写文件头部
String file_name = file.getName();
Long file_length = file.length();
writeFileHead(client, START);
writeFileHead(client, file_name);
writeFileHead(client, String.valueOf(file_length));
//写文件内容
while((len = channel.read(sendBuffer)) != -1) {
sendBuffer.flip();
int send = client.write(sendBuffer);
log.info("已发送文件总字节数" + (sum += len));
log.info("i发送-->" + (i++) + " len:" + len + " send:" + send);
// 考虑到服务器缓冲区满的情况
while(send == 0){
Thread.sleep(10);
send = client.write(sendBuffer);
log.info("i重新-->" + i + " len:" + len + " send:" + send);
}
sendBuffer.clear();
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
channel.close();
fis.close();
client.close();
} catch (Exception e) {
e.printStackTrace();
}
}
}
/**
* @category 写文件头部
* @param channel
* @param data
* @throws IOException
*/
public void writeFileHead(SocketChannel channel, String data) throws IOException {
byte[] d_content = data.getBytes();
byte[] d_len = i2b(d_content.length);
sendBuffer.put(d_len);
sendBuffer.put(d_content);
}
private byte[] i2b(int i) {
//4个字节
return new byte[] {
(byte) ((i >> 24) & 0xFF),
(byte) ((i >> 16) & 0xFF),
(byte) ((i >> 8) & 0xFF),
(byte) (i & 0xFF),
};
}
public static void main(String[] args){
new NioSocketClient(12345);
}
}

服务器端:NettyServer.java

package com.test.nio.netty;
import java.net.InetSocketAddress;
import java.util.logging.Logger;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class NettyServer {
private static final Logger log = Logger.getLogger(NettyServer.class.getName());
private InetSocketAddress inetSocketAddress;
public NettyServer(int port) {
this.inetSocketAddress = new InetSocketAddress(port);
}
public void start() {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup);
b.channel(NioServerSocketChannel.class);
b.option(ChannelOption.SO_KEEPALIVE, true);
b.childHandler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) throws Exception {
//ch.pipeline().addLast(new StringEncoder());
ch.pipeline().addLast(new FileReceiveHandler());
}
});
log.info("server listen on " + inetSocketAddress.getPort() + ".");
ChannelFuture f = b.bind(inetSocketAddress).sync();
f.channel().closeFuture().sync();
} catch (Exception e) {
// TODO: handle exception
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) {
NettyServer ns = new NettyServer(12345);
ns.start();
}
}

Handler代码:FileReceivedHandler.java

package com.test.nio.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.io.FileOutputStream;
import java.util.logging.Logger;
public class FileReceiveHandler extends ChannelInboundHandlerAdapter {
private final static Logger log = Logger.getLogger(FileReceiveHandler.class.getName());
private static String R_FLAG;	//是否第一次接收数据
private static String FILE_NAME = "DEFAULT";	//默认文件名
private static String STATE = "START";	//文件传输标志
private static Long FILE_LENGTH = 0L;	//文件总长度
private static Long READ_LENGTH = 0L;	//文件已读长度
private FileOutputStream fos;
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
writeMessageToClient(ctx, "connect success");
log.info("--Client in--");
R_FLAG = "FRIST";
READ_LENGTH = 0L;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
ByteBuf buf = (ByteBuf) msg;
initialFileReceivedState(buf);
if(isStart()) {
saveFile(ctx, buf);
} else {
writeMessageToClient(ctx, "file hasn't been transported yet");
}
releaseBuf(buf);
closeContext(ctx);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
ctx.close();
}
/**
* @category 向客户端写回文件传输进度
* @param ctx
* @param transportLength
*/
private void writeTransportProcess(ChannelHandlerContext ctx, Long transportLength) {
long t_length = transportLength.longValue();
long f_length = FILE_LENGTH.longValue();
String process = "file transport percent ";
String percent = String.format("%.1f", (t_length / (f_length * 1.0)) * 100) + "%";
writeMessageToClient(ctx, process + percent);
}
/**
* @category 向客户端写回信息
* @param ctx
* @param msg
*/
private void writeMessageToClient(ChannelHandlerContext ctx, String msg) {
ctx.write(msg);
}
/**
* @category 释放ByteBuf空间
* @param buf
*/
private void releaseBuf(ByteBuf buf) {
buf.clear();
buf.release();
}
/**
* @category 判断文件传输是否开始
* @return
*/
private boolean isStart() {
boolean flag = false;
if("START".equalsIgnoreCase(STATE)) {
flag = true;
}
return flag;
}
/**
* @category 关闭Context
* @return
*/
private void closeContext(ChannelHandlerContext ctx) {
if(FILE_LENGTH.longValue() == READ_LENGTH.longValue()) {
ctx.close();
}
}
/**
* @category 文件存储
* @param buf
* @throws Exception
*/
private void saveFile(ChannelHandlerContext ctx, ByteBuf buf) throws Exception {
if (buf.isReadable()) {
READ_LENGTH += buf.readableBytes();
//向客户端写回文件传输进度
writeTransportProcess(ctx, READ_LENGTH);
byte[] bytes = new byte[buf.readableBytes()];
buf.readBytes(bytes);
fos.write(bytes);
fos.flush();
}
}
/**
* @category 将字节转化为整数
* @param b
* @return
*/
private int b2i(byte[] b) {
int value = 0;
for (int i = 0; i < 4; i++) {
int shift = (4 - 1 - i) * 8;
value += (b[i] & 0x000000FF) << shift;
}
return value;
}
/**
* @category 读取头部信息
* @param in
* @return
*/
private String readFileHead(ByteBuf in) {
byte[] stateBytes = new byte[4];
in.readBytes(stateBytes);
int stateLength = b2i(stateBytes);
byte[] sendState = new byte[stateLength];
in.readBytes(sendState);
return new String(sendState);
}
/**
* @category 初始化从客户端发过来的文件头部信息
* @param in
* @throws Exception
*/
private void initialFileReceivedState(ByteBuf in) throws Exception {
if("FRIST".equalsIgnoreCase(R_FLAG)) {
STATE = readFileHead(in);
FILE_NAME = readFileHead(in);
FILE_LENGTH = Long.valueOf(readFileHead(in));
fos = new FileOutputStream("F:\" + FILE_NAME);
log.info("server: state->" + STATE);
log.info("server: fileName->" + FILE_NAME);
log.info("server: fileLength->" + FILE_LENGTH);
R_FLAG = "OTHER";
}
}
}

测试过程中，可能会出现"您的软件中途释放了一个连接"等类似错误，是因为在写入头部数据时出现差错，目前仍然在解决中...