Netty之文件传输

╰+哭是因爲堅強的太久メ 2022-07-12 23:46 280阅读 0赞

1、工程的目录结构

20161215223056082

2、GZIP进行压缩和解压的工具类

  1. import java.io.ByteArrayInputStream;
  2. import java.io.ByteArrayOutputStream;
  3. import java.io.File;
  4. import java.io.FileInputStream;
  5. import java.io.FileOutputStream;
  6. import java.util.zip.GZIPInputStream;
  7. import java.util.zip.GZIPOutputStream;
  9. public class GzipUtils {
  11.     public static byte[] gzip(byte[] data) throws Exception {
  12.         ByteArrayOutputStream bos = new ByteArrayOutputStream();
  13.         GZIPOutputStream gzip = new GZIPOutputStream(bos);
  14.         gzip.write(data);
  15.         gzip.finish();
  16.         gzip.close();
  17.         byte[] ret = bos.toByteArray();
  18.         bos.close();
  19.         return ret;
  20.     }
  22.     public static byte[] ungzip(byte[] data) throws Exception {
  23.         ByteArrayInputStream bis = new ByteArrayInputStream(data);
  24.         GZIPInputStream gzip = new GZIPInputStream(bis);
  25.         byte[] buf = new byte[1024];
  26.         int num = -1;
  27.         ByteArrayOutputStream bos = new ByteArrayOutputStream();
  28.         while ((num = gzip.read(buf, 0, buf.length)) != -1) {
  29.             bos.write(buf, 0, num);
  30.         }
  31.         gzip.close();
  32.         bis.close();
  33.         byte[] ret = bos.toByteArray();
  34.         bos.flush();
  35.         bos.close();
  36.         return ret;
  37.     }
  39.     public static void main(String[] args) throws Exception {
  41.         // 读取文件
  42.         String readPath = System.getProperty("user.dir") + File.separatorChar
  43.                 + "sources" + File.separatorChar + "005.jpg";
  44.         File file = new File(readPath);
  45.         FileInputStream in = new FileInputStream(file);
  46.         byte[] data = new byte[in.available()];
  47.         in.read(data);
  48.         in.close();
  50.         System.out.println("文件原始大小:" + data.length);
  52.         // 测试压缩
  53.         byte[] ret1 = GzipUtils.gzip(data);
  54.         System.out.println("压缩之后大小:" + ret1.length);
  56.         // 还原文件
  57.         byte[] ret2 = GzipUtils.ungzip(ret1);
  58.         System.out.println("还原之后大小:" + ret2.length);
  60.         // 写出文件
  61.         String writePath = System.getProperty("user.dir") + File.separatorChar
  62.                 + "receive" + File.separatorChar + "005.jpg";
  63.         FileOutputStream fos = new FileOutputStream(writePath);
  64.         fos.write(ret2);
  65.         fos.close();
  67.     }
  69. }

3、在客户端的请求Request中，加入要传输的文件(private byte[] attachment)

  1. import java.io.Serializable;
  2. public class Request implements Serializable {
  3.     /**
  4.      * 
  5.      */
  6.     private static final long serialVersionUID = -2813211330451521507L;
  7.     private String id;
  8.     private String name;
  9.     private String requestMessage;
  10.     private byte[] attachment;
  12.     public String getId() {
  13.         return id;
  14.     }
  16.     public void setId(String id) {
  17.         this.id = id;
  18.     }
  20.     public String getName() {
  21.         return name;
  22.     }
  24.     public void setName(String name) {
  25.         this.name = name;
  26.     }
  28.     public String getRequestMessage() {
  29.         return requestMessage;
  30.     }
  32.     public void setRequestMessage(String requestMessage) {
  33.         this.requestMessage = requestMessage;
  34.     }
  36.     public byte[] getAttachment() {
  37.         return attachment;
  38.     }
  40.     public void setAttachment(byte[] attachment) {
  41.         this.attachment = attachment;
  42.     }
  44. }

