求一个Java多线程阻塞模式通信的例子

你是要阻塞线程通信还是非阻塞啊？

服务器端

package multithread4;
import java.io.*;
import java.net.*;
import java.util.concurrent.*;

public class EchoServer {
private int port=8000;
private ServerSocket serverSocket;
private ExecutorService executorService; //线程池
private final int POOL_SIZE=4; //单个CPU时线程池中工作线程的数目

private int portForShutdown=8001; //用于监听关闭服务器命令的端口
private ServerSocket serverSocketForShutdown;
private boolean isShutdown=false; //服务器是否已经关闭

private Thread shutdownThread=new Thread(){ //负责关闭服务器的线程
public void start(){
this.setDaemon(true); //设置为守护线程（也称为后台线程）
super.start();
}

public void run(){
while (!isShutdown) {
Socket socketForShutdown=null;
try {
socketForShutdown= serverSocketForShutdown.accept();
BufferedReader br = new BufferedReader(
new InputStreamReader(socketForShutdown.getInputStream()));
String command=br.readLine();
if(command.equals("shutdown")){
long beginTime=System.currentTimeMillis();
socketForShutdown.getOutputStream().write("服务器正在关闭\r\n".getBytes());
isShutdown=true;
//请求关闭线程池
//线程池不再接收新的任务，但是会继续执行完工作队列中现有的任务
executorService.shutdown();

//等待关闭线程池，每次等待的超时时间为30秒
while(!executorService.isTerminated())
executorService.awaitTermination(30,TimeUnit.SECONDS);

serverSocket.close(); //关闭与EchoClient客户通信的ServerSocket
long endTime=System.currentTimeMillis();
socketForShutdown.getOutputStream().write(("服务器已经关闭，"+
"关闭服务器用了"+(endTime-beginTime)+"毫秒\r\n").getBytes());
socketForShutdown.close();
serverSocketForShutdown.close();

}else{
socketForShutdown.getOutputStream().write("错误的命令\r\n".getBytes());
socketForShutdown.close();
}
}catch (Exception e) {
e.printStackTrace();
}
}
}
};

public EchoServer() throws IOException {
serverSocket = new ServerSocket(port);
serverSocket.setSoTimeout(60000); //设定等待客户连接的超过时间为60秒
serverSocketForShutdown = new ServerSocket(portForShutdown);

//创建线程池
executorService= Executors.newFixedThreadPool(
Runtime.getRuntime().availableProcessors() * POOL_SIZE);

shutdownThread.start(); //启动负责关闭服务器的线程
System.out.println("服务器启动");
}

public void service() {
while (!isShutdown) {
Socket socket=null;
try {
socket = serverSocket.accept(); //可能会抛出SocketTimeoutException和SocketException
socket.setSoTimeout(60000); //把等待客户发送数据的超时时间设为60秒
executorService.execute(new Handler(socket)); //可能会抛出RejectedExecutionException
}catch(SocketTimeoutException e){
//不必处理等待客户连接时出现的超时异常
}catch(RejectedExecutionException e){
try{
if(socket!=null)socket.close();
}catch(IOException x){}
return;
}catch(SocketException e) {
//如果是由于在执行serverSocket.accept()方法时，
//ServerSocket被ShutdownThread线程关闭而导致的异常，就退出service()方法
if(e.getMessage().indexOf("socket closed")!=-1)return;
}catch(IOException e) {
e.printStackTrace();
}
}
}

public static void main(String args[])throws IOException {
new EchoServer().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(socketOut,true);
}
private BufferedReader getReader(Socket socket)throws IOException{
InputStream socketIn = socket.getInputStream();
return new BufferedReader(new InputStreamReader(socketIn));
}
public String echo(String msg) {
return "echo:" + msg;
}
public void run(){
try {
System.out.println("New connection accepted " +
socket.getInetAddress() + ":" +socket.getPort());
BufferedReader br =getReader(socket);
PrintWriter pw = getWriter(socket);

String msg = null;
while ((msg = br.readLine()) != null) {
System.out.println(msg);
pw.println(echo(msg));
if (msg.equals("bye"))
break;
}
}catch (IOException e) {
e.printStackTrace();
}finally {
try{
if(socket!=null)socket.close();
}catch (IOException e) {e.printStackTrace();}
}
}
}

客户端

package multithread4;
import java.net.*;
import java.io.*;
public class AdminClient{
public static void main(String args[]){
Socket socket=null;
try{
socket=new Socket("localhost",8001);
//发送关闭命令
OutputStream socketOut=socket.getOutputStream();
socketOut.write("shutdown\r\n".getBytes());

//接收服务器的反馈
BufferedReader br = new BufferedReader(
new InputStreamReader(socket.getInputStream()));
String msg=null;
while((msg=br.readLine())!=null)
System.out.println(msg);
}catch(IOException e){
e.printStackTrace();
}finally{
try{
if(socket!=null)socket.close();
}catch(IOException e){e.printStackTrace();}
}
}
}

程序分Server和Client

服务器端打开侦听的端口，一有客户端连接就创建两个新的线程来负责这个连接

一个负责客户端发送的信息(ClientMsgCollectThread 类)，

另一个负责通过该Socket发送数据（ServerMsgSendThread ）

Server.java代码如下:

/*
* 创建日期 2009-3-7
*
* TODO 要更改此生成的文件的模板，请转至
* 窗口 － 首选项 － Java － 代码样式 － 代码模板
*/
package faue.MutiUser;

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

