[java,file]文件收发流程简述和代码实现

1830阅读　3评论2015-07-01　夏目玲子
分类：Java

在这篇博文中，简单的梳理一下 Server 端分多次将一个大文件发送给 Client 端的情形。
消息类型，暂时定为
{
  1. Server_Ready_Send_File
  2. Server_Sending_File
  3. Server_End_Send_File
  4. Client_Ready_Recv_File
}

1. Server 启动

2. Client 请求连接

3. Server - Client 连接成功创建

4. Server 在创建连接的句柄方法中，创建消息{类型: Server_Ready_Send_File , 数据:<long:文件大小(字节) ， byte[] 文件名称>}
  
5. Client 在接收到该消息之后，便在指定的路径下面创建文件，且，打开文件流，准备写操作；准备工作做好了之后，便向 Server  发送消息 {类型:Client_Ready_Send_File, 数据:<0>}

6. Server 在接收到 Client 的消息之后，便将文件打开，并且执行循环，读取文件中的内容，将文件中的数据封装成消息  {类型:Server_Sending_File,数据<文件中的数据转换为二进制类型>}
    循环按行读取文件，当得到结束标语'end of the file!' 的时候，便，将消息的类型转换为 Server_End_Send_File 。 然后，将该文件内容写入之后，发送消息。随后关闭掉与 client 的连接。

7. Client 在接收 Server 的 Server_Sending_File 之后，将其写入到打开的文件中，随后当接收到来自 Server_End_Send_File 类型的消息之后，便可以将该消息写入到文件中，
   然后，该 flush() ，flush,该 close(), close . 关闭文件输入流之后，断开与 Server 之间的连接。

下面的代码经过测试，已经可以成功的分块传递大小为 17699324B 的文件，不过代码中仍旧有很多不太恰当，没有细化的地方。
{比如说传递的文件名称的统一，以及每次传递的文件字节数目在小于 (short = 2 byte = 16 bit 可以用来表示 2^15 B 大小的文件 - 3 byte )}
可以最大化传输的文件行数目。但是，这个项目代码暂时可以作为模板使用。

项目的结构与前一篇草稿篇介绍的一样，只不过，在这里我们并不适用 ClientHandlerListener , 而是直接将互相发送的代码写到各自的 Handler 中。

/src/
  |----------common/
                    |-------- MessageType
                    |-------- Message
                    |-------- MessageDecoder
                    |-------- MessageEncoder
  |---------- client/
                   |-------- Client 
                   |-------- ClientHandler
                   |-------- ClientHandlerListener
  |---------- server/
                   |-------- Server
                   |-------- ServerHandler


运行代码之前，建议在 /tmp/ 路径下面创建两个名称为 test1.txt ; test2.txt 的文件，
可以根据需要测试的数据量的大小来设置文件的大小，建议将两个文件设置一个超大，一个正常大小。 大文件用来测试程序运行过程中的稳定性，以及大数据量的发送情况。

common/MessageType.java

点击(此处)折叠或打开

  1. package org.kylin.zhang.common;



  5. /**
  6.  * Created by root on 6/30/15.
  7.  */
  8. public enum MessageType
  9. {
  10.     READY_SEND_FILE((byte)0x01) ,
  11.     READY_RECV_FILE((byte)0x04) ,
  12.     FILE_SENDING((byte)0x02) ,
  13.     FILE_END((byte)0x03) ,
  14.     SHUT_DOWN((byte)0x00),
  15.     UNKNOWN ((byte)0x00) ;


  18.     private final byte b ;

  20.     private MessageType ( byte b )
  21.     {
  22.         this.b = b ;
  23.     }


  26.     public static MessageType fromByte ( byte b )
  27.     {
  28.         for (MessageType code : values())
  29.         {
  30.             if (code.b == b )
  31.                 return code ;
  32.         }

  34.         return UNKNOWN ;
  35.     }

  37.     public byte getByte ()
  38.     {
  39.         return this.b ;
  40.     }
  41. }

common/Message.java