4、服务端的Response

  1. import java.io.Serializable;
  2. public class Response implements Serializable {
  4.     /**
  5.      * 
  6.      */
  7.     private static final long serialVersionUID = -5640678664176009458L;
  8.     private String id;
  9.     private String name;
  10.     private String responseMessage;
  12.     public String getId() {
  13.         return id;
  14.     }
  16.     public void setId(String id) {
  17.         this.id = id;
  18.     }
  20.     public String getName() {
  21.         return name;
  22.     }
  24.     public void setName(String name) {
  25.         this.name = name;
  26.     }
  28.     public String getResponseMessage() {
  29.         return responseMessage;
  30.     }
  32.     public void setResponseMessage(String responseMessage) {
  33.         this.responseMessage = responseMessage;
  34.     }
  36.     @Override
  37.     public String toString() {
  38.         return "Response [id=" + id + ", name=" + name + ", responseMessage="
  39.                 + responseMessage + "]";
  40.     }
  42. }

5、Jboss序列化工具

  1. import io.netty.handler.codec.marshalling.DefaultMarshallerProvider;
  2. import io.netty.handler.codec.marshalling.DefaultUnmarshallerProvider;
  3. import io.netty.handler.codec.marshalling.MarshallerProvider;
  4. import io.netty.handler.codec.marshalling.MarshallingDecoder;
  5. import io.netty.handler.codec.marshalling.MarshallingEncoder;
  6. import io.netty.handler.codec.marshalling.UnmarshallerProvider;
  8. import org.jboss.marshalling.MarshallerFactory;
  9. import org.jboss.marshalling.Marshalling;
  10. import org.jboss.marshalling.MarshallingConfiguration;
  12. /**
  13.  * Marshalling工厂
  14.  * 
  15.  */
  16. public final class MarshallingCodeCFactory {
  18.     /**
  19.      * 创建Jboss Marshalling解码器MarshallingDecoder
  20.      * 
  21.      * @return MarshallingDecoder
  22.      */
  23.     public static MarshallingDecoder buildMarshallingDecoder() {
  24.         // 首先通过Marshalling工具类的精通方法获取Marshalling实例对象 参数serial标识创建的是java序列化工厂对象。
  25.         final MarshallerFactory marshallerFactory = Marshalling
  26.                 .getProvidedMarshallerFactory("serial");
  27.         // 创建了MarshallingConfiguration对象，配置了版本号为5
  28.         final MarshallingConfiguration configuration = new MarshallingConfiguration();
  29.         configuration.setVersion(5);
  30.         // 根据marshallerFactory和configuration创建provider
  31.         UnmarshallerProvider provider = new DefaultUnmarshallerProvider(
  32.                 marshallerFactory, configuration);
  33.         // 构建Netty的MarshallingDecoder对象，俩个参数分别为provider和单个消息序列化后的最大长度
  34.         MarshallingDecoder decoder = new MarshallingDecoder(provider,
  35.                 1024 * 1024 * 1);
  36.         return decoder;
  37.     }
  39.     /**
  40.      * 创建Jboss Marshalling编码器MarshallingEncoder
  41.      * 
  42.      * @return MarshallingEncoder
  43.      */
  44.     public static MarshallingEncoder buildMarshallingEncoder() {
  45.         final MarshallerFactory marshallerFactory = Marshalling
  46.                 .getProvidedMarshallerFactory("serial");
  47.         final MarshallingConfiguration configuration = new MarshallingConfiguration();
  48.         configuration.setVersion(5);
  49.         MarshallerProvider provider = new DefaultMarshallerProvider(
  50.                 marshallerFactory, configuration);
  51.         // 构建Netty的MarshallingEncoder对象，MarshallingEncoder用于实现序列化接口的POJO对象序列化为二进制数组
  52.         MarshallingEncoder encoder = new MarshallingEncoder(provider);
  53.         return encoder;
  54.     }
  55. }

