Apache Mina: StreamIoHandler传输文件处理

通过StreamIoHandler来进行文件的传输

1. 创建通过接收的BufferedInputStream写输出BufferedOutputStream的方法

public class IoStreamThreadWork extends Thread {
	
	public static final int BUFFER_SIZE = 1024*2;  
     
	private BufferedInputStream bis;  
	private BufferedOutputStream bos;
	
	public BufferedInputStream getBis() {
		return bis;
	}
	public void setBis(BufferedInputStream bis) {
		this.bis = bis;
	}
	public BufferedOutputStream getBos() {
		return bos;
	}
	public void setBos(BufferedOutputStream bos) {
		this.bos = bos;
	}  
	
	public IoStreamThreadWork(InputStream in, OutputStream os){  
        bis = new BufferedInputStream(in);  
        bos = new BufferedOutputStream(os);  
    }  
    public synchronized void run() {  
        byte[] bufferByte = new byte[BUFFER_SIZE];  
        int tempData = 0;  
        try {  
            while((tempData = bis.read(bufferByte)) != -1 ){  
                bos.write(bufferByte, 0, tempData);  
            }  
            try {  
                bos.flush();  
            } catch (IOException e) {  
                e.printStackTrace();  
            }  
        } catch (IOException e) {  
            e.printStackTrace();  
        }finally{  
            try {  
                bos.close();  
                bis.close();  
            } catch (IOException e) {  
                e.printStackTrace();  
            }  
        }  
    }  

}

2. 创建Server端服务及其StreamIoHandler
Server.java

public class Server {
	
	public Server(){
		
	}
	
	public void init() throws IOException{
		IoAcceptor acceptor = new NioSocketAcceptor();
		
        ObjectSerializationCodecFactory factory = new ObjectSerializationCodecFactory();
        factory.setDecoderMaxObjectSize(Integer.MAX_VALUE);  
        factory.setEncoderMaxObjectSize(Integer.MAX_VALUE);
		
		acceptor.getFilterChain().addLast("logger", new LoggingFilter());
	
        acceptor.setHandler(new MyStreamIoHandler());
		acceptor.getSessionConfig().setReadBufferSize(2048);
		acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 10);

        acceptor.setDefaultLocalAddress(new InetSocketAddress(Constants.PORT));
        acceptor.bind();// 启动监听  
	}

	/**
	 * @param args
	 * @throws IOException 
	 */
	public static void main(String[] args) throws IOException {
		Server server = new Server();
		server.init();

	}

}

MyStreamIoHandler.java

public class MyStreamIoHandler extends StreamIoHandler {
	
	

	@Override
	public void sessionOpened(IoSession session) {
		System.out.println("客户端连接了:"+session.getRemoteAddress());  
		super.sessionOpened(session);
	}

	@Override
	protected void processStreamIo(IoSession session, InputStream in,
			OutputStream out) {
		
		//设定一个线程池  
        //参数说明：最少数量3，最大数量6 空闲时间 3秒  
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3, 6, 3,TimeUnit.SECONDS,   
                //缓冲队列为3  
                new ArrayBlockingQueue<Runnable>(3),  
                //抛弃旧的任务  
                new ThreadPoolExecutor.DiscardOldestPolicy());  
        FileOutputStream fos = null;  
        File receiveFile = new File("e:\\hello.doc");  
        try {  
            fos = new FileOutputStream(receiveFile);  
        } catch (FileNotFoundException e1) {  
            e1.printStackTrace();  
        }  
        //将线程放入线程池 当连接很多时候可以通过线程池处理  
        threadPool.execute(new IoStreamThreadWork(in,fos));  
	}
}

3. 创建Client端连接机器StreamIoHandler
Client.java

public class Client {
	
	public Client(){
		super();
	}
	public void connect() throws InterruptedException{
		NioSocketConnector connector = new NioSocketConnector();
		
	    ObjectSerializationCodecFactory factory = new ObjectSerializationCodecFactory();  
	    factory.setDecoderMaxObjectSize(Integer.MAX_VALUE);  
	    factory.setEncoderMaxObjectSize(Integer.MAX_VALUE);  

        // Configure the service.
        connector.setConnectTimeoutMillis(Constants.CONNECT_TIMEOUT);
        //connector.getFilterChain().addLast("codec",new ProtocolCodecFilter(factory));
        connector.getFilterChain().addLast("logger", new LoggingFilter());

        connector.setHandler(new ClientStreamIoHandler());
        

        IoSession session;
        for (;;) {
            try {
                ConnectFuture future = connector.connect(new InetSocketAddress(Constants.HOSTNAME, Constants.PORT));
                future.awaitUninterruptibly();
                session = future.getSession();
                break;
            } catch (RuntimeIoException e) {
                System.err.println("Failed to connect.");
                e.printStackTrace();
                Thread.sleep(5000);
            }
        }

        // wait until the summation is done
        session.getCloseFuture().awaitUninterruptibly();
        
        connector.dispose();
	}

	/**
	 * @param args
	 * @throws InterruptedException 
	 */
	public static void main(String[] args) throws InterruptedException {
		Client client = new Client();
		client.connect();

	}

}

ClientStreamIoHandler.java

public class ClientStreamIoHandler extends StreamIoHandler {

	@Override
	protected void processStreamIo(IoSession session, InputStream in,
			OutputStream out) {
		
        //客户端发送文件  
        File sendFile = new File("D:\\ttt.doc");  
        FileInputStream fis = null;  
        try {  
            fis = new FileInputStream(sendFile);  
              
        } catch (FileNotFoundException e) {  
            e.printStackTrace();  
        }  
        //放入线程让其执行  
         //客户端一般都用一个线程实现即可 不用线程池  
        new IoStreamThreadWork(fis,out).start();  
        return;  

	}

}

评论

1 楼 u014549257 2014-08-15