ไปสตรีม

ไปสตรีม

ไลบรารีนี้เป็นเฟรมเวิร์กสำหรับการวิเคราะห์การประมวลผลสตรีม มีไว้เพื่อใช้เป็นไลบรารี่สำหรับโปรแกรม go ที่จำเป็นต้องประมวลผลสตรีมข้อมูลปริมาณมาก

ประกอบด้วยกราฟที่เชื่อมต่อแหล่งที่มากับตัวดำเนินการ 1 ตัวขึ้นไป และสิ้นสุดที่อ่างล้างจาน ผู้ดำเนินการส่งข้อมูลจากที่หนึ่งไปยังอีกที่หนึ่งด้วยช่องทาง go กราฟตัวอย่างในการเข้ารหัสวัตถุให้เร็วคือ:

 var from *util.MemoryBuffer
// fill up from

var to *util.MemoryBuffer

ch := stream.NewOrderedChain()
ch.Add(source.NewNextReaderSource(from))
timingOp, _, dur := timing.NewTimingOp()
ch.Add(timingOp)
ch.Add(compress.NewSnappyEncodeOp())
ch.Add(sink.NewWriterSink(to))

ch.Start()

log.Printf("RES: Compress Snappy.ttRatio %v", float64(to.ByteSize())/float64(from.ByteSize()))
log.Printf("RES: Compress Snappy.ttBuffered Items: %dtItem: %vttook: %vtrate: %dtsize: %Etsize/item: %E", to.Len(), *counter, *dur, int(    float64(*counter)/(*dur).Seconds()), float64(to.ByteSize()), float64(to.ByteSize())/float64(*counter))

ตัวดำเนินการเป็นส่วนประกอบหลักของห่วงโซ่ พวกเขาประมวลผลสิ่งอันดับเพื่อสร้างผลลัพธ์ แหล่งที่มาคือตัวดำเนินการที่ไม่มีเอาต์พุต ซิงก์คือตัวดำเนินการที่ไม่มีอินพุต ตัวดำเนินการใช้ stream.Operator หากต้องใช้อินพุตก็จะใช้ stream.In; ถ้ามันสร้างเอาต์พุตจะใช้ stream.Out

ผู้ทำแผนที่ให้วิธีง่ายๆ ในการติดตั้งตัวดำเนินการ mapper.NewOp() รับฟังก์ชั่นของรูปแบบ func(input stream.Object, out Outputer) ซึ่งประมวลผลอินพุตและเอาต์พุตไปยังวัตถุ Outputer ผู้ทำแผนที่จะถูกขนานโดยอัตโนมัติ เครื่องกำเนิดไฟฟ้าให้วิธีการในการจัดเก็บข้อมูลแบบเธรดภายในเครื่องของผู้ทำแผนที่ผ่านการปิด คุณยังสามารถให้ฟังก์ชันพิเศษแก่ผู้ทำแผนที่ได้ หลังจากที่พวกเขาประมวลผลทูเพิลสุดท้ายเสร็จแล้ว

คุณยังสามารถแบ่งข้อมูลของลูกโซ่ออกเป็นลูกโซ่อื่นๆ ได้ stream.Fanout รับอินพุตและคัดลอกไปยังเครือข่ายอื่น ๆ อีก N แห่ง ผู้จัดจำหน่ายรับอินพุตและวางลงบน 1 ของเชน N ตามฟังก์ชันการทำแผนที่

สร้อยสามารถสั่งหรือสั่งได้ Ordered chains จะรักษาลำดับของสิ่งอันดับจากอินพุตไปยังเอาต์พุต (แม้ว่าตัวดำเนินการยังคงใช้ความขนาน)

การคอมไพล์: ไปสร้าง

การทดสอบ: ไปทดสอบ