package GroupTalk;
import java.io.*;
import java.net.*;
import java.util.*;
import java.util.concurrent.*;
public class GroupServer {
private static Set Members; // 在线æˆå‘˜é›†åˆ
private int port = 8013;
private ServerSocket serverSocket;
private ExecutorService executorService; //线程æ±
private final int POOL_SIZE = 15; //å•ä¸ªCPUæ—¶çº¿ç¨‹æ± ä¸å·¥ä½œçº¿ç¨‹çš„æ•°ç›®
public GroupServer() throws IOException {
Members = new HashSet();
serverSocket = new ServerSocket(port);//å¯åŠ¨æœåŠ¡å™¨
//创建线程æ±
//Runtimeçš„availableProcessors()方法返回当å‰ç³»ç»Ÿçš„CPUçš„æ•°ç›®
//系统的CPUè¶Šå¤šï¼Œçº¿ç¨‹æ± ä¸å·¥ä½œçº¿ç¨‹çš„数目也越多
executorService = Executors.newFixedThreadPool(
Runtime.getRuntime().availableProcessors() * POOL_SIZE);
System.out.println("多用户æœåŠ¡å™¨å¯åŠ¨");
}
//在GroupServer主类ä¸å®šä¹‰ç¾¤ç»„消æ¯å‘é€æ–¹æ³•ã€‚
public static void sendToAllMembers(String msg) throws IOException {
PrintWriter pw;
Socket tempSocket;
OutputStream socketOut;
Iterator it = Members.iterator();
while (it.hasNext()) {//é历在线æˆå‘˜set集åˆ.
tempSocket = (Socket) it.next(); //å–出一个æˆå‘˜
try{
socketOut = tempSocket.getOutputStream();//得到输出æµ
}catch(SocketException e){
removeMember(tempSocket);
continue;
}
//装饰æˆå—符æµ
pw = new PrintWriter(new OutputStreamWriter(socketOut, "GB2312"), true);
pw.println(msg);//å‘é€èŠå¤©ä¿¡æ¯ç»™æˆå‘˜
}
System.out.println(Members.size());
}//end while群组å‘é€ç»“æŸã€‚
//在GroupServer主类ä¸å®šä¹‰ç»„员退出方法。
public static void removeMember(Socket socket) {
Members.remove(socket);//åˆ é™¤ä¸€ä¸ªæˆå‘˜
}
public void service() {
while (true) {
Socket socket = null;
try {
socket = serverSocket.accept(); //监å¬å®¢æˆ·è¯·æ±‚, 阻塞è¯å¥.
//接å—一个客户请求,ä»Žçº¿ç¨‹æ± ä¸æ‹¿å‡ºä¸€ä¸ªçº¿ç¨‹ä¸“门处ç†è¯¥å®¢æˆ·.
GroupServer.Members.add(socket); // å°†æ–°æ·»åŠ çš„ç”¨æˆ·æ·»åŠ è¿›ç”¨æˆ·é›†åˆ
executorService.execute(new Handler(socket));
} catch (IOException e) {
}
}
}
public static void main(String args[]) throws IOException {
new GroupServer().service();
}
}
//定义内部类，实现线程接å£
class Handler implements Runnable {
private Socket socket;
public Handler(Socket socket) {
this.socket = socket;
}
private PrintWriter getWriter(Socket socket) throws IOException {
OutputStream socketOut = socket.getOutputStream();
return new PrintWriter(new OutputStreamWriter(socketOut, "GB2312"), true);
}
private BufferedReader getReader(Socket socket) throws IOException {
InputStream socketIn = socket.getInputStream();
return new BufferedReader(new InputStreamReader(socketIn, "GB2312"));
}
public void run() {//覆盖线程体
try {
System.out.println("New connection accepted " + socket.getInetAddress());
BufferedReader br = getReader(socket);//å—节装饰æˆå—符æµ
//PrintWriter pw = getWriter(socket); //å—节装饰æˆå—符æµ
String msg = null;
while ((msg = br.readLine()) != null) {
//pw.println("From ThreadServer:" + msg);//send to client.
GroupServer.sendToAllMembers(msg); // send to all menbers
if (msg.contains("bye".subSequence(0, 2))) {
System.out.println(socket.getInetAddress() + ":" + "Exit");
break;
}
}
} catch (IOException e) {
e.printStackTrace();
} finally {
try {
if (socket != null) {
socket.close();
}
} catch (IOException e) {
}
}
}
}