/**
* 服务器端
*
* @author Faue
*/
public class Server extends ServerSocket {

private static final int SERVER_PORT = 10000;

/**
* 构造方法，用于实现连接的监听
*
* @throws IOException
*/
public Server() throws IOException {
super(SERVER_PORT);

try {
while (true) {
Socket socket = super.accept();

new Thread(new ClientMsgCollectThread(socket), "getAndShow"
+ socket.getPort()).start();
new Thread(new ServerMsgSendThread(socket), "send"
+ socket.getPort()).start();

}
} catch (IOException e) {
e.printStackTrace();
}

}

public static void main(String[] args) throws IOException {
new Server();
}

/**
* 该类用于创建接收客户端发来的信息并显示的线程
*
* @author Faue
* @version 1.0.0
*/
class ClientMsgCollectThread implements Runnable {

private Socket client;

private BufferedReader in;

private StringBuffer inputStringBuffer = new StringBuffer("Hello");

/**
* 得到Socket的输入流
*
* @param s
* @throws IOException
*/
public ClientMsgCollectThread(Socket s) throws IOException {
client = s;

in = new BufferedReader(new InputStreamReader(client
.getInputStream(), "GBK"));
}

public void run() {
try {

while (!client.isClosed()) {
inputStringBuffer.delete(0, inputStringBuffer.length());
inputStringBuffer.append(in.readLine());

System.out.println(getMsg(inputStringBuffer.toString()));
}
} catch (IOException e) {
//e.printStackTrace();
System.out.println(client.toString() + " is closed!");

}
}

/**
* 构造显示的字符串
*
* @param line
* @return
*/
private String getMsg(String line) {
return client.toString() + " says:" + line;
}

}

/**
* 该类用于创建发送数据的线程
*
* @author Faue
* @version 1.0.0
*/
class ServerMsgSendThread implements Runnable {

private Socket client;

private PrintWriter out;

private BufferedReader keyboardInput;

private StringBuffer outputStringBuffer = new StringBuffer("Hello");

/**
* 得到键盘的输入流
*
* @param s
* @throws IOException
*/
public ServerMsgSendThread(Socket s) throws IOException {
client = s;

out = new PrintWriter(client.getOutputStream(), true);
keyboardInput = new BufferedReader(new InputStreamReader(System.in));

}

public void run() {
try {

while (!client.isClosed()) {
outputStringBuffer.delete(0, outputStringBuffer.length());
outputStringBuffer.append(keyboardInput.readLine());

out.println(outputStringBuffer.toString());
}
} catch (IOException e) {
//e.printStackTrace();
System.out.println(client.toString() + " is closed!");

}
}

}

}

客户端：

实现基于IP地址的连接，连接后也创建两个线程来实现信息的发送和接收

/*
* 创建日期 2009-3-7
*
*/
package faue.MutiUser;

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

/**
* 客户端
*
* @author Faue
*/
public class Client {

private Socket mySocket;

/**
* 创建线程的构造方法
*
* @param IP
* @throws IOException
*/
public Client(String IP) throws IOException {

try {
mySocket = new Socket(IP, 10000);
new Thread(new ServerMsgCollectThread(mySocket), "getAndShow"
+ mySocket.getPort()).start();
new Thread(new ClientMsgSendThread(mySocket), "send"
+ mySocket.getPort()).start();

} catch (IOException e) {
//e.printStackTrace();
System.out.println("Server.IP:" + IP
+ " port:10000 can not be Connected");
}
}

public static void main(String[] args) throws IOException {
try {
new Client(args[0]);
} catch (Exception e) {
System.out.println("输入的IP地址错误");
}
}

/**
* 该类用于创建接收服务端发来的信息并显示的线程
*
* @author Faue
* @version 1.0.0
*/
class ServerMsgCollectThread implements Runnable {

private Socket client;

private BufferedReader in;

private StringBuffer inputStringBuffer = new StringBuffer("Hello");

/**
* 得到Socket的输入流
*
* @param s
* @throws IOException
*/
public ServerMsgCollectThread(Socket s) throws IOException {
client = s;

in = new BufferedReader(new InputStreamReader(client
.getInputStream(), "GBK"));
}

public void run() {
try {

while (!client.isClosed()) {
inputStringBuffer.delete(0, inputStringBuffer.length());
inputStringBuffer.append(in.readLine());
System.out.println(getMsg(inputStringBuffer.toString()));
}
} catch (IOException e) {
//e.printStackTrace();
System.out.println(client.toString() + " is closed!");
System.exit(0);
}
}

/**
* 构造输入字符串
*
* @param line
* @return
*/
private String getMsg(String line) {
return client.toString() + " says:" + line;
}

}

/**
* 该类用于创建发送数据的线程
*
* @author Faue
* @version 1.0.0
*/
class ClientMsgSendThread implements Runnable {

private Socket client;

private PrintWriter out;

private BufferedReader keyboardInput;

private StringBuffer outputStringBuffer = new StringBuffer("Hello");

/**
* 得到键盘的输入流
*
* @param s
* @throws IOException
*/
public ClientMsgSendThread(Socket s) throws IOException {
client = s;

out = new PrintWriter(client.getOutputStream(), true);
keyboardInput = new BufferedReader(new InputStreamReader(System.in));

}

public void run() {
try {

while (!client.isClosed()) {
outputStringBuffer.delete(0, outputStringBuffer.length());
outputStringBuffer.append(keyboardInput.readLine());

out.println(outputStringBuffer.toString());
}
out.println("--- See you, bye! ---");
} catch (IOException e) {
//e.printStackTrace();
System.out.println(client.toString() + " is closed!");
System.exit(0);
}
}

}

}