基于异步IO的socket通信程序

package server; import java.nio.channels.SelectionKey; import java.nio.channels.SocketChannel; import java.util.Collection; import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Scanner; import java.util.Set; import common.Message; import common.MessageType; import common.SocketResource; import common.SocketUtilForMessage; import common.SystemMessage; import common.SystemOrderType; import common.TextMessage; import common.UserInfoMessage; public class ServerSocketForMessage extends SocketInServer { private SocketUtilForMessage util; private Map<String,Integer> userMap; @Override public boolean startSocket(int port) { util = new SocketUtilForMessage(); userMap = new HashMap<String, Integer>(); return super.startSocket(port); } @Override public void doRead(SelectionKey key) { SocketChannel socketChannel = (SocketChannel)key.channel(); Message message = util.readMessage(socketChannel); if(message == null){ //System.out.println(" 关闭连接通道"); this.closeChannel(socketChannel); return; } //遍历所有已连接socket，注册写事件 /*this.stringForSending = str; Collection<SocketChannel> collection = this.getSocketChannelMap().values(); Iterator<SocketChannel> iterator = collection.iterator(); while(iterator.hasNext()){ SocketChannel sChannel = iterator.next(); //iterator.remove(); if(sChannel == socketChannel) continue; sChannel.keyFor(this.getSelector()).interestOps(SelectionKey.OP_WRITE); }*/ handleMessage(message,socketChannel); System.out.println(" 信息接受完毕"); } @Override public void doWrite(SelectionKey key) { SocketChannel socketChannel = (SocketChannel)key.channel(); if(key.attachment() != null){ Message message = (Message)key.attachment(); Integer size = util.writeMessage(socketChannel, message); if(size == null){ //System.out.println(" 关闭连接通道"); this.closeChannel(socketChannel); return; } System.out.println(); key.attach(null); } List<Message> messageList = SocketResource.getSocketResource(socketChannel.hashCode()).getMessageList(); Iterator<Message> iterator = messageList.iterator(); while(iterator.hasNext()){ Message message = iterator.next(); iterator.remove(); Integer size = util.writeMessage(socketChannel, message); if(size == null){ //System.out.println(" 关闭连接通道"); this.closeChannel(socketChannel); return; } System.out.println(); } key.interestOps(SelectionKey.OP_READ); System.out.println(" 信息发送完毕"); } public void handleMessage(Message message,SocketChannel socketChannel){ if(message.getMessageType() == MessageType.UserInfoMessage){ UserInfoMessage userInfoMessage = (UserInfoMessage)message; System.out.print("用户名："+userInfoMessage.getAddresser()+",密码："+userInfoMessage.getPassword() +",请求验证..."); if(checkUser(userInfoMessage.getAddresser(),userInfoMessage.getPassword())){ System.out.println(" 请求通过"); //添加用户到在线用户表，创建资源区，注册写事件，附加系统信息 userMap.put(userInfoMessage.getAddresser(), socketChannel.hashCode()); SocketResource socketResource = SocketResource.newResourceInstance(socketChannel); socketResource.setUsername(userInfoMessage.getAddresser()); socketResource.setPassword(userInfoMessage.getPassword()); socketChannel.keyFor(this.getSelector()).interestOps(SelectionKey.OP_WRITE).attach(new SystemMessage(SystemOrderType.ConnectionPermitted)); }else{ //注册写事件，附加系统信息 System.out.println(" 拒绝请求"); //this.getSocketChannelMap().remove(socketChannel); //this.closeChannel(socketChannel); socketChannel.keyFor(this.getSelector()).interestOps(SelectionKey.OP_WRITE).attach(new SystemMessage(SystemOrderType.ConnectionRefused)); } }else if(message.getMessageType() == MessageType.TextMessage){ TextMessage textMessage = (TextMessage)message; System.out.print(" 消息内容为\""+textMessage.getTextContent()+"\" "); if(textMessage.isPrivate()){ String recipient = textMessage.getRecipient(); if(userMap.get(recipient) != null){ Integer socketChannelId = userMap.get(recipient); SocketResource socketResource = SocketResource.getSocketResource(socketChannelId); socketResource.getSocketChannel().keyFor(this.getSelector()).interestOps(SelectionKey.OP_WRITE); socketResource.getMessageList().add(textMessage); }else{ this.persistMessage(textMessage); } }else{ //getUserSet(textMessage.getRecipient()); //临时测试，所有在线用户充当组员 Collection<Integer> keys = userMap.values(); Iterator<Integer> iterator = keys.iterator(); while(iterator.hasNext()){ Integer key = iterator.next(); SocketResource socketResource = SocketResource.getSocketResource(key); socketResource.getSocketChannel().keyFor(this.getSelector()).interestOps(SelectionKey.OP_WRITE); socketResource.getMessageList().add(textMessage); } } }else{ //待开发消息类型 } } public boolean checkUser(String username,String password){ //用户登录验证，待开发 return true; } public void persistMessage(Message message){ //持久化消息,待开发 } public Set<String> getUserSet(String userGroup){ //根据组名拿到组员集，待开发 return null; } public static void main(String[] args){ final ServerSocketForMessage server = new ServerSocketForMessage(); if(!server.startSocket(7788)) return; new Thread(new Runnable() { @Override public void run() { server.handleOPS(); } }).start(); Scanner scanner = new Scanner(System.in); while(scanner.hasNext()){ String str = scanner.next() + "<系统消息>"; SystemMessage systemMessage = new SystemMessage(str); //Collection<SocketChannel> collection = server.getSocketChannelMap().values(); Collection<Integer> keys = server.userMap.values(); Iterator<Integer> iterator = keys.iterator(); while(iterator.hasNext()){ Integer key = iterator.next(); SocketResource socketResource = SocketResource.getSocketResource(key); systemMessage.setRecipient(socketResource.getUsername()); socketResource.getSocketChannel().keyFor(server.getSelector()).interestOps(SelectionKey.OP_WRITE); socketResource.getMessageList().add(systemMessage); } } } }