ยกตัวอย่างเพื่ออธิบายการควบคุมการสื่อสารของเธรดของกระแสอินพุตและเอาต์พุตของไปป์ไลน์ไปป์ไลน์ใน Java

PipedOutputStream และ PipedInputStream

ใน Java, PipedOutputStream และ PipedInputStream เป็นสตรีมเอาท์พุทไปป์ไลน์และสตรีมอินพุตไปป์ไลน์ตามลำดับ
ฟังก์ชั่นของพวกเขาคืออนุญาตให้มัลติเธรดสื่อสารระหว่างเธรดผ่านท่อ เมื่อใช้การสื่อสารไปป์ไลน์ต้องใช้ PipedOutputStream และ PipedInputStream ร่วมกัน
เมื่อใช้การสื่อสารไปป์ไลน์กระบวนการทั่วไปคือ: เราเขียนข้อมูลไปยัง PipedOutputStream ในเธรด A และข้อมูลเหล่านี้จะถูกส่งไปยัง PipedInputStream โดยอัตโนมัติที่สอดคล้องกับ PipedOutputStream จากนั้นเก็บไว้ในบัฟเฟอร์ของ PipedInputStream; ในเวลานี้เธรด B อ่านข้อมูลใน PipedInputStream สิ่งนี้สามารถตระหนักถึงการสื่อสารระหว่างเธรด A และ Thread B
ด้านล่างเราดูตัวอย่างของการสื่อสารผ่านท่อในมัลติเธรด ตัวอย่างรวมถึง 3 คลาส: receiver.java, pipedstreamtest.java และ sender.java
รหัสของตัวรับสัญญาณ Java มีดังนี้:

 นำเข้า java.io.ioException; นำเข้า Java.io.pipedInputStream; @suppresswarnings ("ทั้งหมด") / *** เธรดตัวรับสัญญาณ* / ตัวรับสัญญาณระดับสาธารณะขยายเธรด {// วัตถุสตรีมอินพุตไปป์ไลน์ // มันถูกผูกไว้กับวัตถุ "pipedoutputstream" // สิ่งนี้ช่วยให้คุณได้รับข้อมูลของ "pipedoutputstream" จากนั้นให้ผู้ใช้อ่าน Private PipedInputStream ใน = New PipedInputStream (); // รับวัตถุ "ท่ออินพุตท่อ" PIPEDINPUTSTREAM GetInputStream () {return in; } @Override โมฆะสาธารณะ Run () {readMessageOnce (); // readmessagecontinued (); } // อ่านข้อมูลหนึ่งครั้งจาก "Pipe Input Stream" โมฆะสาธารณะ readMessageOnce () {// แม้ว่าขนาดของ BUF คือ 2048 ไบต์ แต่จะอ่านได้มากที่สุด 1024 ไบต์จาก "ท่ออินพุตท่อ" // เนื่องจากขนาดบัฟเฟอร์ของ "กระแสอินพุตท่อ" เป็นเพียง 1024 ไบต์โดยค่าเริ่มต้น ไบต์ [] buf = ไบต์ใหม่ [2048]; ลอง {int len ​​= in.read (buf); System.out.println (สตริงใหม่ (buf, 0, len)); in.close (); } catch (ioexception e) {e.printstacktrace (); }} // เมื่ออ่าน> 1024 ไบต์จาก "สตรีมอินพุตท่อ" หยุดอ่านโมฆะสาธารณะ readMessageContinued () {int total = 0; ในขณะที่ (จริง) {byte [] buf = byte ใหม่ [1024]; ลอง {int len ​​= in.read (buf); ทั้งหมด += len; System.out.println (สตริงใหม่ (buf, 0, len)); // หากจำนวนทั้งหมดของไบต์ที่อ่านคือ> 1024 ให้ออกจากลูป ถ้า (ทั้งหมด> 1024) หยุด; } catch (ioexception e) {e.printstacktrace (); }} ลอง {in.close (); } catch (ioexception e) {e.printstacktrace (); -

รหัสของ Sender.java มีดังนี้:

 นำเข้า java.io.ioException; นำเข้า java.io.pipedOutputStream; @suppresswarnings ("ทั้งหมด")/ *** เธรดผู้ส่ง*/ ผู้ส่งคลาสสาธารณะขยายเธรด {// วัตถุกระแสเอาต์พุตท่อส่งออก // มันถูกผูกไว้กับวัตถุ "PipedInputStream" // สิ่งนี้อนุญาตให้ส่งข้อมูลไปยังข้อมูลของ "PipedInputStream" และจากนั้นผู้ใช้สามารถอ่านข้อมูลจาก "PipedInputStream" Private PipedOutputStream Out = ใหม่ pipedOutputStream (); // รับวัตถุ "ท่อส่งออก" PIPEDOUTPUTSTREAM GETOUTPUTSTREAM () {return out; } @Override โมฆะสาธารณะ Run () {writeshortMessage (); // writelongmessage (); } // เขียนข้อความสั้น ๆ ไปยัง "สตรีมเอาท์พุทท่อ": "นี่เป็นข้อความสั้น ๆ ลอง {out.write (strinfo.getBytes ()); out.close (); } catch (ioexception e) {e.printstacktrace (); }} // เขียนข้อความยาวไปยัง "ท่อส่งออกท่อ" โมฆะส่วนตัว writeLongMessage () {StringBuilder sb = new StringBuilder (); // เขียน 1020 ไบต์ผ่าน A for loop สำหรับ (int i = 0; i <102; i ++) sb.append ("0123456789"); // เขียน 26 ไบต์ sb.append ("abcdefghijklmnopqrstuvwxyz"); // ความยาวทั้งหมดของ str คือ 1020+26 = 1046 ไบต์สตริง str = sb.toString (); ลอง {// เขียน 1046 ไบต์ลงใน "ท่อส่งออกท่อ" out.write (str.getBytes ()); out.close (); } catch (ioexception e) {e.printstacktrace (); -

รหัสของ pipedstreamtest.java มีดังนี้:

 นำเข้า java.io.pipedInputStream; นำเข้า java.io.pipedOutputStream; นำเข้า java.io.ioException; @SuppressWarnings ("ทั้งหมด") / *** โปรแกรมอินเทอร์แอคทีฟสำหรับกระแสอินพุตท่อ ตัวรับสัญญาณ T2 = ตัวรับสัญญาณใหม่ (); pipedOutputStream out = t1.getOutputStream (); PipedInputStream ใน = t2.getInputStream (); ลอง {// การเชื่อมต่อท่อ สาระสำคัญของสองประโยคต่อไปนี้เหมือนกัน //out.connect(in); in.connect (ออก); /** * วิธีการเริ่มต้นของคลาสเธรด: * ทำให้เธรดเริ่มดำเนินการ; เครื่องเสมือน Java เรียกใช้วิธีการเรียกใช้เธรด * ผลลัพธ์คือสองเธรดทำงานพร้อมกัน เธรดปัจจุบัน (ส่งคืนจากการโทรไปยังวิธีการเริ่มต้น) และเธรดอื่น ๆ (ดำเนินการวิธีการเรียกใช้) * มันผิดกฎหมายที่จะเริ่มต้นด้ายหลายครั้ง โดยเฉพาะอย่างยิ่งเมื่อเธรดเสร็จสิ้นการดำเนินการเสร็จแล้วมันไม่สามารถรีสตาร์ทได้ */ t1.start (); t2.start (); } catch (ioexception e) {e.printstacktrace (); -

ผลการทำงาน:

 นี่เป็นข้อความสั้น ๆ

ภาพประกอบ:
(1) in.connect (ออก); เชื่อมโยง "กระแสอินพุตท่อ" และ "สตรีมเอาต์พุตท่อ" ตรวจสอบซอร์สโค้ดของ Connect () ใน PipedOutputStream.java และ PipedInputStream.java; เรารู้ว่ามีการเชื่อมต่อ (ใน); เทียบเท่ากับ in.connect (ออก);
(2)

 t1.start (); // เริ่มเธรด "ผู้ส่ง" t2.start (); // เริ่มเธรด "ตัวรับสัญญาณ"

ก่อนอื่นตรวจสอบซอร์สโค้ดของผู้ส่ง. java และเรียกใช้ฟังก์ชัน Run () หลังจากเริ่มเธรด ในการเรียกใช้ () ของผู้ส่ง. java โทร writeshortmessage ();
ฟังก์ชั่นของ WriteshortMessage (); คือการเขียนข้อมูล "นี่คือข้อความสั้น ๆ " ไปยัง "กระแสเอาท์พุทท่อ"; ข้อมูลนี้จะได้รับจาก "กระแสอินพุตท่อ" มาดูกันว่าสิ่งนี้ประสบความสำเร็จได้อย่างไร
ก่อนอื่นให้ดูที่ซอร์สโค้ดของการเขียน (byte b []) และกำหนดใน outputStream.java PipedOutputStream.java สืบทอดมาจาก outputstream.java; ซอร์สโค้ดของการเขียน (byte b []) ใน outputstream.java มีดังนี้:

 โมฆะสาธารณะเขียน (ไบต์ b []) พ่น IOException {เขียน (b, 0, b.length);}

ในความเป็นจริงการเขียน (ไบต์ b []) คือการเขียนการโทร (ไบต์ b [], int ปิด, int len) ฟังก์ชั่นใน pipedoutputstream.java เมื่อดูที่ซอร์สโค้ดของการเขียน (ไบต์ b [], int ปิด, int len) เราพบว่ามันจะเรียก sink.receive (b, ปิด, len); เพิ่มเติมที่คำจำกัดความของการรับ (ไบต์ b [], int ปิด, int len) เรารู้ว่า sink.receive (b, ปิด, len) คือการบันทึกข้อมูลใน "กระแสท่อเอาท์พุท" ลงในบัฟเฟอร์ของ "ท่อกระแสอินพุตท่อ" ขนาดเริ่มต้นของบัฟเฟอร์บัฟเฟอร์ของ "สตรีมอินพุตท่อ" คือ 1024 ไบต์
ณ จุดนี้เรารู้ว่า: t1.start () เริ่มเธรดผู้ส่งและเธรดผู้ส่งจะเขียนข้อมูล "นี่คือข้อความสั้น ๆ " ไปยัง "กระแสท่อส่งออก"; และ "สตรีมเอาท์พุทท่อ" จะถ่ายโอนข้อมูลไปยัง "กระแสอินพุตท่อ" นั่นคือมันจะถูกบันทึกไว้ในบัฟเฟอร์ของ "กระแสอินพุตท่อ"
ต่อไปเราจะดูว่า "วิธีที่ผู้ใช้อ่านข้อมูลจากบัฟเฟอร์ของ 'กระแสอินพุตท่อ'" นี่คือการกระทำของเธรดตัวรับสัญญาณ
t2.start () จะเริ่มเธรดตัวรับสัญญาณดังนั้นจึงเรียกใช้ฟังก์ชันตัวรับสัญญาณ Java Run () เมื่อดูที่ซอร์สโค้ดของตัวรับสัญญาณเรารู้ว่าเรียกใช้ () เรียกใช้ readMessageOnce ()
readMessageOnce () คือการโทรในการอ่าน (buf) เพื่ออ่านข้อมูลจาก "สตรีมอินพุตท่อใน" และบันทึกลงใน buf
จากการวิเคราะห์ข้างต้นเรารู้อยู่แล้วว่าข้อมูลในบัฟเฟอร์ของ "กระแสอินพุตท่อใน" คือ "นี่คือข้อความสั้น ๆ "; ดังนั้นข้อมูลของ BUF คือ "นี่คือข้อความสั้น ๆ "
เพื่อให้เข้าใจถึงท่อลึกซึ้งยิ่งขึ้น เราจะทำการทดลองเล็ก ๆ สองครั้งต่อไปนี้
การทดลองที่ 1: แก้ไขผู้ส่ง. java
จะ

 Public Void Run () {WriteshortMessage (); // writeLongMessage ();}

ดัดแปลงเป็น

 โมฆะสาธารณะเรียกใช้ () {// writeshortMessage (); writeLongMessage ();}

เรียกใช้โปรแกรม ผลการทำงานคือ:

ข้อมูลเหล่านี้ถูกเขียนไปยัง "กระแสเอาต์พุตท่อ" ผ่าน writeLongMessage () จากนั้นถ่ายโอนไปยัง "กระแสอินพุตท่อ" จากนั้นเก็บไว้ในบัฟเฟอร์ของ "กระแสอินพุตท่อ"; จากนั้นอ่านจากบัฟเฟอร์โดยผู้ใช้
จากนั้นสังเกตซอร์สโค้ดของ WriteLongMessage () เราสามารถพบว่าความยาวของ STR คือ 1,046 ไบต์และจากนั้นผลลัพธ์ของการวิ่งเพียง 1024 ไบต์! ทำไมสิ่งนี้ถึงเกิดขึ้น?
เหตุผลนั้นง่าย: ขนาดเริ่มต้นของบัฟเฟอร์ของสตรีมอินพุตไปป์ไลน์คือ 1024 ไบต์ ดังนั้นมากที่สุดคือ 1,024 ไบต์สามารถเขียนได้
โดยการสังเกตซอร์สโค้ดของ PipedInputStream.java เราสามารถเข้าใจได้อย่างละเอียดมากขึ้น

 INT สุดท้ายคงที่ int default_PIPE_SIZE = 1024; PIPEDINPUTSTREAM () {initPipe (default_pipe_size);}

ตัวสร้างเริ่มต้นการเรียกใช้ initpipe (default_pipe_size) และซอร์สโค้ดของมันมีดังนี้:

 โมฆะส่วนตัว initPipe (int pipesize) {ถ้า (pipesize <= 0) {โยน unlegalargumentException ใหม่ ("ขนาดท่อ <= 0"); } buffer = byte ใหม่ [pipesize];}

จากนี้เราสามารถรู้ได้ว่าขนาดเริ่มต้นของบัฟเฟอร์บัฟเฟอร์คือ 1024 ไบต์
การทดลองที่ 2: ดำเนินการต่อเพื่อปรับเปลี่ยน receiver.java บนพื้นฐานของ "การทดลอง 1"
จะ

 โมฆะสาธารณะเรียกใช้ () {readMessageOnce (); // readMessagecontinued ();}

ดัดแปลงเป็น

 โมฆะสาธารณะเรียกใช้ () {// readMessageOnce (); READMESSAGECONTINUED ();}

เรียกใช้โปรแกรม ผลการทำงานคือ:

ผลลัพธ์นี้เป็นข้อมูลที่สมบูรณ์ที่เขียนลงใน "อินพุตบัฟเฟอร์"

PipedWriter และ PipedReader
PipedWriter เป็นสตรีมเอาต์พุตไปป์ไลน์อักขระซึ่งสืบทอดมาจากนักเขียน
PipedReader เป็นสตรีมอินพุตของตัวอักษรไปป์ไลน์ที่สืบทอดมาจากนักเขียน
ฟังก์ชั่นของ PipedWriter และ PipedReader คือการสื่อสารระหว่างเธรดผ่านท่อ เมื่อใช้การสื่อสารไปป์ไลน์ต้องใช้ PipedWriter และ PipedReader ร่วมกัน
ด้านล่างเราดูตัวอย่างของการสื่อสารผ่าน PipedWriter และ PipedReader ในมัลติเธรด ตัวอย่างรวมถึง 3 คลาส: receiver.java, sender.java และ pipetest.java
รหัสของตัวรับสัญญาณ Java มีดังนี้:

 นำเข้า java.io.ioException; นำเข้า java.io.pipedReader; @suppresswarnings ("ทั้งหมด") / *** เธรดตัวรับสัญญาณ* / ตัวรับสัญญาณระดับสาธารณะขยายเธรด {// วัตถุสตรีมอินพุตไปป์ไลน์ // มันถูกผูกไว้กับวัตถุ "pipedwriter" // สิ่งนี้ช่วยให้คุณได้รับข้อมูลของ "pipedwriter" จากนั้นให้ผู้ใช้อ่าน Private PipedReader ใน = new PipedReader (); // รับ "ท่ออินพุตท่อ" pipedreader getReader () {return in; } @Override โมฆะสาธารณะ Run () {readMessageOnce (); // readmessagecontinued (); } // อ่านข้อมูลหนึ่งครั้งจาก "Pipe Input Stream" โมฆะสาธารณะ readMessageOnce () {// แม้ว่าขนาดของ BUF คือ 2048 อักขระ แต่จะอ่านได้ที่ 1024 อักขระจาก "กระแสอินพุตท่อ" // เนื่องจากขนาดบัฟเฟอร์ของ "กระแสอินพุตท่อ" เป็นเพียง 1024 อักขระตามค่าเริ่มต้น ถ่าน [] buf = ถ่านใหม่ [2048]; ลอง {int len ​​= in.read (buf); System.out.println (สตริงใหม่ (buf, 0, len)); in.close (); } catch (ioexception e) {e.printstacktrace (); }} // เมื่ออ่าน> 1024 อักขระจาก "กระแสอินพุตท่อ" หยุดอ่านโมฆะสาธารณะ readMessageContinued () {int total = 0; ในขณะที่ (จริง) {char [] buf = char ใหม่ [1024]; ลอง {int len ​​= in.read (buf); ทั้งหมด += len; System.out.println (สตริงใหม่ (buf, 0, len)); // หากจำนวนอักขระทั้งหมดที่อ่านคือ> 1024 ลูปจะถูกออก ถ้า (ทั้งหมด> 1024) หยุด; } catch (ioexception e) {e.printstacktrace (); }} ลอง {in.close (); } catch (ioexception e) {e.printstacktrace (); -

รหัสของ Sender.java มีดังนี้:

 นำเข้า java.io.ioException; นำเข้า java.io.pipedWriter; @suppresswarnings ("ทั้งหมด")/ *** เธรดผู้ส่ง*/ ผู้ส่งคลาสสาธารณะขยายเธรด {// วัตถุกระแสเอาต์พุตท่อส่งออก // มันถูกผูกไว้กับวัตถุ "pipedReader" // สิ่งนี้อนุญาตให้ส่งข้อมูลไปยังข้อมูลของ "pipedReader" และจากนั้นผู้ใช้สามารถอ่านข้อมูลจาก "pipedReader" Private PipedWriter Out = New PipedWriter (); // รับวัตถุ "ท่อส่งออก" PIPEDWRITER PIPEDWRITER GETWRITER () {return out; } @Override โมฆะสาธารณะ Run () {writeshortMessage (); // writelongmessage (); } // เขียนข้อความสั้น ๆ ไปยัง "สตรีมเอาท์พุทท่อ": "นี่เป็นข้อความสั้น ๆ ลอง {out.write (strinfo.tochararray ()); out.close (); } catch (ioexception e) {e.printstacktrace (); }} // เขียนข้อความยาวไปยัง "ท่อส่งออกท่อ" โมฆะส่วนตัว writeLongMessage () {StringBuilder sb = new StringBuilder (); // เขียน 1020 อักขระผ่าน A for loop สำหรับ (int i = 0; i <102; i ++) sb.append ("0123456789"); // เขียน 26 อักขระเพิ่มเติม sb.append ("abcdefghijklmnopqrstuvwxyz"); // ความยาวทั้งหมดของ str คือ 1020+26 = 1046 อักขระสตริง str = sb.toString (); ลอง {// เขียน 1046 อักขระลงใน "ท่อส่งออกท่อ" out.write (str); out.close (); } catch (ioexception e) {e.printstacktrace (); -

รหัสของ pipetest.java มีดังนี้:

 นำเข้า java.io.pipedReader; นำเข้า java.io.pipedWriter; นำเข้า java.io.ioException; @SuppressWarnings ("ทั้งหมด") / *** โปรแกรมแบบอินเทอร์แอคทีฟสำหรับสตรีมอินพุตท่อส่งสัญญาณ ตัวรับสัญญาณ T2 = ตัวรับสัญญาณใหม่ (); pipedWriter out = t1.getWriter (); pipedReader ใน = t2.getReader (); ลอง {// การเชื่อมต่อท่อ สาระสำคัญของสองประโยคต่อไปนี้เหมือนกัน //out.connect(in); in.connect (ออก); /** * วิธีการเริ่มต้นของคลาสเธรด: * ทำให้เธรดเริ่มดำเนินการ; เครื่องเสมือน Java เรียกใช้วิธีการเรียกใช้เธรด * ผลลัพธ์คือสองเธรดทำงานพร้อมกัน เธรดปัจจุบัน (ส่งคืนจากการโทรไปยังวิธีการเริ่มต้น) และเธรดอื่น ๆ (ดำเนินการวิธีการเรียกใช้) * มันผิดกฎหมายที่จะเริ่มต้นด้ายหลายครั้ง โดยเฉพาะอย่างยิ่งเมื่อเธรดเสร็จสิ้นการดำเนินการเสร็จแล้วมันไม่สามารถรีสตาร์ทได้ */ t1.start (); t2.start (); } catch (ioexception e) {e.printstacktrace (); -

ผลการทำงาน:
นี่เป็นข้อความสั้น ๆ
ผลลัพธ์คำอธิบาย:

(1)

 in.connect (ออก);

ฟังก์ชั่นของมันคือการเชื่อมโยง "สตรีมอินพุตท่อ" และ "สตรีมเอาต์พุตท่อ" ตรวจสอบซอร์สโค้ดของ Connect () ใน pipedWriter.java และ pipedReader.java; เรารู้ว่ามีการเชื่อมต่อ (ใน); เทียบเท่ากับ in.connect (ออก);
(2)

 t1.start (); // เริ่มเธรด "ผู้ส่ง" t2.start (); // เริ่มเธรด "ตัวรับสัญญาณ"

ก่อนอื่นตรวจสอบซอร์สโค้ดของผู้ส่ง. java และเรียกใช้ฟังก์ชัน Run () หลังจากเริ่มเธรด ในการเรียกใช้ () ของผู้ส่ง. java โทร writeshortmessage ();
ฟังก์ชั่นของ WriteshortMessage (); คือการเขียนข้อมูล "นี่คือข้อความสั้น ๆ " ไปยัง "กระแสเอาท์พุทท่อ"; ข้อมูลนี้จะได้รับจาก "กระแสอินพุตท่อ" มาดูกันว่าสิ่งนี้ประสบความสำเร็จได้อย่างไร
ก่อนอื่นให้ดูที่ซอร์สโค้ดของการเขียน (Char Char. PipedWriter.java สืบทอดมาจาก Writer.java; ซอร์สโค้ดของการเขียน (Char C []) ใน Writer.java มีดังนี้:

 โมฆะสาธารณะเขียน (ถ่าน cbuf []) พ่น IOException {เขียน (cbuf, 0, cbuf.length);}

ในความเป็นจริงเขียน (ถ่าน C []) คือการเขียนการโทร (ถ่าน C [], int ปิด, int len) ฟังก์ชั่นใน pipedWriter.java เมื่อดูที่ซอร์สโค้ดของการเขียน (Char C [], int ปิด, int len) เราพบว่ามันจะเรียก sink.receive (CBUF, OFF, LEN); เพิ่มเติมที่คำจำกัดความของการรับ (char c [], int ปิด, int len) เรารู้ว่า sink.receive (cbuf, OFF, LEN) คือการบันทึกข้อมูลใน "ท่อส่งออกท่อ" ลงในบัฟเฟอร์ของ "กระแสอินพุตท่อ" ขนาดเริ่มต้นของบัฟเฟอร์บัฟเฟอร์ของ "กระแสอินพุตท่อ" คือ 1024 อักขระ
ณ จุดนี้เรารู้ว่า: t1.start () เริ่มเธรดผู้ส่งและเธรดผู้ส่งจะเขียนข้อมูล "นี่คือข้อความสั้น ๆ " ไปยัง "กระแสท่อส่งออก"; และ "สตรีมเอาท์พุทท่อ" จะถ่ายโอนข้อมูลไปยัง "กระแสอินพุตท่อ" นั่นคือมันจะถูกบันทึกไว้ในบัฟเฟอร์ของ "กระแสอินพุตท่อ"
ต่อไปเราจะดูว่า "วิธีที่ผู้ใช้อ่านข้อมูลจากบัฟเฟอร์ของ 'กระแสอินพุตท่อ'" นี่คือการกระทำของเธรดตัวรับสัญญาณ
t2.start () จะเริ่มเธรดตัวรับสัญญาณดังนั้นจึงเรียกใช้ฟังก์ชันตัวรับสัญญาณ Java Run () เมื่อดูที่ซอร์สโค้ดของตัวรับสัญญาณเรารู้ว่าเรียกใช้ () เรียกใช้ readMessageOnce ()
readMessageOnce () คือการโทรในการอ่าน (buf) เพื่ออ่านข้อมูลจาก "สตรีมอินพุตท่อใน" และบันทึกลงใน buf
จากการวิเคราะห์ข้างต้นเรารู้อยู่แล้วว่าข้อมูลในบัฟเฟอร์ของ "กระแสอินพุตท่อใน" คือ "นี่คือข้อความสั้น ๆ "; ดังนั้นข้อมูลของ BUF คือ "นี่คือข้อความสั้น ๆ "
เพื่อให้เข้าใจถึงท่อลึกซึ้งยิ่งขึ้น เราจะทำการทดลองเล็ก ๆ สองครั้งต่อไปนี้
การทดลองที่ 1: แก้ไขผู้ส่ง. java
จะ

 Public Void Run () {WriteshortMessage (); // writeLongMessage ();}

ดัดแปลงเป็น

 โมฆะสาธารณะเรียกใช้ () {// writeshortMessage (); writeLongMessage ();}

เรียกใช้โปรแกรม ผลการดำเนินการมีดังนี้:

จากนี้เราจะเห็นว่าโปรแกรมทำงานไม่ถูกต้อง! โยน java.io.ioexception: Pipe ปิด
ทำไมสิ่งนี้ถึงเกิดขึ้น?
ฉันจะวิเคราะห์การไหลของโปรแกรม
(1) ใน pipetest ให้เชื่อมต่อท่ออินพุตและเอาต์พุตผ่าน in.connect (out); จากนั้นเริ่มสองเธรด T1.Start () เริ่มต้นผู้ส่งเธรดและ T2.Start () เริ่มต้นเครื่องรับเธรด
(2) หลังจากเริ่มเธรดผู้ส่งข้อมูลจะถูกเขียนลงใน "ท่อส่งออก" ผ่าน writeLongMessage () และ out.write (str.tochararray ()) เขียนทั้งหมด 1046 อักขระ ตามซอร์สโค้ดของ PipedWriter ฟังก์ชั่นการเขียน () ของ PIPEDWriter จะเรียกฟังก์ชันรับ () ของ PIPEDREADER เมื่อดูฟังก์ชั่นการรับ () ของ PipEdReader เรารู้ว่า PipEdReader จะเก็บบัฟเฟอร์ข้อมูลที่ยอมรับไว้ หากคุณสังเกตฟังก์ชั่นรับ () อย่างระมัดระวังมีรหัสต่อไปนี้:

 ในขณะที่ (ใน == ออก) {ถ้า ((readside! = null) &&! readside.isalive ()) {โยน iOexception ใหม่ ("ท่อแตก"); } / * เต็ม: เตะผู้อ่านที่รออยู่ * / notifyall (); ลอง {รอ (1,000); } catch (interruptedException ex) {โยน java.io.Io.InterruptedioException ใหม่ (); -

ค่าเริ่มต้นของเข้าและออกอยู่ใน = -1, out = 0, ตามลำดับ; รวมกับข้างต้นในขณะที่ (ใน == ออก) เรารู้ว่าความหมายของมันคือทุกครั้งที่ตัวละครถูกเขียนลงในท่อเงื่อนไขใน == ออกจะพบ จากนั้น NotifyAll () จะถูกเรียกให้ปลุก "เธรดที่อ่านไปป์ไลน์"
นั่นคือทุกครั้งที่ตัวละครถูกเขียนลงในท่อมันจะบล็อกและรอให้อ่านเธรดอื่น ๆ
อย่างไรก็ตามขนาดเริ่มต้นของบัฟเฟอร์ของ PipedReader คือ 1024! อย่างไรก็ตามมีข้อมูล 1046 ที่จะเขียนในเวลานี้! ดังนั้นจึงสามารถเขียนอักขระได้ครั้งละ 1024 ตัว
(03) หลังจากเริ่มเธรดตัวรับสัญญาณ readMessageOnce () จะถูกเรียกให้อ่านสตรีมอินพุตท่อ การอ่าน 1024 อักขระจะเสร็จสิ้นและปิด () จะถูกเรียกให้ปิดท่อ
จากการวิเคราะห์ของ (02) และ (03) จะเห็นได้ว่าผู้ส่งจำเป็นต้องเขียน 1046 อักขระลงในท่อ ในหมู่พวกเขาสามารถเขียนอักขระ 1024 ตัวแรก (ความจุบัฟเฟอร์คือ 1024) สามารถเขียนได้ตามปกติและหนึ่งจะอ่านสำหรับแต่ละการเขียน เมื่อเขียนอักขระ 1025 ให้เขียน () ใน pipedwriter.java ยังคงถูกเรียกตามลำดับ; จากนั้นรับ () ใน pipedReader.java เรียกว่า; ใน pipedReader.java ฟังก์ชันรับ (int c) จะถูกเรียกในที่สุด ในเวลานี้สตรีมอินพุตไปป์ไลน์ถูกปิดนั่นคือปิดโดยใช้การอ่านเป็นจริงดังนั้นโยน iOexception ใหม่ ("ปิดท่อ") จะถูกโยนลงไป
เรายังคงแก้ไข "ทดสอบหนึ่ง" เพื่อแก้ปัญหา
การทดลองที่ 2: ดำเนินการต่อเพื่อแก้ไขตัวรับสัญญาณ Java บนพื้นฐานของ "การทดลองที่ 1"

 โมฆะสาธารณะเรียกใช้ () {readMessageOnce (); // readMessagecontinued ();}

ดัดแปลงเป็น

 โมฆะสาธารณะเรียกใช้ () {// readMessageOnce (); READMESSAGECONTINUED ();}

ในเวลานี้โปรแกรมสามารถทำงานได้ตามปกติ ผลการทำงานคือ: