Mina的使用简介

川长思鸟来 2022-07-14 05:28 185阅读 0赞

相信大家都遇到过Socket通信的问题，下面给大家介绍些我在公司项目遇到过的这样的功能，介绍Mina的基本使用。

什么都不说了，直接上代码。

/** * @author dongyu * 文件传输服务器 */ public class MinaFileServerHandler extends StreamIoHandler {
        public static final int PORT = 8888;
        public static final String TAG = MinaFileServerHandler.class.getSimpleName();
    
        private static MinaFileServerHandler mMinaFileServerHandler = null;
    
        private long mMediaId = -1;
    
        private String mToSaveFilePath = "";
    
        private boolean mIsToSaveFile = false;
    
        private IoSession session;
        private  String md5sum;
    private   NioSocketAcceptor acceptor;
        public void setMediaId(long id) {
            mMediaId = id;
        }
    
        public void setToSaveFilePath(String path) {
            mToSaveFilePath = path;
        }
    
        public void setIsToSaveFile(boolean isToSaveFile) {
            mIsToSaveFile = isToSaveFile;
        }
    
        public String getMd5sum() {
            return md5sum;
        }
    
        public void setMd5sum(String md5sum) {
            this.md5sum = md5sum;
        }
    
        public static MinaFileServerHandler getInstance() {
            if (null == mMinaFileServerHandler) {
                synchronized (MinaFileServerHandler.class) {
                    if (null == mMinaFileServerHandler) {
                        mMinaFileServerHandler = new MinaFileServerHandler();
                    }
                }
            }
            return mMinaFileServerHandler;
        }
    
        public IoSession getSession() {
            return session;
        }
    
        public void setSession(IoSession session) {
            LogUtils.e(TAG, " setSession session =  "+session+" this.session =  "+this.session);
            this.session = session;
        }
    
        @Override
        public void sessionOpened(IoSession session) {
            LogUtils.e("MinaFileServerHandler", "客户端连接了:" + session.getRemoteAddress());
            super.sessionOpened(session);
        }
    
        protected void processStreamIo(IoSession session, InputStream in, OutputStream out) {
            //设定一个线程池
            //参数说明：最少数量3，最大数量6 空闲时间 3秒
            ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, 6, 3, TimeUnit.SECONDS,
                    //缓冲队列为3
                    new ArrayBlockingQueue<Runnable>(3),
                    //抛弃旧的任务
                    new ThreadPoolExecutor.DiscardOldestPolicy());
            FileOutputStream fos = null;
            //将线程放入线程池 当连接很多时候可以通过线程池处理
            LogUtils.e(TAG, "processStreamIo to save excel file id = " + mMediaId);
            LogUtils.e(TAG, "processStreamIo to save excel file mIsToSaveFile = " + mIsToSaveFile);
            LogUtils.e(TAG, "processStreamIo to save excel file mToSaveFilePath = " + mToSaveFilePath);
            //以当前时间作为要保存的文件名
            IoStreamThreadWork mIoStreamThreadWork = new IoStreamThreadWork(in, mToSaveFilePath, mIsToSaveFile);
            mIoStreamThreadWork.setMediaId(mMediaId);
            LogUtils.e(TAG, "session =  "+getSession());
            LogUtils.e(TAG, "session =  "+session+" this.session =  "+this);
            mIoStreamThreadWork.setSession(getSession());
            mIoStreamThreadWork.setMd5sum(getMd5sum());
            threadPool.execute(mIoStreamThreadWork);
            //直接启动线程 连接很少可以选用下面
    //      new IoStreamThreadWork(in,fos).start();
        }
    
        @Override
        public void messageReceived(IoSession session, Object buf) {
    
    
            super.messageReceived(session, buf);
        }
    
        @Override
        public void sessionClosed(IoSession session) throws Exception {
            super.sessionClosed(session);
    
    
        }
    
        @Override
        public void sessionIdle(IoSession session, IdleStatus status) {
            super.sessionIdle(session, status);
            LogUtils.e("MinaFileServerHandler", "文件服务器进入空闲状态！");
        }
    
        @Override
        public void exceptionCaught(IoSession session, Throwable cause) {
            super.exceptionCaught(session, cause);
            LogUtils.disposeThrowable(TAG,cause);
        }
    
        public void createServerStream() {
            if(null==acceptor) {
                //建立一个无阻塞服务端socket 用nio
                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(this);
                InetSocketAddress inetSocketAddress = null;
                try {
                    inetSocketAddress = new InetSocketAddress(3345);
                    acceptor.bind(inetSocketAddress);
                    LogUtils.e("文件服务器已经开启：", " " + 3345);
                    Toast.makeText(MBroadcastApplication.getMyContext(), "音响接收文件服务启动成功", Toast.LENGTH_LONG).show();
                } catch (IOException e) {
                    LogUtils.e("文件服务器已经开启：", " " + "服务器启动异常...");
                    LogUtils.e("文件服务器已经开启：", " " + e);
                    Toast.makeText(MBroadcastApplication.getMyContext(), "音响接收文件服务启动失败" + e, Toast.LENGTH_LONG).show();
                    acceptor.dispose(false);
                    e.printStackTrace();
                }
            }
        }
    
    }

