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;  
}
}