Socket编程及mina框架简单示例

Posted 2022-12-09 洽洽老大

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Socket编程及mina框架简单示例相关的知识，希望对你有一定的参考价值。

要实现客户端与服务器的长连接，可以使用socket的方式连接服务器与客户端。在这篇文章中，将用原生的方式实现socket的服务器端和客户端，然后用Mina框架再实现一次。
原生方式上：
客户端可实现如下：
SocketClient:

package socketClient;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.Socket;

public class SocketClient 

    public int port = 9898;
    public String hostAddress = "127.0.0.1";

    public static void main(String[] args) 
        SocketClient client = new SocketClient();
        client.start();
    

    private void start() 
        BufferedReader inputReader = null;
        OutputStreamWriter output = null;
        Socket socket = null;
        try 
            socket = new Socket(hostAddress, port);
            inputReader = new BufferedReader(new InputStreamReader(System.in));
            output = new OutputStreamWriter(socket.getOutputStream());
            String inputContent;
            int count = 0;
            while (!(inputContent = inputReader.readLine()).equals("bye")) 
                output.write(inputContent);
                    output.write("\\n");
                output.flush();
                getServerMsg(socket);
            
         catch (IOException e) 
            e.printStackTrace();
         finally 
            try 
                output.close();
                inputReader.close();
                socket.close();
             catch (IOException e) 
                e.printStackTrace();
            
        
    

    private void getServerMsg(Socket socket) 
        new Thread(new Runnable() 

            @Override
            public void run() 
                BufferedReader reader = null;
                try 
                    reader = new BufferedReader(new InputStreamReader(
                            socket.getInputStream()));
                    String serverMsg;
                    while ((serverMsg = reader.readLine()) != null) 
                        System.out.println("server say: " + serverMsg);
                    
                 catch (IOException e) 
                    e.printStackTrace();
                 finally 
                    try 
                        reader.close();
                     catch (IOException e) 
                        e.printStackTrace();
                    
                
            
        ).start();

服务器端：

package com.socket.tra;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.ServerSocket;
import java.net.Socket;

public class SocketServer 

    public static void main(String[] args) 
        SocketServer server = new SocketServer();
        server.startServer();
    

    private void startServer() 
        ServerSocket serverSocket = null;
        Socket socket = null;
        try 
            serverSocket = new ServerSocket(9898);
            while (true) 
                socket = serverSocket.accept();
                System.out.println(socket.hashCode() + " is connect");
                connect(socket);
            
         catch (IOException e) 
            e.printStackTrace();
        
    

    private void connect(final Socket socket) 
        new Thread(new Runnable() 

            public void run() 
                BufferedReader reader = null;
                OutputStreamWriter writer = null;

                try 
                    reader = new BufferedReader(new InputStreamReader(
                            socket.getInputStream()));
                    writer = new OutputStreamWriter(socket.getOutputStream());
                    String msg;
                    while ((msg = reader.readLine()) != null) 
                        System.out.println(socket.hashCode()+"say: "+msg);
                        writer.write(msg + "\\n");
                        writer.flush();
                    
                 catch (IOException e) 
                    e.printStackTrace();
                 finally 
                    try 
                        writer.close();
                        reader.close();
                        socket.close();
                     catch (IOException e) 
                        e.printStackTrace();
                    
                
            
        ).start();

这是socket基本的用法，但在实际开发中，我们一般用封装好的框架来实现，Apache mina就是能够帮助用户开发高性能和高伸缩性网络应用程序的框架。它通过Java nio技术基于TCP/IP和UDP/IP协议提供了抽象的、事件驱动的、异步的API。
具体用法可以去官网了解下，这里提供一个简单的使用示例，实现跟上面原生方法同样的功能。
版本一：
服务器端:

package com.socket;

import java.io.IOException;
import java.net.InetSocketAddress;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.transport.socket.nio.NiosocketAcceptor;

public class Main 
    public static int port = 9898;
    public static void main(String[] args) 

        NioSocketAcceptor acceptor = new NioSocketAcceptor();

        try 
            //设置handler
            acceptor.setHandler(new MyHandler());
            //设置过滤器
            acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new TextLineCodecFactory()));
            //绑定端口号
            acceptor.bind(new InetSocketAddress(port));

         catch (IOException e) 
            e.printStackTrace();

MyHandler:

package com.socket;

import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;