6、服务端的实现

  1. import io.netty.bootstrap.ServerBootstrap;
  2. import io.netty.channel.ChannelFuture;
  3. import io.netty.channel.ChannelInitializer;
  4. import io.netty.channel.ChannelOption;
  5. import io.netty.channel.EventLoopGroup;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioServerSocketChannel;
  9. import io.netty.handler.logging.LogLevel;
  10. import io.netty.handler.logging.LoggingHandler;
  12. public class Server {
  14.     public Server() {
  15.     }
  17.     public void bind(int port) throws Exception {
  18.         // 配置NIO线程组
  19.         EventLoopGroup bossGroup = new NioEventLoopGroup();
  20.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  21.         try {
  22.             // 服务器辅助启动类配置
  23.             ServerBootstrap b = new ServerBootstrap();
  24.             b.group(bossGroup, workerGroup)
  25.                     .channel(NioServerSocketChannel.class)
  26.                     .handler(new LoggingHandler(LogLevel.INFO))
  27.                     .childHandler(new ChildChannelHandler())//
  28.                     .option(ChannelOption.SO_BACKLOG, 1024) // 设置tcp缓冲区 // (5)
  29.                     .childOption(ChannelOption.SO_KEEPALIVE, true); // (6)
  30.             // 绑定端口 同步等待绑定成功
  31.             ChannelFuture f = b.bind(port).sync(); // (7)
  32.             // 等到服务端监听端口关闭
  33.             f.channel().closeFuture().sync();
  34.         } finally {
  35.             // 优雅释放线程资源
  36.             workerGroup.shutdownGracefully();
  37.             bossGroup.shutdownGracefully();
  38.         }
  39.     }
  41.     /**
  42.      * 网络事件处理器
  43.      */
  44.     private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
  45.         @Override
  46.         protected void initChannel(SocketChannel ch) throws Exception {
  47.             // 添加Jboss的序列化，编解码工具
  48.             ch.pipeline().addLast(
  49.                     MarshallingCodeCFactory.buildMarshallingEncoder());
  50.             ch.pipeline().addLast(
  51.                     MarshallingCodeCFactory.buildMarshallingDecoder());
  52.             // 处理网络IO
  53.             ch.pipeline().addLast(new ServerHandler());
  54.         }
  55.     }
  57.     public static void main(String[] args) throws Exception {
  58.         new Server().bind(9999);
  59.     }
  60. }

7、服务端的Handler的实现

  1. import java.io.File;
  2. import java.io.FileOutputStream;
  4. import com.netty.utils.GzipUtils;
  6. import io.netty.channel.ChannelHandlerAdapter;
  7. import io.netty.channel.ChannelHandlerContext;
  9. public class ServerHandler extends ChannelHandlerAdapter {
  10.     // 用于获取客户端发送的信息
  11.     @Override
  12.     public void channelRead(ChannelHandlerContext ctx, Object msg)
  13.             throws Exception {
  15.         Request req = (Request) msg;
  16.         System.out.println("Server : " + req.getId() + ", " + req.getName()
  17.                 + ", " + req.getRequestMessage());
  18.         // 进行图片资源的还原
  19.         byte[] attachment = GzipUtils.ungzip(req.getAttachment());
  21.         // 获取图片的保存目录
  22.         String path = System.getProperty("user.dir") + File.separatorChar
  23.                 + "receive" + File.separatorChar + req.getName();
  24.         // 进行图片的保存
  25.         FileOutputStream fos = new FileOutputStream(path);
  26.         fos.write(attachment);
  27.         fos.close();
  29.         // 给客户端，响应数据
  30.         Response resp = new Response();
  31.         resp.setId(req.getId());
  32.         resp.setName("resp" + req.getId());
  33.         resp.setResponseMessage("响应内容" + req.getId());
  34.         ctx.writeAndFlush(resp);// .addListener(ChannelFutureListener.CLOSE);
  36.     }
  38.     @Override
  39.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
  40.             throws Exception {
  41.         // cause.printStackTrace();
  42.         ctx.close();
  43.     }
  44. }

