Java网络编程中怎样用管道流

Java网络编程中怎样用管道流

管道流可以实现两个线程之间，二进制数据的传输。
　　管道流就像一条管道，一端输入数据，别一端则输出数据。通常要分别用两个不同的线程来控制它们。
　　使用方法如下：
　　[html] view plaincopy
　　import java.io.IOException;
　　import java.io.PipedInputStream;
　　import java.io.PipedOutputStream;
　　
　　public class PipedInputStreamTest {
　　
　　public static void main(String[] args) {
　　//管道输出流
　　PipedOutputStream out = new PipedOutputStream();
　　//管道输入流
　　PipedInputStream in = null;
　　try {
　　//连接两个管道流。或者调用connect(Piped..);方法也可以
　　in = new PipedInputStream(out);
　　Thread read = new Thread(new Read(in));
　　Thread write = new Thread(new Write(out));
　　//启动线程
　　read.start();
　　write.start();
　　} catch (IOException e) {
　　e.printStackTrace();
　　}
　　}
　　}
　　
　　class Write implements Runnable {
　　PipedOutputStream pos = null;
　　
　　public Write(PipedOutputStream pos) {
　　this.pos = pos;
　　}
　　
　　public void run() {
　　try {
　　System.out.println("程序将在3秒后写入数据，请稍等。。。");
　　Thread.sleep(3000);
　　pos.write("wangzhihong".getBytes());
　　pos.flush();
　　} catch (IOException e) {
　　e.printStackTrace();
　　} catch (InterruptedException e) {
　　e.printStackTrace();
　　} finally {
　　try {
　　if (pos != null) {
　　pos.close();
　　}
　　} catch (IOException e) {
　　e.printStackTrace();
　　}
　　}
　　}
　　}
　　
　　class Read implements Runnable {
　　PipedInputStream pis = null;
　　
　　public Read(PipedInputStream pis) {
　　this.pis = pis;
　　}
　　
　　public void run() {
　　byte[] buf = new byte[1024];
　　try {
　　pis.read(buf);
　　System.out.println(new String(buf));
　　} catch (IOException e) {
　　e.printStackTrace();
　　} finally {
　　try {
　　if (pis != null) {
　　pis.close();
　　}
　　} catch (IOException e) {
　　e.printStackTrace();
　　}
　　}
　　}
　　}

Server端
------------------
import java.io.*;
import java.net.*;
import java.util.Scanner;
public class Server {
public static int PORT = 1007;
private ServerSocket server;
public Server() throws IOException {
server = new ServerSocket(PORT);
}
public static void main(String[] args) throws Exception {
System.out.println("--Server--");
Server server = new Server();
server.accept();
}
public void accept() throws IOException {
Socket client = server.accept();//等待客户端的连接
getInfo(client.getInputStream());//启动等待消息线程
toInfo(client.getOutputStream());//启动发送消息线程
}
//等待客户端发送消息
public void getInfo(InputStream in) throws IOException {
final Scanner sc = new Scanner(in);//获取客户端的输入流
new Thread() {
@Override
public void run() {
while(true) {
if(sc.hasNextLine()) {//如果客户端有发送消息过来
System.out.println("Client:" + sc.nextLine());//打印客户端的消息
}
}
}
}.start();
}
//发送消息到客户端
public void toInfo(OutputStream out) throws IOException {
final PrintWriter pw = new PrintWriter(out, true);//获取客户端的输出流，自动清空缓存的内容
final Scanner sc = new Scanner(System.in);//获取控制台的标准输入流，从控制台输入数据
new Thread() {
@Override
public void run() {
while(true) {
pw.println(sc.nextLine());//将输入的数据发送给客户端
}
}
}.start();
}
}
Client端
------------------------
import java.io.*;
import java.net.Socket;
import java.util.Scanner;
public class Client {
public static void main(String[] args) throws Exception, IOException {
System.out.println("--Client--");
Client client = new Client();
client.connection("localhost", Server.PORT);
}
public void connection(String host, int port) throws IOException {
Socket client = new Socket(host, port);
getInfo(client.getInputStream());
toInfo(client.getOutputStream());
}
public void getInfo(InputStream in) throws IOException {
final Scanner sc = new Scanner(in);
new Thread() {
@Override
public void run() {
while(true) {
if(sc.hasNextLine()) {
System.out.println("Server:" + sc.nextLine());
}
}
}
}.start();
}
public void toInfo(OutputStream out) throws IOException {
final PrintWriter pw = new PrintWriter(out, true);
final Scanner sc = new Scanner(System.in);
new Thread() {
@Override
public void run() {
while(true) {
pw.println(sc.nextLine());
}
}
}.start();
}
}

