13518219792
用异步输入输出流编写Socket进程通信程序
创新互联专业为企业提供仪陇网站建设、仪陇做网站、仪陇网站设计、仪陇网站制作等企业网站建设、网页设计与制作、仪陇企业网站模板建站服务，10余年仪陇做网站经验，不只是建网站，更提供有价值的思路和整体网络服务。
在Merlin中加入了用于实现异步输入输出机制的应用程序接口包：java.nio（新的输入输出包，定义了很多基本类型缓冲(Buffer)），java.nio.channels(通道及选择器等，用于异步输入输出)，java.nio.charset（字符的编码解码）。通道(Channel)首先在选择器(Selector)中注册自己感兴趣的事件，当相应的事件发生时，选择器便通过选择键(SelectionKey)通知已注册的通道。然后通道将需要处理的信息，通过缓冲（Buffer）打包，编码/解码,完成输入输出控制。
通道介绍：
这里主要介绍ServerSocketChannel和 SocketChannel.它们都是可选择的(selectable)通道，分别可以工作在同步和异步两种方式下（注意，这里的可选择不是指可以选择两种工作方式，而是指可以有选择的注册自己感兴趣的事件）。可以用channel.configureBlocking(Boolean )来设置其工作方式。与以前版本的API相比较，ServerSocketChannel就相当于ServerSocket(ServerSocketChannel封装了ServerSocket),而SocketChannel就相当于Socket（SocketChannel封装了Socket）。当通道工作在同步方式时，编程方法与以前的基本相似，这里主要介绍异步工作方式。
所谓异步输入输出机制，是指在进行输入输出处理时，不必等到输入输出处理完毕才返回。所以异步的同义语是非阻塞（None Blocking）。在服务器端，ServerSocketChannel通过静态函数open()返回一个实例serverChl。然后该通道调用serverChl.socket().bind()绑定到服务器某端口，并调用register（Selector sel, SelectionKey.OP_ACCEPT）注册OP_ACCEPT事件到一个选择器中（ServerSocketChannel只可以注册OP_ACCEPT事件）。当有客户请求连接时，选择器就会通知该通道有客户连接请求，就可以进行相应的输入输出控制了；在客户端，clientChl实例注册自己感兴趣的事件后（可以是OP_CONNECT,OP_READ,OP_WRITE的组合），调用clientChl.connect(InetSocketAddress )连接服务器然后进行相应处理。注意，这里的连接是异步的，即会立即返回而继续执行后面的代码。
选择器和选择键介绍：
选择器（Selector）的作用是：将通道感兴趣的事件放入队列中，而不是马上提交给应用程序，等已注册的通道自己来请求处理这些事件。换句话说，就是选择器将会随时报告已经准备好了的通道，而且是按照先进先出的顺序。那么，选择器是通过什么来报告的呢？选择键(SelectionKey)。选择键的作用就是表明哪个通道已经做好了准备，准备干什么。你也许马上会想到，那一定是已注册的通道感兴趣的事件。不错，例如对于服务器端serverChl来说，可以调用key.isAcceptable()来通知serverChl有客户端连接请求。相应的函数还有：SelectionKey.isReadable(),SelectionKey.isWritable()。一般的，在一个循环中轮询感兴趣的事件（具体可参照下面的代码）。如果选择器中尚无通道已注册事件发生，调用Selector.select()将阻塞，直到有事件发生为止。另外，可以调用selectNow()或者select(long timeout)。前者立即返回，没有事件时返回0值；后者等待timeout时间后返回。一个选择器最多可以同时被63个通道一起注册使用。
应用实例：
下面是用异步输入输出机制实现的客户/服务器实例程序――程序清单1（限于篇幅，只给出了服务器端实现，读者可以参照着实现客户端代码）：
程序类图
public class NBlockingServer {
int port = 8000;
int BUFFERSIZE = 1024;
Selector selector = null;
ServerSocketChannel serverChannel = null;
HashMap clientChannelMap = null;//用来存放每一个客户连接对应的套接字和通道
public NBlockingServer( int port ) {
this.clientChannelMap = new HashMap();
this.port = port;
}
public void initialize() throws IOException {
//初始化，分别实例化一个选择器，一个服务器端可选择通道
this.selector = Selector.open();
this.serverChannel = ServerSocketChannel.open();
this.serverChannel.configureBlocking(false);
InetAddress localhost = InetAddress.getLocalHost();
InetSocketAddress isa = new InetSocketAddress(localhost, this.port );
this.serverChannel.socket().bind(isa);//将该套接字绑定到服务器某一可用端口
}
//结束时释放资源
public void finalize() throws IOException {
this.serverChannel.close();
this.selector.close();
}
//将读入字节缓冲的信息解码
public String decode( ByteBuffer byteBuffer ) throws
CharacterCodingException {
Charset charset = Charset.forName( "ISO-8859-1" );
CharsetDecoder decoder = charset.newDecoder();
CharBuffer charBuffer = decoder.decode( byteBuffer );
String result = charBuffer.toString();
return result;
}
//监听端口，当通道准备好时进行相应操作
public void portListening() throws IOException, InterruptedException {
//服务器端通道注册OP_ACCEPT事件
SelectionKey acceptKey =this.serverChannel.register( this.selector,
SelectionKey.OP_ACCEPT );
//当有已注册的事件发生时,select()返回值将大于0
while (acceptKey.selector().select() 0 ) {
System.out.println("event happened");
//取得所有已经准备好的所有选择键
Set readyKeys = this.selector.selectedKeys();
//使用迭代器对选择键进行轮询
Iterator i = readyKeys.iterator();
while (i
else if ( key.isReadable() ) {//如果是通道读准备好事件
System.out.println("Readable");
//取得选择键对应的通道和套接字
SelectableChannel nextReady =
(SelectableChannel) key.channel();
Socket socket = (Socket) key.attachment();
//处理该事件，处理方法已封装在类ClientChInstance中
this.readFromChannel( socket.getChannel(),
(ClientChInstance)
this.clientChannelMap.get( socket ) );
}
else if ( key.isWritable() ) {//如果是通道写准备好事件
System.out.println("writeable");
//取得套接字后处理，方法同上
Socket socket = (Socket) key.attachment();
SocketChannel channel = (SocketChannel)
socket.getChannel();
this.writeToChannel( channel,"This is from server!");
}
}
}
}
//对通道的写操作
public void writeToChannel( SocketChannel channel, String message )
throws IOException {
ByteBuffer buf = ByteBuffer.wrap( message.getBytes() );
int nbytes = channel.write( buf );
}
//对通道的读操作
public void readFromChannel( SocketChannel channel, ClientChInstance clientInstance )
throws IOException, InterruptedException {
ByteBuffer byteBuffer = ByteBuffer.allocate( BUFFERSIZE );
int nbytes = channel.read( byteBuffer );
byteBuffer.flip();
String result = this.decode( byteBuffer );
//当客户端发出”@exit”退出命令时，关闭其通道
if ( result.indexOf( "@exit" ) = 0 ) {
channel.close();
}
else {
clientInstance.append( result.toString() );
//读入一行完毕，执行相应操作
if ( result.indexOf( "\n" ) = 0 ){
System.out.println("client input"+result);
clientInstance.execute();
}
}
}
//该类封装了怎样对客户端的通道进行操作，具体实现可以通过重载execute()方法
public class ClientChInstance {
SocketChannel channel;
StringBuffer buffer=new StringBuffer();
public ClientChInstance( SocketChannel channel ) {
this.channel = channel;
}
public void execute() throws IOException {
String message = "This is response after reading from channel!";
writeToChannel( this.channel, message );
buffer = new StringBuffer();
}
//当一行没有结束时，将当前字窜置于缓冲尾
public void append( String values ) {
buffer.append( values );
}
}
//主程序
public static void main( String[] args ) {
NBlockingServer nbServer = new NBlockingServer(8000);
try {
nbServer.initialize();
} catch ( Exception e ) {
e.printStackTrace();
System.exit( -1 );
}
try {
nbServer.portListening();
}
catch ( Exception e ) {
e.printStackTrace();
}
}
}
程序清单1
小结：
从以上程序段可以看出，服务器端没有引入多余线程就完成了多客户的客户/服务器模式。该程序中使用了回调模式(CALLBACK)。需要注意的是，请不要将原来的输入输出包与新加入的输入输出包混用，因为出于一些原因的考虑，这两个包并不兼容。即使用通道时请使用缓冲完成输入输出控制。该程序在Windows2000,J2SE1.4下，用telnet测试成功。
Thread t=new Thread(){
public void run(){
//保存信息操作
}
}
t.start();
//同时做别的事情.
通常同步意味着一个任务的某个处理过程会对多个线程在用串行化处理，而异步则意味着某个处理过程可以允许多个线程同时处理。异步通常代表着更好的性能，因为它很大程度上依赖于缓冲，是典型的使用空间换时间的做法，例如在计算机当中，高速缓存作为cpu和磁盘io之间的缓冲地带协调cpu高速计算能力和磁盘的低速读写能力。
（1）：重新启动一个java程序就启动了一个进程
可以用操作系统命令行启动 Runtime.getRuntime().exec("java -classpath . XXX");
（2）：可不可以在接收消息的模块中的addtolist函数中添加一个专门的处理函数，函数执行时先向list中添加消息，然后探测当前有没有处理线程，如果没有，则启动线程。
（3）：想省点工作，可以用BlockingQueue来代替list，这样线程等待和唤醒不用写代码实现了，如果非要用list，那么就做好同步
list的小例子：
Java codeclass MessageConsumer extends Thead { private ListYourMessageType list; private boolean running = true; public MessageConsumer(ListYourMessageType list) {this.list = list;} public void run() { while (running) { YourMessageType msg = null; try { synchronized(list) { while (list.size() == 0) { list.wait(); } msg = list.remove(0); list.notiryAll(); } } catch (Exception e) { e.printStackTrace(); } if (msg == null) continue; //System.out.println(msg); //print message } }}//调用sampleclass ShareModule { ListYourMessageType list = new ArrayListYourMessageType(); ...}public class Main { public static void main(String[] args) { ShareMudule sm; //so on ... Thread t = new MessageConsumer(sm.list); t.start(); ... }}
1. 使用wait和notify方法
这个方法其实是利用了锁机制，直接贴代码:
public class Demo1 extends BaseDemo{ private final Object lock = new Object(); @Override public void callback(long response) { System.out.println("得到结果"); System.out.println(response); System.out.println("调用结束"); synchronized (lock) { lock.notifyAll(); } } public static void main(String[] args) { Demo1 demo1 = new Demo1(); demo1.call(); synchronized (demo1.lock){ try { demo1.lock.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } System.out.println("主线程内容"); } }
可以看到在发起调用后，主线程利用wait进行阻塞，等待回调中调用notify或者notifyAll方法来进行唤醒。注意，和大家认知的一样，这里wait和notify都是需要先获得对象的锁的。在主线程中最后我们打印了一个内容，这也是用来验证实验结果的，如果没有wait和notify，主线程内容会紧随调用内容立刻打印;而像我们上面的代码，主线程内容会一直等待回调函数调用结束才会进行打印。
没有使用同步操作的情况下，打印结果:发起调用 调用返回 主线程内容 得到结果 1 调用结束
而使用了同步操作后:
发起调用 调用返回 得到结果 9 调用结束 主线程内容2. 使用条件锁
和方法一的原理类似:
public class Demo2 extends BaseDemo { private final Lock lock = new ReentrantLock(); private final Condition con = lock.newCondition(); @Override public void callback(long response) { System.out.println("得到结果"); System.out.println(response); System.out.println("调用结束"); lock.lock(); try { con.signal(); }finally { lock.unlock(); } } public static void main(String[] args) { Demo2 demo2 = new Demo2(); demo2.call(); demo2.lock.lock(); try { demo2.con.await(); } catch (InterruptedException e) { e.printStackTrace(); }finally { demo2.lock.unlock(); } System.out.println("主线程内容"); } }
基本上和方法一没什么区别，只是这里使用了条件锁，两者的锁机制有所不同。
在整个思路上要调整一下
1、会有很多线程给一个队列上添加任务
2、有一个或者多个线程逐个执行队列的任务
考虑一下几点：
1、没有任务时，队列执行线程处于等待状态
2、添加任务时，激活队列执行线程，全部run起来，首先抢到任务的执行，其他全部wait
给个小例子吧
package org;
import java.util.LinkedList;
import java.util.List;
public class Queues {
public static ListTask queue = new LinkedListTask();
/**
* 假如 参数o 为任务
* @param o
*/
public static void add (Task t){
synchronized (Queues.queue) {
Queues.queue.add(t); //添加任务
Queues.queue.notifyAll();//激活该队列对应的执行线程，全部Run起来
}
}
static class Task{
public void test(){
System.out.println("我被执行了");
}
}
}
package org;
import java.util.List;
public class Exec implements Runnable{
@Override
public void run() {
while(true){
synchronized (Queues.queue) {
while(Queues.queue.isEmpty()){ //
try {
Queues.queue.wait(); //队列为空时，使线程处于等待状态
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("wait...");
}
Queues.Task t= Queues.queue.remove(0); //得到第一个
t.test(); //执行该任务
System.out.println("end");
}
}
}
public static void main(String[] args) {
Exec e = new Exec();
for (int i = 0; i 2; i++) {
new Thread(e).start(); //开始执行时，队列为空，处于等待状态
}
//上面开启两个线程执行队列中的任务，那就是先到先得了
//添加一个任务测试
Queues.Task t =new Queues.Task();
Queues.add(t); //执行该方法，激活所有对应队列，那两个线程就会开始执行啦
}
}
上面的就是很简单的例子了