点击(此处)折叠或打开

  1. package org.kylin.zhang.common;

  3. import java.io.Serializable ;

  5. /**
  6.  * Created by root on 6/30/15.
  7.  */

  9. public class Message implements Serializable
  10. {
  11.     private MessageType type ;
  12.     private short length ;
  13.     private byte [] data ;

  15.     public Message ()
  16.     {}

  18.     public Message (MessageType type , short len , byte [] data )
  19.     {
  20.         this.type = type ;
  21.         this.length = len ;
  22.         this.data = data ;
  23.     }

  25.     public short getLength ()
  26.     {
  27.         return this.length ;
  28.     }
  29.     public void setLength ( short len )
  30.     {
  31.         this.length = len ;
  32.     }


  35.     public MessageType getType ()
  36.     {
  37.         return this.type ;
  38.     }

  40.     public void setType ( MessageType type )
  41.     {
  42.         this.type = type ;
  43.     }

  45.     public byte [] getData ()
  46.     {
  47.         return this.data ;
  48.     }

  50.     public void setData ( byte [] data )
  51.     {
  52.         this.data = data ;
  53.     }


  56.     @Override
  57.     public String toString ()
  58.     {
  59.         return "\nmessage type :"+this.type +"\n"
  60.                 +"message length :"+this.length +"\n"
  61.                 +"message content :"+ new String (this.data) +"\n" ;
  62.     }




  67. }

common/MessageDecoder.java / not changed
common/MessageEncoder.java / not changed


client/Client.java

点击(此处)折叠或打开

  1. package org.kylin.zhang.client;

  3. import org.kylin.zhang.common.* ;
  4. import org.jboss.netty.bootstrap.ClientBootstrap ;
  5. import org.jboss.netty.channel.ChannelFactory ;
  6. import org.jboss.netty.channel.ChannelPipeline ;
  7. import org.jboss.netty.channel.ChannelPipelineFactory ;
  8. import org.jboss.netty.channel.Channels ;
  9. import org.jboss.netty.channel.group.ChannelGroup ;
  10. import org.jboss.netty.channel.group.DefaultChannelGroup ;
  11. import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory ;

  13. import java.io.* ;
  14. import java.net.InetSocketAddress ;
  15. import java.util.concurrent.Executors ;
  16. import java.util.concurrent.atomic.AtomicInteger ;

  18. /**
  19.  * Created by root on 7/1/15.
  20.  */
  21. public class Client implements ClientHandlerListener
  22. {

  24.     private final String host ;
  25.     private final int port ;



  29.     private ClientHandler handler ;
  30.     private ChannelFactory clientFactory ;
  31.     private ChannelGroup channelGroup ;


  34.     private String file_path ;
  35.     private String file_name ;
  36.     private long file_len ;
  37.     private BufferedReader input ;

  39.     public Client ( String host , int port )
  40.     {
  41.         this.host = host ;
  42.         this.port = port ;

  44.         // default received stored path
  45.         // make sure its existances before running the program

  47.         this.file_path = "/tmp/data/" ;
  48.     }



  52.     public void messageReceived (Message message)
  53.     {
  54.          // we not use it in this test
  55.     }

  57.    public boolean start ()
  58.    {
  59.        this.clientFactory = new NioClientSocketChannelFactory(
  60.                Executors.newCachedThreadPool(), Executors.newCachedThreadPool()) ;


  63.         this.channelGroup = new DefaultChannelGroup(this + "-channelGroup") ;

  65.         this.handler = new ClientHandler (this, this.channelGroup ) ;

  67.        ChannelPipelineFactory pipelineFactory = new ChannelPipelineFactory() {
  68.            public ChannelPipeline getPipeline() throws Exception {

  70.                ChannelPipeline pipeline = Channels.pipeline() ;
  71.                pipeline.addLast("encoder" , MessageEncoder.getInstance()) ;
  72.                pipeline.addLast("decoder" , new MessageDecoder ()) ;
  73.                pipeline.addLast("handler" , handler) ;


  76.                return pipeline ;
  77.             }
  78.        } ;

  80.        ClientBootstrap bootstrap = new ClientBootstrap (this.clientFactory) ;

  82.        bootstrap.setOption("reuseAddress" , true ) ;
  83.        bootstrap.setOption ("tcpNoDealy" , true ) ;
  84.        bootstrap.setOption ("keepAlive" , true ) ;
  85.        bootstrap.setPipelineFactory(pipelineFactory ) ;


  88.        boolean connected = bootstrap.connect( new InetSocketAddress(host, port ))
  89.                .awaitUninterruptibly().isSuccess();

  91.        if ( !connected)
  92.        {
  93.            this.stop() ;
  94.        }

  96.        return connected ;
  97.    }

  99.     public void stop ()
  100.     {
  101.         if ( this.channelGroup != null )
  102.             this.channelGroup.close() ;
  103.         if ( this.clientFactory != null )
  104.             this.clientFactory.releaseExternalResources();
  105.     }
  106.     private void flood()
  107.     {
  108.         if ((this.channelGroup == null) || (this.clientFactory == null)) {
  109.             System.out.println("do not have any resources, return ");
  110.             return;
  111.         }

  113.          /// runnin only once
  114.         {
  115.             String data = "KyLin_Zhang" ;

  117.             Message msg = new Message(MessageType.FILE_SENDING ,(short)(data.getBytes().length+3), data.getBytes()) ;
  118.           // System.out.println("client flood :" + msg ) ;

  120.             this.handler.sendMessage(msg);

  122.         }
  123.     }

  125.     public static void main ( String [] args ) throws InterruptedException
  126.     {
  127.         final Client client = new Client("kylin", 9999) ;



  131.             if (!client.start())
  132.             {
  133.                 System.out.println("client failed to start");
  134.                 System.exit(-1);
  135.                 return;
  136.             }

  138.         System.out.println("Client started ....") ;

  140.        // System.out.println("call flood") ;
  141.      // client.flood() ;


  144.          Runtime.getRuntime().addShutdownHook(
  145.                new Thread()
  146.                {
  147.                    @Override
  148.                     public void run ()
  149.                    {
  150.                         client.stop () ;
  151.                    }
  152.                }) ;
  153.     }
  154. }