管道流就像一条条水管只是有些水管没有水而已！网路流就像外来水管裏面有水！通过编程对接将网路流连接到IO流或然後将它打印出来或做其它处理。

一起学习，管道流满搞的，用着比socket的stream麻烦
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.*;
import java.nio.channels.*;
import java.nio.*;
import java.util.*;
import java.nio.charset.*;
public class TCP extends Thread{
private SocketChannel channel;
private ServerSocket serverSocket;
private ServerSocketChannel serverSocketChannel;
private ByteBuffer readBuffer;
private ByteBuffer sendBuffer;
private Boolean isAccept=false;
private boolean isConnect=false;
private Thread accept;
private Thread connect;

public TCP(int port,String addr) {
try {
readBuffer=ByteBuffer.allocate(1024);
serverSocketChannel=ServerSocketChannel.open();
serverSocket=serverSocketChannel.socket();
serverSocket.setReuseAddress(true);
serverSocket.bind(new InetSocketAddress(port));
channel=SocketChannel.open();
channel.connect(new InetSocketAddress(InetAddress.getByName(addr),port));
accept=new Thread(){
public void run(){
Selector selector;
try {
selector=Selector.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.register(selector,SelectionKey.OP_ACCEPT);
isAccept=false;
selectors(selector);
} catch (ClosedChannelException ex) {
ex.printStackTrace();
} catch (IOException ex) {
ex.printStackTrace();
}
}
};
connect=new Thread(){
public void run(){
try{
Selector selector;
selector=Selector.open();
channel.configureBlocking(false);
channel.register(selector,SelectionKey.OP_WRITE|SelectionKey.OP_READ);
isConnect=false;
selectors(selector);
}catch (Exception ex){
ex.printStackTrace();
}
}
};
} catch (IOException ex) {
ex.printStackTrace();
System.out.println("d1");
}catch(Exception ex){
System.out.println("d2");
}
}
private void service(){
if(isConnect){
connect.start();
}
if(isAccept){
accept.start();
}
}
private void selectors(Selector selector){
try {
while (selector.select()>0){
Set readyKeys=selector.selectedKeys();
Iterator it=readyKeys.iterator();
while(it.hasNext()){
SelectionKey key=null;
key=it.next();
it.remove();
if(key.isAcceptable()){
//System.out.println("isAcceptable");
ServerSocketChannel ssc=(ServerSocketChannel)key.channel();
SocketChannel socketChannel=ssc.accept();
socketChannel.configureBlocking(false);
socketChannel.register(selector,SelectionKey.OP_READ|SelectionKey.OP_WRITE);
}
if(key.isReadable()){
synchronized(readBuffer){
// System.out.println("isReadable");
ByteBuffer buffer=ByteBuffer.allocate(1024);
SocketChannel socketChannel=(SocketChannel)key.channel();
socketChannel.read(buffer);
readBuffer=buffer;
}
}
if(key.isWritable()){
synchronized(sendBuffer){
// System.out.println("isWritable");
SocketChannel channel=(SocketChannel)key.channel();
if(sendBuffer!=null)
channel.write(sendBuffer);
}
}

}
try {
sleep(1);
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
} catch (ClosedChannelException ex) {
ex.printStackTrace();
} catch (IOException ex) {
ex.printStackTrace();
}
}
public void send(ByteBuffer buff){
this.sendBuffer=buff;
}
public ByteBuffer get(){
return readBuffer;
}
public void accpet(){
isAccept=true;
}
public void connect(){
isConnect=true;
}
public void run(){
while (true){
service();
}
}
}

一起学习，管道流满搞的，用着比socket的stream麻烦
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.*;
import java.nio.channels.*;
import java.nio.*;
import java.util.*;
import java.nio.charset.*;
public class TCP extends Thread{
private SocketChannel channel;
private ServerSocket serverSocket;
private ServerSocketChannel serverSocketChannel;
private ByteBuffer readBuffer;
private ByteBuffer sendBuffer;
private Boolean isAccept=false;
private boolean isConnect=false;
private Thread accept;
private Thread connect;

public TCP(int port,String addr) {
try {
readBuffer=ByteBuffer.allocate(1024);
serverSocketChannel=ServerSocketChannel.open();
serverSocket=serverSocketChannel.socket();
serverSocket.setReuseAddress(true);
serverSocket.bind(new InetSocketAddress(port));
channel=SocketChannel.open();
channel.connect(new InetSocketAddress(InetAddress.getByName(addr),port));
accept=new Thread(){
public void run(){
Selector selector;
try {
selector=Selector.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.register(selector,SelectionKey.OP_ACCEPT);
isAccept=false;
selectors(selector);
} catch (ClosedChannelException ex) {
ex.printStackTrace();
} catch (IOException ex) {
ex.printStackTrace();
}
}
};
connect=new Thread(){
public void run(){
try{
Selector selector;
selector=Selector.open();
channel.configureBlocking(false);
channel.register(selector,SelectionKey.OP_WRITE|SelectionKey.OP_READ);
isConnect=false;
selectors(selector);
}catch (Exception ex){
ex.printStackTrace();
}
}
};
} catch (IOException ex) {
ex.printStackTrace();
System.out.println("d1");
}catch(Exception ex){
System.out.println("d2");
}
}
private void service(){
if(isConnect){
connect.start();
}
if(isAccept){
accept.start();
}
}
private void selectors(Selector selector){
try {
while (selector.select()>0){
Set readyKeys=selector.selectedKeys();
Iterator<SelectionKey> it=readyKeys.iterator();
while(it.hasNext()){
SelectionKey key=null;
key=it.next();
it.remove();
if(key.isAcceptable()){
//System.out.println("isAcceptable");
ServerSocketChannel ssc=(ServerSocketChannel)key.channel();
SocketChannel socketChannel=ssc.accept();
socketChannel.configureBlocking(false);
socketChannel.register(selector,SelectionKey.OP_READ|SelectionKey.OP_WRITE);
}
if(key.isReadable()){
synchronized(readBuffer){
// System.out.println("isReadable");
ByteBuffer buffer=ByteBuffer.allocate(1024);
SocketChannel socketChannel=(SocketChannel)key.channel();
socketChannel.read(buffer);
readBuffer=buffer;
}
}
if(key.isWritable()){
synchronized(sendBuffer){
// System.out.println("isWritable");
SocketChannel channel=(SocketChannel)key.channel();
if(sendBuffer!=null)
channel.write(sendBuffer);
}
}

}
try {
sleep(1);
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
} catch (ClosedChannelException ex) {
ex.printStackTrace();
} catch (IOException ex) {
ex.printStackTrace();
}
}
public void send(ByteBuffer buff){
this.sendBuffer=buff;
}
public ByteBuffer get(){
return readBuffer;
}
public void accpet(){
isAccept=true;
}
public void connect(){
isConnect=true;
}
public void run(){
while (true){
service();
}
}
}