/** * Created by dongyu @文件传输的Thread

*/ public class MyThread extends Thread  {
        private final String TAG = this.getClass().getSimpleName();
        File msendFile;
        String ip;
        private IoSession session = null;
        public String getIp() {
            return ip;
        }
    
        public void setIp(String ip) {
            this.ip = ip;
        }
    
    
        public File getSendFile() {
            return msendFile;
        }
    
        public void setSendFile(File sendFile) {
            LogUtils.e("MinaFileClient", "this = " + this);
            LogUtils.e("MinaFileClient","setSendFile sendFile = " + sendFile);
            this.msendFile = sendFile;
        }
    
        public void setSendFile(String path) {
            LogUtils.e("MinaFileClient", "this = " + this);
            LogUtils.e("MinaFileClient","setSendFile path = " + path);
            this.msendFile = new File(path);
        }
    
        @Override
        public void run() {
            super.run();
            int port = 3345;
            NioSocketConnector connector = new NioSocketConnector();
            DefaultIoFilterChainBuilder chain = connector.getFilterChain();
            ObjectSerializationCodecFactory factory = new ObjectSerializationCodecFactory();
            factory.setDecoderMaxObjectSize(Integer.MAX_VALUE);
            factory.setEncoderMaxObjectSize(Integer.MAX_VALUE);
            connector.setHandler(new MinaFileClient(this));
            ConnectFuture connectFuture = connector.connect(new InetSocketAddress(ip,port));
            connectFuture.awaitUninterruptibly();//写上这句为了得到下面的session 意思是等待连接创建完成 让创建连接由异步变同步
    
        }
    }

/** * @author dongyu 文件传输客户端 */ public class MinaFileClient extends StreamIoHandler {
       private  final  String TAG=this.getClass().getSimpleName();
       private IoSession session;
    
       private MyThread mMyThread;
    
       public void setSession(IoSession session) {
          this.session = session;
       }
    
       public IoSession getSession() {
          return session;
       }
    
       public MinaFileClient(MyThread thread){
          mMyThread = thread;
       }
    
       @Override
       public void sessionCreated(IoSession session) throws Exception {
          super.sessionCreated(session);
          LogUtils.e(TAG,"建立链接");
       }
    
       @Override
       public void messageReceived(IoSession session, Object buf) {
          super.messageReceived(session, buf);
       }
    
       @Override
       protected void processStreamIo(IoSession session, InputStream in, OutputStream out) {
          // 客户端发送文件
          LogUtils.e(TAG,session.getId()+"  ");
          LogUtils.e(TAG, "this = " + this);
          LogUtils.e(TAG, "processStreamIo sendFile = " + mMyThread.getSendFile());
          //File sendFile = new File(Environment.getExternalStorageDirectory().toString()+"/1.xlsx");
          FileInputStream fis = null;
          try {
             fis = new FileInputStream(mMyThread.getSendFile());
    
          } catch (FileNotFoundException e) {
             e.printStackTrace();
             LogUtils.e(TAG,e.toString());
          }
          // 放入线程让其执行
          // 客户端一般都用一个线程实现即可 不用线程池
          new IoStreamThreadWork(fis, out).start();
       }
    
       @Override
       public void exceptionCaught(IoSession session, Throwable cause) {
          super.exceptionCaught(session, cause);
    
          LogUtils.e(TAG,cause+"  ");
          LogUtils.disposeThrowable(TAG,cause);
       }
    
       public void createClienStream(String ip) {
          int port = 3345;
          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(this);
          ConnectFuture connectFuture = connector.connect(new InetSocketAddress(ip,port));
          connectFuture.awaitUninterruptibly();//写上这句为了得到下面的session 意思是等待连接创建完成 让创建连接由异步变同步
          //后来表明我开始的想法不行 动态依旧不能做到
    //      @SuppressWarnings("unused")
    //      IoSession session = connectFuture.getSession();
    //      setSession(session);
       }
    }

上面就可以实现文件的传输了。