client/ClientHandler.java

点击(此处)折叠或打开

  1. package org.kylin.zhang.client;

  3. import org.jboss.netty.channel.*;
  4. import org.jboss.netty.channel.group.ChannelGroup ;

  6. import org.kylin.zhang.common.* ;

  8. import java.io.*;

  10. /**
  11.  * Created by root on 7/1/15.
  12.  */
  13. public class ClientHandler extends SimpleChannelUpstreamHandler
  14. {

  16.     private final ClientHandlerListener listener ;
  17.     private final ChannelGroup channelGroup ;
  18.     private Channel channel ;

  20.     private String file_path ;
  21.     private String file_name ;
  22.     private long file_len ;
  23.     private BufferedWriter output ;


  26.     public ClientHandler(ClientHandlerListener listener , ChannelGroup channelGroup)
  27.     {
  28.         this.listener = listener ;
  29.         this.channelGroup = channelGroup ;

  31.         this.file_path = "/home/" ;
  32.     }

  34.     @Override
  35.     public void messageReceived ( ChannelHandlerContext ctx, MessageEvent e)
  36.         throws Exception
  37.     {
  38.         if ( e.getMessage () instanceof Message )
  39.         {
  40.             // output this
  41.           // this.listener.messageReceived((Message)e.getMessage()) ;

  43.             Message message = (Message)e.getMessage() ;

  45.             // print out message

  47.        // System.out.println("message received from server "+ message ) ;
  48.             // after receive message , create a new file and write the message content into it


  51.        // this.channelGroup.disconnect() ;

  53.             // in this branch , we first get the message type and then if else if it
  54.             // by different kinds of messages , we use different methods
  55.             // if - READY_SEND_FILE ---> we extract file name , and file len
  56.             // as static member values, and create the file at target path
  57.             // get its InputStream object , ready to write , and return the 'READY_RECV_FILE'
  58.             // message to the server

  60.             // if - FILE_SENDING ----> we extract file contents and write each line
  61.             // into the opened file

  63.             // if -- FILE_END ---> we write the last block of the message data , then
  64.             // we flush and then close the file



  68.             MessageType recv_msg_type = message.getType() ;

  70.             if ( recv_msg_type == MessageType.READY_SEND_FILE)
  71.             {
  72.                 String data = new String (message.getData()) ;

  74.                 file_name = data.substring( data.indexOf(':')+1) ;


  77.                 this.file_len = Long.decode(data.substring(0 , data.lastIndexOf(':'))) ;

  79.                 System.out.println ("we got file_name : "+ this.file_name ) ;
  80.                 System.out.println ("we got file_len : "+this.file_len ) ;

  82.                 // we create file and then open it , and then , create response message to server

  84.                 File file = new File( this.file_path+this.file_name ) ;

  86.                 if ( file.exists())
  87.                     file.delete() ;

  89.                 try
  90.                 {
  91.                     file.createNewFile();
  92.                     this.output = new BufferedWriter( new FileWriter( file )) ;
  93.                 }
  94.                 catch (Exception ex )
  95.                 {}

  97.                 // here we create message of type of READY_RECV_FILE

  99.                 data = "client get ready" ;
  100.                 Message response_msg = new Message(MessageType.READY_RECV_FILE , (short)(3+data.getBytes().length),
  101.                         data.getBytes()) ;
  102.                 this.channel.write(response_msg) ;
  103.             }

  105.             if ( recv_msg_type == MessageType.FILE_SENDING)
  106.             {
  107.                 String data = new String (message.getData()) ;
  108.                 this.output.newLine();
  109.                 this.output.write(data);
  110.                 this.output.flush();

  112.             }

  114.             if ( recv_msg_type == MessageType.FILE_END)
  115.             {
  116.                 String data = new String (message.getData()) ;

  118.                 this.output.newLine();
  119.                 this.output.write(data) ;
  120.                 this.output.flush() ;

  122.                 this.output.close() ;

  124.                 // create message with type of SHUT_DOWN , after server received this it will
  125.                 // disconnect the channel

  127.                 data = "client gonning to shut down" ;
  128.                 Message msg = new Message (MessageType.SHUT_DOWN , (short)(3+data.getBytes().length) ,
  129.                         data.getBytes()) ;

  131.                 e.getChannel().write(msg) ;

  133.                 try
  134.                 {
  135.                     System.out.println("client going to shut down ......") ;
  136.                     Thread.sleep(4000);
  137.                     e.getChannel().disconnect() ;
  138.                 }
  139.                 catch ( Exception ex )
  140.                 {}
  141.             }

  143.         }
  144.         else
  145.         {
  146.             super.messageReceived(ctx, e ) ;
  147.         }

  149.     }

  151.     @Override
  152.     public void channelConnected( ChannelHandlerContext ctx, ChannelStateEvent e)
  153.         throws Exception
  154.     {
  155.         System.out.println("client channelConnected: ") ;
  156.         this.channel = e.getChannel();



  160.         this.channelGroup.add(e.getChannel()) ;

  162.     }

  164.     public void sendMessage ( Message msg )
  165.     {
  166.         if ( this.channel != null )
  167.         {
  168.             System.out.println("sendMessage : "+ msg ) ;
  169.             this.channel.write(msg) ;

  171.         }
  172.     }
  173. }

