用mina传输资料

用mina传输文件

package com.rose.common.frame.mina;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
/**
* 用于mina 服务器上传下载
* 流处理线程公共类
* @author rose
*
*/
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();
}
}
}
}

服务端：

package com.rose.common.frame.mina;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.serialization.ObjectSerializationCodecFactory;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.handler.stream.StreamIoHandler;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
/**
* @author rose
* 文件传输服务器
*/
public class MinaFileServer extends StreamIoHandler{
public static final int PORT = 8888;
@Override
public void sessionOpened(IoSession session) {
System.out.println("客户端连接了:"+session.getRemoteAddress());
super.sessionOpened(session);
}
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:\\test.pdf");
try {
fos = new FileOutputStream(receiveFile);
} catch (FileNotFoundException e1) {
e1.printStackTrace();
}
//将线程放入线程池当连接很多时候可以通过线程池处理
threadPool.execute(new IoStreamThreadWork(in,fos));
//直接启动线程连接很少可以选用下面
// new IoStreamThreadWork(in,fos).start();
}
public void createServerStream(){
//建立一个无阻塞服务端socket 用nio
NioSocketAcceptor acceptor = new NioSocketAcceptor();
//创建接收过滤器也就是你要传送对象的类型
DefaultIoFilterChainBuilder chain = acceptor.getFilterChain();
//===========过滤器创建好了就开始设定============
//设定对象传输工厂
ObjectSerializationCodecFactory factory = new ObjectSerializationCodecFactory();
//设定传输最大值
factory.setDecoderMaxObjectSize(Integer.MAX_VALUE);// 设定后服务器可以接收大数据
factory.setEncoderMaxObjectSize(Integer.MAX_VALUE);
chain.addLast("logging", new LoggingFilter());//这个用于打印日志可以不写
//设定服务端消息处理器
acceptor.setHandler(new MinaFileServer());
InetSocketAddress inetSocketAddress = null;
try {
inetSocketAddress = new InetSocketAddress(8888);
acceptor.bind(inetSocketAddress);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println("文件服务器已经开启："+8888);
}
public static void main(String[] args) {
MinaFileServer server = new MinaFileServer();
server.createServerStream();
}
}

客户端：

package com.rose.common.frame.mina;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.serialization.ObjectSerializationCodecFactory;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.handler.stream.StreamIoHandler;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
/**
* @author rose
* 文件传输客户端
*/
public class MinaFileClient extends StreamIoHandler{
IoSession session;
public void setSession(IoSession session) {
this.session = session;
}
public IoSession getSession() {
return session;
}
@Override
protected void processStreamIo(IoSession session, InputStream in,
OutputStream out) {
//客户端发送文件
File sendFile = new File("F:\\ttt.pdf");
FileInputStream fis = null;
try {
fis = new FileInputStream(sendFile);
} catch (FileNotFoundException e) {
e.printStackTrace();
}
//放入线程让其执行
//客户端一般都用一个线程实现即可不用线程池
new IoStreamThreadWork(fis,out).start();
return;
}
public void createClienStream(){
int port = 8888;
String local = "127.0.0.1";
NioSocketConnector connector = new NioSocketConnector();
DefaultIoFilterChainBuilder chain = connector.getFilterChain();
ObjectSerializationCodecFactory factory = new ObjectSerializationCodecFactory();
factory.setDecoderMaxObjectSize(Integer.MAX_VALUE);
factory.setEncoderMaxObjectSize(Integer.MAX_VALUE);
chain.addLast("logging", new LoggingFilter());//用于打印日志可以不写
connector.setHandler(new MinaFileClient());
ConnectFuture connectFuture = connector.connect(new InetSocketAddress(local,port));
connectFuture.awaitUninterruptibly();//写上这句为了得到下面的session 意思是等待连接创建完成让创建连接由异步变同步
//后来表明我开始的想法不行动态依旧不能做到
// @SuppressWarnings("unused")
// IoSession session = connectFuture.getSession();
// setSession(session);
}
public static void main(String[] args) {
MinaFileClient client = new MinaFileClient();
client.createClienStream();
}
}

相关推荐