NIO之socket交互。

参考： https://www.cnblogs.com/yanghuahui/p/3686054.html

https://www.jianshu.com/p/c26a25feb77e

参考上面的代码，写了份demo。雷同率高达98.888888888888~%

本来想加个服务端断掉，客户端尝试重连的功能。发现难度有点大。如有解决方案，希望可以告知，感激不尽。

测试代码

IChat.java

package com.boot.demo.test.io.socket.chat;

/**
 * @author braska
 * @date 2020/3/23
 **/
public interface IChat {
    void run() throws Exception;
}

　

AbstractChat.java

package com.boot.demo.test.io.socket.chat;

import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;

/**
 * @author braska
 * @date 2020/3/23
 **/
public abstract class AbstractChat implements IChat {
    protected final String USER_CONTENT_SPLIT = "#@#";
    protected final String USER_EXISTS_ERROR = "system message: user exist, please change a name";
    protected final String WELCOME = "welcome %s to chat room! Online numbers: %s";

    protected Charset charset = Charset.forName("UTF-8");

    protected int port;
    protected Selector selector;
    protected SelectionKey currentKey;

    @Override
    public void run() throws Exception {
        while (this.selector.select() > 0) {
            for (Iterator<SelectionKey> iter = this.selector.selectedKeys().iterator(); iter.hasNext(); ) {
                currentKey = iter.next();
                iter.remove();
                keyHandler(currentKey);
            }
        }
    }

    protected abstract void keyHandler(SelectionKey key) throws Exception;

    protected SocketChannel getChannel() {
        return (SocketChannel) currentKey.channel();
    }

    protected String content() throws Exception {
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        StringBuilder builder = new StringBuilder(256);
        while (getChannel().read(byteBuffer) > 0) {
            byteBuffer.flip();
            builder.append(charset.decode(byteBuffer));
            byteBuffer.clear();
        }
        return builder.toString();
    }
}

　　

ClientServer.java

package com.boot.demo.test.io.socket.chat;

import com.google.common.base.Strings;
import com.google.common.collect.Maps;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.*;
import java.util.Map;

/**
 * @author braska
 * @date 2020/3/23
 **/
public class ChatServer extends AbstractChat {
    private int port;
    private final Map<String, SocketChannel> users;

    public ChatServer(int port) throws Exception {
        this.users = Maps.newConcurrentMap();
        this.port = port;
        ServerSocketChannel channel = ServerSocketChannel.open();
        channel.socket().bind(new InetSocketAddress(this.port));
        channel.configureBlocking(false);
        this.selector = Selector.open();
        channel.register(selector, SelectionKey.OP_ACCEPT);
    }

    @Override
    public void keyHandler(SelectionKey key) {
        try {
            if (key.isAcceptable()) {    // 客户端连入
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
                sc.register(this.selector, SelectionKey.OP_READ);
                key.interestOps(SelectionKey.OP_ACCEPT);
                System.out.println("Server is listening from client :" + sc.getRemoteAddress());
                sc.write(charset.encode("Please input your name:"));
            } else if (key.isReadable()) {  // 接收客户端消息
                SocketChannel channel = getChannel();
                key.interestOps(SelectionKey.OP_READ);
                String content = content();
                if (!Strings.isNullOrEmpty(content)) {
                    String[] nameAndContent = content.split(USER_CONTENT_SPLIT);
                    if (nameAndContent != null && nameAndContent.length == 1) {
                        if (users.containsKey(nameAndContent[0]) && users.get(nameAndContent[0]).isConnected()) {
                            channel.write(charset.encode(USER_EXISTS_ERROR));
                        } else {
                            users.put(nameAndContent[0], channel);
                            dispatch(String.format(WELCOME, nameAndContent[0], countUsers()), null);
                        }
                    } else {
                        content = nameAndContent[0] + " said: " +
                                content.replace(String.format("%s%s", nameAndContent[0], USER_CONTENT_SPLIT), "");
                        if (users.containsKey(nameAndContent[0])) {
                            dispatch(content, channel);
                        }
                    }
                }
            }
        } catch (Exception e) {
            if (key != null) {
                key.cancel();
                if (key.channel() != null) {
                    try {
                        for (Map.Entry<String, SocketChannel> entry : users.entrySet()) {
                            if (entry.getValue().equals(key.channel())) {
                                dispatch(entry.getKey() + " has gone.", key.channel());
                                break;
                            }
                        }
                        key.channel().close();
                    } catch (IOException e1) {
                        e1.printStackTrace();
                    }
                }
            }
        }
    }