client/ClientHandlerListener.java / not use
/ not use it , copy it from the previous blog is ok
server/Server.java

点击(此处)折叠或打开

  1. package org.kylin.zhang.server;

  3. import org.kylin.zhang.common.* ;

  5. import org.jboss.netty.bootstrap.ServerBootstrap;
  6. import org.jboss.netty.channel.Channel ;
  7. import org.jboss.netty.channel.ChannelPipeline ;
  8. import org.jboss.netty.channel.ChannelPipelineFactory ;
  9. import org.jboss.netty.channel.Channels ;
  10. import org.jboss.netty.channel.ServerChannelFactory ;
  11. import org.jboss.netty.channel.group.DefaultChannelGroup;
  12. import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory ;

  14. import java.net.InetSocketAddress ;
  15. import java.util.concurrent.Executors ;


  18. /**
  19.  * Created by root on 7/1/15.
  20.  */
  21. public class Server {

  23.     private final String host ;
  24.     private final int port ;
  25.     private DefaultChannelGroup channelGroup ;
  26.     private ServerChannelFactory serverFactory ;



  30.     public Server ( String host , int port )
  31.     {
  32.         this.host = host ;
  33.         this.port = port ;

  35.     }

  37.     public boolean start ()
  38.     {
  39.         this.serverFactory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
  40.                 Executors.newCachedThreadPool()) ;

  42.         this.channelGroup = new DefaultChannelGroup (this + "-channelGroup") ;

  44.         ChannelPipelineFactory pipelineFactory = new ChannelPipelineFactory ()
  45.         {
  46.             public ChannelPipeline getPipeline() throws Exception
  47.             {
  48.                 ChannelPipeline pipeline = Channels.pipeline() ;
  49.                 pipeline.addLast ("encoder" , MessageEncoder.getInstance()) ;
  50.                 pipeline.addLast("decoder" , new MessageDecoder() ) ;
  51.                 pipeline.addLast("handler" , new ServerHandler(channelGroup)) ;

  53.                 return pipeline ;
  54.             }
  55.         } ;

  57.         ServerBootstrap bootstrap = new ServerBootstrap(this.serverFactory) ;
  58.         bootstrap.setOption("reuseAddress" , true) ;
  59.         bootstrap.setOption ("child.tcpNoDelay" , true ) ;
  60.         bootstrap.setOption ("child.keepAlive" , true ) ;
  61.         bootstrap.setPipelineFactory(pipelineFactory) ;


  64.         Channel channel = bootstrap.bind( new InetSocketAddress(this.host, this.port)) ;

  66.         if ( !channel.isBound())
  67.         {
  68.             this.stop() ;
  69.             return false ;
  70.         }

  72.         this.channelGroup.add(channel) ;
  73.         return true ;
  74.     }

  76.     public void stop ()
  77.     {
  78.         if (this.channelGroup != null )
  79.             this.channelGroup.close () ;
  80.         if ( this.serverFactory != null )
  81.             this.serverFactory.releaseExternalResources();
  82.     }


  85.     public static void main ( String [] args )
  86.     {
  87.         final Server server = new Server("kylin", 9999 ) ;

  89.         if ( !server.start ())
  90.         {
  91.             System.out.println("server failed to run ") ;
  92.             System.exit(-1);

  94.             return ; // not really needed
  95.         }

  97.         System.out.println("server started ..... ") ;

  99.         Runtime.getRuntime().addShutdownHook( new Thread ()
  100.         {
  101.             @Override
  102.             public void run ()
  103.             {
  104.                 server.stop() ;
  105.             }
  106.         });
  107.     }


  110. }