8、客户端的实现

  1. import io.netty.bootstrap.Bootstrap;
  2. import io.netty.channel.ChannelFuture;
  3. import io.netty.channel.ChannelInitializer;
  4. import io.netty.channel.ChannelOption;
  5. import io.netty.channel.EventLoopGroup;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioSocketChannel;
  10. public class Client {
  12.     /**
  13.      * 连接服务器
  14.      * 
  15.      * @param port
  16.      * @param host
  17.      * @throws Exception
  18.      */
  19.     public void connect(int port, String host) throws Exception {
  20.         // 配置客户端NIO线程组
  21.         EventLoopGroup group = new NioEventLoopGroup();
  22.         try {
  23.             // 客户端辅助启动类 对客户端配置
  24.             Bootstrap b = new Bootstrap();
  25.             b.group(group)//
  26.                     .channel(NioSocketChannel.class)//
  27.                     .option(ChannelOption.TCP_NODELAY, true)//
  28.                     .handler(new MyChannelHandler());//
  29.             // 异步链接服务器 同步等待链接成功
  30.             ChannelFuture f = b.connect(host, port).sync();
  32.             // 等待链接关闭
  33.             f.channel().closeFuture().sync();
  35.         } finally {
  36.             group.shutdownGracefully();
  37.             System.out.println("客户端优雅的释放了线程资源...");
  38.         }
  40.     }
  42.     /**
  43.      * 网络事件处理器
  44.      */
  45.     private class MyChannelHandler extends ChannelInitializer<SocketChannel> {
  46.         @Override
  47.         protected void initChannel(SocketChannel ch) throws Exception {
  48.             // 添加Jboss的序列化，编解码工具
  49.             ch.pipeline().addLast(
  50.                     MarshallingCodeCFactory.buildMarshallingEncoder());
  51.             ch.pipeline().addLast(
  52.                     MarshallingCodeCFactory.buildMarshallingDecoder());
  53.             // 处理网络IO
  54.             ch.pipeline().addLast(new ClientHandler());// 处理网络IO
  55.         }
  57.     }
  59.     public static void main(String[] args) throws Exception {
  60.         new Client().connect(9999, "127.0.0.1");
  62.     }
  64. }

9、客户端的Handler的实现

  1. import java.io.File;
  2. import java.io.FileInputStream;
  4. import com.netty.utils.GzipUtils;
  6. import io.netty.channel.ChannelHandlerAdapter;
  7. import io.netty.channel.ChannelHandlerContext;
  8. import io.netty.util.ReferenceCountUtil;
  10. //用于读取客户端发来的信息
  11. public class ClientHandler extends ChannelHandlerAdapter {
  13.     // 客户端与服务端，连接成功的售后
  14.     @Override
  15.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  16.         // 进行图片的传输
  17.         for (int i = 0; i < 5; i++) {
  18.             String fileName = "00" + (i + 1) + ".jpg";
  19.             // 进行文件的读取
  20.             String path = System.getProperty("user.dir") + File.separatorChar
  21.                     + "sources" + File.separatorChar + fileName;
  22.             File file = new File(path);
  24.             Request req = new Request();
  25.             req.setId("" + i);
  26.             req.setName(file.getName());
  27.             req.setRequestMessage("数据信息" + i);
  29.             // 进行图片的读取
  30.             FileInputStream in = new FileInputStream(file);
  31.             byte[] data = new byte[in.available()];
  32.             in.read(data);
  33.             in.close();
  34.             // 进行数据的压缩
  35.             req.setAttachment(GzipUtils.gzip(data));
  36.             // 向服务端，传送图片信息
  37.             ctx.channel().writeAndFlush(req);
  38.         }
  40.     }
  42.     // 只是读数据，没有写数据的话
  43.     // 需要自己手动的释放的消息
  44.     @Override
  45.     public void channelRead(ChannelHandlerContext ctx, Object msg)
  46.             throws Exception {
  47.         try {
  48.             Response response = (Response) msg;
  49.             System.out.println(response);
  51.         } finally {
  52.             ReferenceCountUtil.release(msg);
  53.         }
  54.     }
  56.     @Override
  57.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
  58.             throws Exception {
  59.         ctx.close();
  60.     }
  62. }

发表评论

表情：
评论列表 （有 0 条评论，280人围观）

还没有评论，来说两句吧...

相关阅读

    相关 netty传输java bean对象

    在上一篇博客(netty入门实现简单的echo程序)中，我们知道了如何使用netty发送一个简单的消息，但是这远远是不够的。在这篇博客中，我们来使用netty发送一个java

    末蓝、末蓝、/ 0 赞/ 70 阅读

    相关 Netty权威指南之文件传输

    本章相关知识点： 文件是最常见的数据源之一，在程序经常需要将数据存储到文件中，比如：图片文件、声音文件等数据文件。在实际使用中，文件都包含一个特定的格式，这个格式需要程序员根

    Dear 丶Dear 丶/ 0 赞/ 65 阅读

    相关 netty去实现文件传输

    在写出了Netty Hello World 和 netty对象传输之后，又觉得不够，看了官网的例子，所以有了现在的这个文件传输。 顺便说下，netty官网的例子真的好，如果

    桃扇骨桃扇骨/ 0 赞/ 299 阅读