public class MyHandler implements IoHandler 

    public void exceptionCaught(IoSession arg0, Throwable arg1)
            throws Exception 
        System.out.println("exception");
    

    public void inputClosed(IoSession arg0) throws Exception 
        System.out.println("inputClosed");
    

    public void messageReceived(IoSession arg0, Object arg1) throws Exception 
        String msg = (String) arg1;
        System.out.println("messageReceived server: " + msg);
        arg0.write(msg);
    

    public void messageSent(IoSession arg0, Object arg1) throws Exception 
        System.out.println("messageSent");
    

    public void sessionClosed(IoSession arg0) throws Exception 
        System.out.println("sessionClosed "+arg0.hashCode());
    

    public void sessionCreated(IoSession arg0) throws Exception 
        System.out.println("sessionCreated "+arg0.hashCode());

    

    public void sessionIdle(IoSession arg0, IdleStatus arg1) throws Exception 
        System.out.println("sessionIdle "+arg0.hashCode()+" , "+arg1);
    

    public void sessionOpened(IoSession arg0) throws Exception 
        System.out.println("sessionOpened "+arg0.hashCode());

客户端：

package socketClient.mina;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.InetSocketAddress;
import java.net.Socket;

import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.transport.socket.nio.NioSocketConnector;

public class SocketClient 

    public int port = 9898;
    public String hostAddress = "127.0.0.1";

    public static void main(String[] args) throws IOException 

        NioSocketConnector connector = new NioSocketConnector();
        connector.setHandler(new MyClientHandler());
        connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(new TextLineCodecFactory()));
        ConnectFuture future = connector.connect(new InetSocketAddress("127.0.0.1", 9898));
        future.awaitUninterruptibly();//等待连接
        IoSession session = future.getSession();
        BufferedReader inputReader = new BufferedReader(new InputStreamReader(System.in));
        String inputContent;
        while (!(inputContent = inputReader.readLine()).equals("bye")) 
            session.write(inputContent);

MyClientHandler:

package socketClient.mina;

import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;

public class MyClientHandler implements IoHandler 

    public void exceptionCaught(IoSession arg0, Throwable arg1)
            throws Exception 
        System.out.println(arg1.getCause());
    

    public void inputClosed(IoSession arg0) throws Exception 
//      System.out.println("inputClosed");
    

    public void messageReceived(IoSession arg0, Object arg1) throws Exception 
        String msg = (String) arg1;
        System.out.println("client messageReceived: " + msg);
    

    public void messageSent(IoSession arg0, Object arg1) throws Exception 
        System.out.println("client messageSent->" + (String)arg1);
    

    public void sessionClosed(IoSession arg0) throws Exception 
        System.out.println("sessionClosed "+arg0.hashCode());
    

    public void sessionCreated(IoSession arg0) throws Exception 
        System.out.println("sessionCreated "+arg0.hashCode());



    

    public void sessionIdle(IoSession arg0, IdleStatus arg1) throws Exception 
        System.out.println("sessionIdle "+arg0.hashCode()+" , "+arg1);
    

    public void sessionOpened(IoSession arg0) throws Exception 
        System.out.println("sessionOpened "+arg0.hashCode());

版本一使用框架写好的TextLineCodecFactory来解析字符串，在实际实用场合中，往往要自定义解析功能，因此版本二自己写一个字符串解析功能。
版本二：
服务器端：
Main: 主函数
MyDecoder: 实现数据的解码
MyEncoder: 实现数据的编码
MyHandler:
MyProtocolFactory: 生成编码和解码器
MyCumulativeEncoder: 实现数据的编码，可将服务器数据进行缓存，防止数据丢失

Main:

package com.socket.r1;

import java.io.IOException;
import java.net.InetSocketAddress;

import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;

public class Main 
    public static int port = 9898;
    public static void main(String[] args) 
        NioSocketAcceptor acceptor = new NioSocketAcceptor();

        try 
            acceptor.setHandler(new MyHandler());
            acceptor.getFilterChain().addLast("logger", new LoggingFilter());
            acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new MyProtocolCodecFactory()));
            acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 125);
            acceptor.bind(new InetSocketAddress(port));     

         catch (IOException e) 
            e.printStackTrace();

MyDecoder:

package com.socket.r1;

import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolDecoder;
import org.apache.mina.filter.codec.ProtocolDecoderOutput;