server/ServerHandler.java

点击(此处)折叠或打开

  1. package org.kylin.zhang.server;

  3. import org.kylin.zhang.common.* ;
  4. import org.jboss.netty.channel.ChannelHandlerContext ;
  5. import org.jboss.netty.channel.ChannelStateEvent ;
  6. import org.jboss.netty.channel.MessageEvent ;
  7. import org.jboss.netty.channel.SimpleChannelUpstreamHandler ;
  8. import org.jboss.netty.channel.group.ChannelGroup ;

  10. import java.io.BufferedReader;
  11. import java.io.File;
  12. import java.io.FileReader;
  13. import java.util.HashMap;
  14. import java.util.Iterator;
  15. import java.util.Map;

  17. /**
  18.  * Created by root on 7/1/15.
  19.  */
  20. public class ServerHandler extends SimpleChannelUpstreamHandler
  21. {
  22.     private final ChannelGroup channelGroup ;
  23.     private Map<String,String> fileHash ;

  25.     public ServerHandler ( ChannelGroup channelGroup )
  26.     {

  28.         this.channelGroup = channelGroup ;

  30.         this.fileHash = new HashMap<String,String>() ;

  32.         // here we initialize the fileHash
  33.         fileHash.put("test1" , "/tmp/test1.txt") ;
  34.         fileHash.put("test2" , "/tmp/test2.txt") ;
  35.     }

  37.     @Override
  38.     public void channelConnected( ChannelHandlerContext ctx, ChannelStateEvent e )
  39.     {
  40.         System.out.println("-------------------------- **server gets a new connection**--------------------------") ;
  41.         this.channelGroup.add(e.getChannel()) ;


  44.         byte [] data = null ;

  46.         File file = new File ( fileHash.get("test1") ) ;
  47.         String len_fname = String.valueOf(file.length()) ;
  48.         len_fname += ":test1" ;

  50.         // file_length:file_name

  52.         data = len_fname.getBytes() ;

  54.         Message msg = new Message(MessageType.READY_SEND_FILE , (short)(3+data.length) ,data);

  56.         e.getChannel().write(msg) ;

  58.      // e.getChannel().disconnect() ;
  59.     }

  61.     @Override
  62.     public void messageReceived ( ChannelHandlerContext ctx, MessageEvent e )
  63.             throws Exception
  64.     {

  66.         if ( e.getMessage() instanceof Message ) {


  69.             Message received_msg = (Message) e.getMessage();

  71.             MessageType recv_msg_type = received_msg.getType () ;

  73.             System.out.println("server received message: " + received_msg);

  75.             if ( recv_msg_type == MessageType.READY_RECV_FILE)
  76.             {
  77.                 // in this open the file and use while cycle send file data <=250 bytes to client
  78.                 // if read the last line 'the end of the file!'
  79.                 // change the response file state from SENDING_FILE to END_FILE
  80.                 // and then close the channel

  82.                 File f = new File (this.fileHash.get("test1")) ;
  83.                 BufferedReader input = new BufferedReader( new FileReader(f) ) ;

  85.                 String line ;

  87.                 while(!( line = input.readLine()).equals("the end of file!"))
  88.                 {
  89.                     Message send_msg = new Message( MessageType.FILE_SENDING , (short)(3+line.getBytes().length) ,
  90.                             line.getBytes()) ;

  92.                     e.getChannel().write(send_msg) ;
  93.                 }

  95.                 if (line.equals("the end of file!"))
  96.                 {
  97.                     Message send_msg = new Message (MessageType.FILE_END, (short)(3+line.getBytes().length),
  98.                             line.getBytes() ) ;
  99.                     e.getChannel().write(send_msg) ;
  100.                 }

  102.             }
  103.             else if ( recv_msg_type == MessageType.SHUT_DOWN)
  104.             {
  105.                 e.getChannel().disconnect() ;
  106.             }
  107.             else
  108.             {
  109.                 System.out.println ("UNKNOWN Message ") ;
  110.             }


  113.         }
  114.         else
  115.         {
  116.             super.messageReceived(ctx, e);
  117.         }

  119.     }
  120. }

运行结果:

先运行服务器端
(我在程序中，使用的是自己的主机别名创建的连接，根据不同的主机需要变动 hostname):

server started .....
-------------------------- **server gets a new connection**--------------------------
server received message:
message type :READY_RECV_FILE
message length :19
message content :client get ready

server received message:
message type :SHUT_DOWN
message length :30
message content :client gonning to shut down


在运行客户端:

Client started ....
client channelConnected:
we got file_name : test1
we got file_len : 17699324
client going to shut down ......

Process finished with exit code 0

end
上一篇：[java,jboss-netty]使用 netty 框架实现自定义消息类型的接收与发送_草稿篇
下一篇：[mysql] mysql 源码安装解压 rpm 包命令

文章评论