java管道流的使用示例

import java.util.*;

import java.io.*;

/**

 * 本程序通过管道流实现了一个双人聊天室, 同一时刻只能有一个人在说，另外一个人在听，交替执行，某一方发出exit命令后，两个线程同时

 * 结束运行，表明双方同时离开聊天室

 * @author Administrator

 *

 */

class Talker extends Thread {

    Scanner s = null;

    // 自己说的内容

    String line = null;

    // 对方说的内容

    String othersLine = null;

    TalkRoom tr = null;

    PipedOutputStream out = new PipedOutputStream();

    PipedInputStream in = new PipedInputStream();

    public Talker(Scanner s, TalkRoom tr) {

        this.tr = tr;

        this.s = s;

    }

    public PipedOutputStream getOut() {

        return out;

    }

    public PipedInputStream getIn() {

        return in;

    }

    // 获取对方发出的信息

    public void getLine(String line) {

        othersLine = line;

    }

    public void run() {

        try {

            while (!("exit".equals(line))) {

                if (!("exit".equals(othersLine))) {

                    line = s.nextLine();

                    tr.speak(out, line);

                } else {

                    break;

                }

            }

            // 离开聊天室

            tr.leave(Thread.currentThread().getName());

        } catch (Exception e) {

            e.printStackTrace();

        } finally {

            try {

                in.close();

                out.close();

            } catch (IOException e) {

                e.printStackTrace();

            }

        }

    }

}

class TalkRoom {

    byte[] buf = new byte[1024];

    ArrayList<Talker> talkers = new ArrayList<Talker>();

    // 某个时刻某个人在说话

    public synchronized void speak(PipedOutputStream out, String s)

            throws IOException, InterruptedException {

        out.write(s.getBytes());

        int len = 0;

        String line = null;

        Iterator<Talker> iterator = talkers.iterator();

        while (iterator.hasNext()) {

            // 获取聊天内容

            Talker talker = iterator.next();

            if (!talker.getName().equals(Thread.currentThread().getName())) {

                talker.getIn().read(buf);

                line = new String(buf, 0, len);

                talker.getLine(s);

            }

        }

        System.out.println(Thread.currentThread().getName() + "说: " + s);    

        // 唤醒其它线程

        notify();

        // 当前线程等待

        wait();

    }

    // 增加聊天人

    public void addTalker(Talker talker) throws IOException {

        int i;

        if (talkers.size() == 0) {

            talkers.add(talker);

        } else {

            for (i = 0; i <= talkers.size() - 1; i++) {

                talker.getIn().connect(talkers.get(i).getOut());

                talkers.get(i).getIn().connect(talker.getOut());

            }

            talkers.add(talker);

        }

    }

    // 离开聊天室

    public synchronized void leave(String talker_Name)

            throws InterruptedException {

        System.out.println(Thread.currentThread().getName() + " leaved ...");

        Iterator<Talker> iterator = talkers.iterator();

        while (iterator.hasNext()) {

            Talker talker = iterator.next();

            if (talker.getName().equals(talker_Name)) {

                talkers.remove(talkers.indexOf(talker));

                break;

            }

        }

        if(talkers.size() >= 1){

            notify();

        }

    }

}

public class Main {

    public static void main(String[] args) throws IOException {

        TalkRoom tr = new TalkRoom();

        Scanner s1 = new Scanner(System.in);

        Scanner s2 = new Scanner(System.in);

        Talker one = new Talker(s1, tr);

        Talker two = new Talker(s2, tr);

        tr.addTalker(one);

        tr.addTalker(two);

        one.start();

        two.start();

    }

}