public class MyDecoder implements ProtocolDecoder 

    public void decode(IoSession session, IoBuffer in, ProtocolDecoderOutput output)
            throws Exception 
        //记录字符流的读取位置
        int startPosition = in.position();
        while(in.hasRemaining())
            byte b = in.get();
            if(b == '\\n')
                int curPosition = in.position();
                //记录字符流的末位置
                int limit = in.limit();
                //将读取指针设置为初始位置
                in.position(startPosition);
                //将结束位置设置为当前读取位置
                in.limit(curPosition);
                IoBuffer buf = in.slice();
                byte[] bytes = new byte[buf.limit()];
                //将截取的内容放进bytes数组
                buf.get(bytes);
                String str = new String(bytes);
                output.write(str);
                in.position(curPosition);
                in.limit(limit);
            
        
    

    public void dispose(IoSession arg0) throws Exception 
        System.out.println("dispose" + arg0.hashCode());

    

    public void finishDecode(IoSession arg0, ProtocolDecoderOutput arg1)
            throws Exception 
        System.out.println("finishDecode" + arg0.hashCode());

MyEncoder:

package com.socket.r1;

import java.nio.charset.Charset;
import java.nio.charset.CharsetEncoder;

import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolEncoder;
import org.apache.mina.filter.codec.ProtocolEncoderOutput;

public class MyEncoder implements ProtocolEncoder 

    public void dispose(IoSession arg0) throws Exception 
        System.out.println("dispose" + arg0.hashCode());    
    

    public void encode(IoSession arg0, Object msg, ProtocolEncoderOutput output)
            throws Exception 

        String s= null;
        if(msg instanceof String)
            s = (String) msg;
        
        if(s!=null)
            CharsetEncoder charsetEncoder = (CharsetEncoder) arg0.getAttribute("encoder");
            if(charsetEncoder ==null)
                charsetEncoder = Charset.defaultCharset().newEncoder();
                arg0.setAttribute("encoder",charsetEncoder);
            
            IoBuffer ioBuffer = IoBuffer.allocate(s.length());
            ioBuffer.setAutoExpand(true);
            ioBuffer.putString(s, charsetEncoder);
            ioBuffer.flip();
            output.write(ioBuffer);

MyHandler跟版本一的一样，这里就不贴代码了。
MyCumulativeDecoder ：

package com.socket.r1;

import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.CumulativeProtocolDecoder;
import org.apache.mina.filter.codec.ProtocolDecoderOutput;

public class MyCumulativeDecoder extends CumulativeProtocolDecoder 

    /**
     * 确认读取完成时return true
     */
    @Override
    protected boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput output)
            throws Exception 
        int startPosition = in.position();
        while(in.hasRemaining())
            byte b = in.get();
            if(b == '\\n')
                int curPosition = in.position();
                int limit = in.limit();
                in.position(startPosition);
                in.limit(curPosition);
                IoBuffer buf = in.slice();
                byte[] bytes = new byte[buf.limit()];
                buf.get(bytes);
                String str = new String(bytes);
                output.write(str);
                in.position(curPosition);
                in.limit(limit);
                return true;
            
        
        //取消此次的读取，将读取位置重置
        in.position(startPosition);
        return false;

MyProtocolCodecFactory ：

package com.socket.r1;

import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFactory;
import org.apache.mina.filter.codec.ProtocolDecoder;
import org.apache.mina.filter.codec.ProtocolEncoder;

public class MyProtocolCodecFactory implements ProtocolCodecFactory 

//  private MyDecoder decoder;
    private MyCumulativeDecoder decoder;
    private MyEncoder encoder;

    public MyProtocolCodecFactory() 
//      decoder = new MyDecoder();
        decoder = new MyCumulativeDecoder();
        encoder = new MyEncoder();
    

    public ProtocolDecoder getDecoder(IoSession arg0) throws Exception 
        return decoder;
    

    public ProtocolEncoder getEncoder(IoSession arg0) throws Exception 
        return encoder;

以上是关于Socket编程及mina框架简单示例的主要内容，如果未能解决你的问题，请参考以下文章

Linux C Socket编程原理及简单实例

大并发量socket 通信框架MINA介绍

长连接神器Mina框架的使用

Mina简单的入门示例

MINA 框架总结整体理解

关于Socket高并发的原理介绍及使用Apache Mina带来线上的问题分析