package main.java.com.founder.study.javaio.aio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* Created by on 2020/7/2.
*/
public class ChatServer {
private static final int DEFAULT_PORT = 8888;
private static final String QUIT = "quit";
private static final int BUFFER = 1024;
final String LOCALHOST = "localhost";
private static final int THREAPOOL = 8;
private AsynchronousChannelGroup channelGroup;
private List<ClientHandler> connectedClients;
private AsynchronousServerSocketChannel serverChannel;
private Charset charset = Charset.forName("UTF-8");
private int port;
public ChatServer(){
this(DEFAULT_PORT);
}
public ChatServer(int port){
this.port = port;
this.connectedClients = new ArrayList<>();
}
public void start(){
ExecutorService executorService = Executors.newFixedThreadPool(THREAPOOL);
try {
channelGroup = AsynchronousChannelGroup.withThreadPool(executorService);
serverChannel = AsynchronousServerSocketChannel.open(channelGroup);
serverChannel.bind(new InetSocketAddress(LOCALHOST,DEFAULT_PORT));
System.out.println("启动服务器监听端口"+port);
while(true){
serverChannel.accept(null,new AcceptHandler());
System.in.read();
}
} catch (IOException e) {
e.printStackTrace();
}
}
private class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel,Object>{
@Override
public void completed(AsynchronousSocketChannel clientChannel, Object attachment) {
if(serverChannel.isOpen()){
serverChannel.accept(null,this);
}
if(clientChannel.isOpen() && clientChannel != null){
ClientHandler handler = new ClientHandler(clientChannel);
ByteBuffer buffer = ByteBuffer.allocate(BUFFER);
//TODO 将新用户添加到在线用户列表
addClient(handler);
clientChannel.read(buffer,buffer,handler);
}
}
@Override
public void failed(Throwable exc, Object attachment) {
System.out.println("连接失败"+exc);
}
}
private void addClient(ClientHandler handler) {
connectedClients.add(handler);
System.out.println("加入连接");
}
private void removeClient(ClientHandler clientHandler) {
connectedClients.remove(clientHandler);
System.out.println("移除连接");
}
private class ClientHandler implements CompletionHandler<Integer,Object>{
private AsynchronousSocketChannel clientChannel;
@Override
public void completed(Integer result, Object attachment) {
ByteBuffer buffer = (ByteBuffer) attachment;
if(buffer != null){
if(result < 0){
//客户端异常
//TODO 将客户移除在线客户列表
}else{
buffer.flip();
String msg = receive(buffer);
System.out.println(msg);
forwardMsg(clientChannel,msg);
buffer.clear();
if(QUIT.equals(readeyToQuit(msg))){
removeClient(this);
}else{
clientChannel.read(buffer,buffer,this);
}
}
}
}
@Override
public void failed(Throwable exc, Object attachment) {
}
public ClientHandler(AsynchronousSocketChannel clientChannel){
this. clientChannel =clientChannel;
}
}
private void forwardMsg(AsynchronousSocketChannel clientChannel, String msg) {
for(ClientHandler handler:connectedClients){
if(!handler.equals(handler.clientChannel)){
try {
InetSocketAddress remoteAddress = (InetSocketAddress)handler.clientChannel.getRemoteAddress();
ByteBuffer buffer = charset.encode(remoteAddress.getPort() + "");
handler.clientChannel.write(buffer,null,handler);
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
private String receive(ByteBuffer buffer) {
CharBuffer charBuffer = charset.decode(buffer);
return String.valueOf(charBuffer);
}
public boolean readeyToQuit(String msg){
return QUIT.equals(msg);
}
}
package main.java.com.founder.study.javaio.aio;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
/**
* Created by on 2020/7/3.
*/
public class ChatClient {
private static final int DEFAULT_PORT = 8888;
private static final String QUIT = "quit";
private static final int BUFFER = 1024;
private static final String LOCALHOST = "localhost";
private Charset charset = Charset.forName("UTF-8");
private AsynchronousSocketChannel clientChannel;
private String host;
private int port;
public ChatClient(){
this(LOCALHOST,DEFAULT_PORT);
}
public ChatClient(String host,int port){
this.host = host;
this.port = port;
}
public boolean readeyToQuit(String msg){
return QUIT.equals(msg);
}
public void start(){
try {
clientChannel= AsynchronousSocketChannel.open();
Future<Void> future = clientChannel.connect(new InetSocketAddress(host, port));
future.get();
//处理用户的输入
new Thread(new UserInputHandler(this)).start();
ByteBuffer buffer = ByteBuffer.allocate(BUFFER);
while (true){
Future<Integer> readResult = clientChannel.read(buffer);
Integer result = readResult.get();
if(result<0){
//服务异常
System.out.println(result);
//close
System.exit(1);
}else{
buffer.flip();
CharBuffer charBuffer = charset.decode(buffer);
String s = String.valueOf(charBuffer);
System.out.println(s);
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
public void send(String msg){
if(msg.isEmpty()){
return;
}
ByteBuffer buffer = charset.encode(msg);
Future<Integer> future = clientChannel.write(buffer);
try {
future.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
package main.java.com.founder.study.javaio.aio;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
/**
* Created by on 2020/6/28.
*/
public class UserInputHandler implements Runnable{
final String QUIT = "quit" ;
private ChatClient chatClient;
public UserInputHandler(ChatClient chatClient ){
this.chatClient = chatClient ;
}
@Override
public void run() {
BufferedReader consoleReader = new BufferedReader(new InputStreamReader(System.in));
while (true){
//先发给服务端
try {
chatClient.send(consoleReader.readLine());
chatClient.readeyToQuit(consoleReader.readLine());
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