    private long countUsers() {
        return this.selector.keys().stream().filter(key -> key.channel() instanceof SocketChannel).count();
    }

    private void dispatch(String content, SelectableChannel channel) {
        selector.keys().stream()
                .filter(key -> key.channel() instanceof SocketChannel)
                .filter(key -> key.channel() != channel)
                .forEach(key -> {
                    SocketChannel c = (SocketChannel) key.channel();
                    try {
                        c.write(charset.encode(content));
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
    }

    public static void main(String[] args) throws Exception {
        new ChatServer(3000).run();
    }
}

　　

ChatClient.java

package com.boot.demo.test.io.socket.chat;

import org.apache.logging.log4j.util.Strings;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Scanner;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author braska
 * @date 2020/3/23
 **/
public class ChatClient extends AbstractChat {
    private String username;
    private Integer retry = 20;
    private int interval = 1000;
    private AtomicInteger count = new AtomicInteger(0);
    private Semaphore semaphore = new Semaphore(1);

    public ChatClient(int port) throws Exception {
        this.port = port;
        connect();
    }

    public void setRetry(Integer retry) {
        this.retry = retry;
    }

    public void setInteval(int inteval) {
        this.interval = inteval;
    }

    private void connect() throws Exception {
        this.selector = Selector.open();
        SocketChannel channel = SocketChannel.open();
        channel.configureBlocking(false);   
        channel.connect(new InetSocketAddress(this.port));
        channel.register(selector, SelectionKey.OP_CONNECT);

        if (semaphore.tryAcquire() && count.get() != 0 && count.get() <= retry) {
            System.out.println("try to connect chat server " + count.get() + "time");
            connect();
        }
    }

    @Override
    public void keyHandler(SelectionKey key) {
        try {
            SocketChannel channel = getChannel();
            if (key.isValid() && key.isConnectable()) {
                if (channel.finishConnect()) {
                    count.set(0);
                    username = "";
                }
                channel.register(this.selector, SelectionKey.OP_READ);
                new Thread(() -> {
                    Scanner scanner = new Scanner(System.in);
                    try {
                        while (scanner.hasNextLine()) {
                            String content = scanner.nextLine();
                            if ("".equals(content)) continue;
                            if (!Strings.isNotBlank(username)) {
                                username = content;
                                content += USER_CONTENT_SPLIT;
                            } else {
                                content = username + USER_CONTENT_SPLIT + content;
                            }
                            try {
                                ChatClient.this.getChannel().write(charset.encode(content));
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                    } finally {
                        scanner.close();
                    }
                }).start();
            } else if (key.isReadable()) {
                key.interestOps(SelectionKey.OP_READ);
                String content = content();
                if (content.equals(USER_EXISTS_ERROR)) {
                    username = "";
                }
                System.out.println(content);
            }
        } catch (Exception e) {
            if (key != null) {
                key.cancel();
                if (key.channel() != null) {
                    try {
                        key.channel().close();
                        // todo 调用连接方法，目前有问题注释掉。
/*                        semaphore.release();
                        count.incrementAndGet();
                        connect();
                        TimeUnit.MILLISECONDS.sleep(interval);*/
                    } catch (Exception e1) {
                        e1.printStackTrace();
                    }
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        new ChatClient(3000).run();
    }
}

　　

重连方法也不是不可用，就是有大bug：服务端启动以后，控制台日志会打印出两条客户端连进来的日志。在线人数就翻倍了。

因为有问题，所以注释掉了。

查看别人的客户端重连代码，发现根本不可用。可能是我应用的场景不对。

@Test
    public void test_finishConnect_connect4() throws Exception {
        SocketChannel socketChannel = SocketChannel.open();
        long begin = 0;
        long end = 0;
        try {
            TimeUnit.SECONDS.sleep(1);
            socketChannel.configureBlocking(false);
            boolean result = socketChannel.connect(new InetSocketAddress(3000));
            if(!result){
                while (!socketChannel.finishConnect()){
                    System.out.println("一直在尝试连接");
                    TimeUnit.SECONDS.sleep(1);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }