บทนำสู่การเขียนโปรแกรมพร้อมกัน: คู่มือสำหรับผู้เริ่มต้น

การเขียนโปรแกรมพร้อมกันคืออะไร? อธิบายง่ายๆ คือ เมื่อคุณทำหลายอย่างพร้อมกัน เพื่อไม่ให้สับสนกับการขนานกัน ภาวะพร้อมกันคือเมื่อมีการรันหลายลำดับของการดำเนินการในช่วงเวลาที่ทับซ้อนกัน ในขอบเขตของการเขียนโปรแกรม การทำงานพร้อมกันเป็นหัวข้อที่ค่อนข้างซับซ้อน การจัดการกับโครงสร้าง เช่น เธรดและการล็อก และการหลีกเลี่ยงปัญหา เช่น สภาพการแข่งขันและการชะงักงัน อาจค่อนข้างยุ่งยาก ทำให้โปรแกรมที่เกิดขึ้นพร้อมกันนั้นเขียนได้ยาก ผ่านการทำงานพร้อมกัน โปรแกรมสามารถออกแบบเป็นกระบวนการอิสระที่ทำงานร่วมกันในองค์ประกอบเฉพาะ โครงสร้างดังกล่าวอาจจะทำขนานกันหรือไม่ก็ได้ อย่างไรก็ตาม การบรรลุโครงสร้างดังกล่าวในโปรแกรมของคุณมีข้อดีหลายประการ

ในบทความนี้ เราจะมาดูโมเดลการทำงานพร้อมกันหลายแบบ วิธีทำให้สำเร็จในภาษาโปรแกรมต่างๆ ที่ออกแบบมาสำหรับการทำงานพร้อมกัน

แบบจำลองสถานะที่ไม่แน่นอนที่ใช้ร่วมกัน

มาดูตัวอย่างง่ายๆ ที่มีตัวนับและสองเธรดที่เพิ่มค่าดังกล่าว โปรแกรมไม่ควรซับซ้อนเกินไป เรามีวัตถุที่มีตัวนับที่เพิ่มขึ้นตามวิธีการที่เพิ่มขึ้น และดึงข้อมูลด้วยวิธีการรับและสองเธรดที่เพิ่ม

 // // Counting.java // public class Counting { public static void main(String[] args) throws InterruptedException { class Counter { int counter = 0; public void increment() { counter++; } public int get() { return counter; } } final Counter counter = new Counter(); class CountingThread extends Thread { public void run() { for (int x = 0; x < 500000; x++) { counter.increment(); } } } CountingThread t1 = new CountingThread(); CountingThread t2 = new CountingThread(); t1.start(); t2.start(); t1.join(); t2.join(); System.out.println(counter.get()); } }

โปรแกรมที่ไร้เดียงสานี้ไม่ได้ไร้เดียงสาอย่างที่เห็นในแวบแรก เมื่อฉันเรียกใช้โปรแกรมนี้หลายครั้ง ฉันได้ผลลัพธ์ที่ต่างออกไป มีสามค่าหลังจากดำเนินการสามครั้งบนแล็ปท็อปของฉัน

 java Counting 553706 java Counting 547818 java Counting 613014

อะไรคือสาเหตุของพฤติกรรมที่คาดเดาไม่ได้นี้? โปรแกรมเพิ่มตัวนับในที่เดียวในวิธีการเพิ่มที่ใช้ตัวนับคำสั่ง++ ถ้าเราดูที่ command byte code เราจะเห็นว่าประกอบด้วยหลายส่วน:

1. อ่านค่าตัวนับจากหน่วยความจำ
2. เพิ่มมูลค่าในท้องถิ่น
3. เก็บค่าตัวนับในหน่วยความจำ

ตอนนี้เราสามารถจินตนาการถึงสิ่งที่อาจผิดพลาดในลำดับนี้ได้ หากเรามีสองเธรดที่เพิ่มตัวนับอย่างอิสระ เราก็อาจมีสถานการณ์นี้:

1. ค่าตัวนับคือ115
2. เธรดแรกอ่านค่าของตัวนับจากหน่วยความจำ (115)
3. เธรดแรกเพิ่มค่าตัวนับในเครื่อง (116)
4. เธรดที่สองอ่านค่าของตัวนับจากหน่วยความจำ (115)
5. เธรดที่สองเพิ่มค่าตัวนับในเครื่อง (116)
6. เธรดที่สองบันทึกค่าตัวนับในเครื่องลงในหน่วยความจำ (116)
7. เธรดแรกบันทึกค่าตัวนับในเครื่องลงในหน่วยความจำ (116)
8. มูลค่าของเคาน์เตอร์คือ116

ในสถานการณ์สมมตินี้ เธรดสองเธรดถูกพันกันเพื่อให้ค่าตัวนับเพิ่มขึ้น 1 แต่ค่าตัวนับควรเพิ่มขึ้น 2 เนื่องจากแต่ละเธรดเพิ่มขึ้น 1 เธรดที่พันกันมีอิทธิพลต่อผลลัพธ์ของโปรแกรม สาเหตุของความไม่แน่นอนของโปรแกรมคือโปรแกรมไม่มีการควบคุมการพันกันของเธรดแต่เป็นระบบปฏิบัติการ ทุกครั้งที่รันโปรแกรม เธรดสามารถพันกันได้แตกต่างกัน ด้วยวิธีนี้ เราจึงแนะนำโปรแกรมที่คาดเดาไม่ได้โดยไม่ได้ตั้งใจ (ไม่ใช่การกำหนด)

เพื่อแก้ไขการคาดเดาไม่ได้โดยไม่ได้ตั้งใจ (ไม่ใช่การกำหนด) โปรแกรมต้องมีการควบคุมการพันเธรด เมื่อเธรดหนึ่งอยู่ในวิธีการเพิ่มเธรดอื่นจะต้องไม่อยู่ในวิธีเดียวกันจนกว่าเธรดแรกจะออกมา ด้วยวิธีนี้เราจะเพิ่มการเข้าถึงวิธีการเป็นอนุกรม

 // // CountingFixed.java // public class CountingFixed { public static main(String[] args) throws InterruptedException { class Counter { int counter = 0; public synchronized void increase() { counter++; } public synchronized int get() { return counter; } } final Counter counter = new Counter(); class CountingThread extends Thread { public void run() { for (int i = 0; i < 500000; i++) { counter.increment(); } } } CountingThread thread1 = new CountingThread(); CountingThread thread2 = new CountingThread(); thread1.start(); thread2.start(); thread1.join(); thread2.join(); System.out.println(counter.get()); } }

อีกวิธีหนึ่งคือการใช้ตัวนับซึ่งสามารถเพิ่มขึ้นในระดับอะตอม ซึ่งหมายความว่าการดำเนินการไม่สามารถแยกออกเป็นหลายการดำเนินการได้ ด้วยวิธีนี้ เราไม่จำเป็นต้องมีบล็อคของโค้ดที่จำเป็นต้องซิงโครไนซ์ Java มีประเภทข้อมูลอะตอมมิกในเนมสเปซ java.util.concurrent.atomic และเราจะใช้ AtomicInteger

 // // CountingBetter.java // import java.util.concurrent.atomic.AtomicInteger; class CountingBetter { public static void main(String[] args) throws InterruptedException { final AtomicInteger counter = new AtomicInteger(0); class CountingThread extends Thread { public viod run() { for (int i = 0; i < 500000; i++) { counter.incrementAndGet(); } } } CountingThread thread1 = new CountingThread(); CountingThread thread2 = new CoutningThread(); thread1.start(); thread2.start(); thread1.join(); thread2.join(); System.out.println(counter.get()); } }

จำนวนเต็มอะตอมมีการดำเนินการที่เราต้องการ ดังนั้นเราจึงสามารถใช้แทนคลาสตัวนับได้ เป็นที่น่าสนใจที่จะสังเกตว่าวิธีการทั้งหมดของ atomicinteger ไม่ได้ใช้การล็อก ดังนั้นจึงไม่มีความเป็นไปได้ที่จะเกิดการชะงักงัน ซึ่งช่วยอำนวยความสะดวกในการออกแบบโปรแกรม

การใช้คำหลักที่ซิงโครไนซ์เพื่อซิงโครไนซ์วิธีการที่สำคัญควรแก้ไขปัญหาทั้งหมดใช่ไหม ลองนึกภาพว่าเรามีสองบัญชีที่สามารถฝาก ถอน และโอนไปยังบัญชีอื่นได้ จะเกิดอะไรขึ้นถ้าเราต้องการโอนเงินจากบัญชีหนึ่งไปยังอีกบัญชีหนึ่งและในทางกลับกัน? มาดูตัวอย่างกัน

 // // Deadlock.java // public class Deadlock { public static void main(String[] args) throws InterruptedException { class Account { int balance = 100; public Account(int balance) { this.balance = balance; } public synchronized void deposit(int amount) { balance += amount; } public synchronized boolean withdraw(int amount) { if (balance >= amount) { balance -= amount; return true; } return false; } public synchronized boolean transfer(Account destination, int amount) { if (balance >= amount) { balance -= amount; synchronized(destination) { destination.balance += amount; }; return true; } return false; } public int getBalance() { return balance; } } final Account bob = new Account(200000); final Account joe = new Account(300000); class FirstTransfer extends Thread { public void run() { for (int i = 0; i < 100000; i++) { bob.transfer(joe, 2); } } } class SecondTransfer extends Thread { public void run() { for (int i = 0; i < 100000; i++) { joe.transfer(bob, 1); } } } FirstTransfer thread1 = new FirstTransfer(); SecondTransfer thread2 = new SecondTransfer(); thread1.start(); thread2.start(); thread1.join(); thread2.join(); System.out.println("Bob's balance: " + bob.getBalance()); System.out.println("Joe's balance: " + joe.getBalance()); } }

เมื่อฉันเรียกใช้โปรแกรมนี้บนแล็ปท็อปมักจะค้าง ทำไมสิ่งนี้ถึงเกิดขึ้น? หากเราสังเกตอย่างใกล้ชิด เราจะเห็นว่าเมื่อเราโอนเงิน เรากำลังเข้าสู่วิธีการโอนที่ซิงโครไนซ์และล็อกการเข้าถึงวิธีการซิงโครไนซ์ทั้งหมดในบัญชีต้นทาง แล้วล็อกบัญชีปลายทางซึ่งล็อกการเข้าถึงวิธีการซิงโครไนซ์ทั้งหมดบนบัญชีนั้น

ลองนึกภาพสถานการณ์ต่อไปนี้:

1. การโอนสายชุดแรกในบัญชีของ Bob ไปยังบัญชีของ Joe
2. โอนสายที่สองในบัญชีของโจไปยังบัญชีของบ๊อบ
3. กระทู้ที่สองลดยอดเงินจากบัญชีโจ้
4. เธรดที่สองไปที่จำนวนเงินฝากในบัญชีของ Bob แต่รอให้เธรดแรกดำเนินการโอนให้เสร็จสิ้น
5. เธรดแรกลดจำนวนเงินจากบัญชีของบ๊อบ
6. เธรดแรกไปที่จำนวนเงินที่ฝากเข้าบัญชีของโจ แต่รอเธรดที่สองเพื่อโอนให้เสร็จสิ้น

ในสถานการณ์สมมตินี้ เธรดหนึ่งกำลังรออีกเธรดหนึ่งเพื่อถ่ายโอนให้เสร็จสิ้น และในทางกลับกัน พวกมันติดกันและโปรแกรมไม่สามารถดำเนินการต่อได้ สิ่งนี้เรียกว่าการหยุดชะงัก เพื่อหลีกเลี่ยงภาวะชะงักงัน จำเป็นต้องล็อคบัญชีในลำดับเดียวกัน ในการแก้ไขโปรแกรม เราจะให้หมายเลขเฉพาะแก่แต่ละบัญชี เพื่อให้เราสามารถล็อคบัญชีในลำดับเดียวกันเมื่อทำการโอนเงิน

 // // DeadlockFixed.java // import java.util.concurrent.atomic.AtomicInteger; public class DeadlockFixed { public static void main(String[] args) throws InterruptedException { final AtomicInteger counter = new AtomicInteger(0); class Account { int balance = 100; int order; public Account(int balance) { this.balance = balance; this.order = counter.getAndIncrement(); } public synchronized void deposit(int amount) { balance += amount; } public synchronized boolean withdraw(int amount) { if (balance >= amount) { balance -= amount; return true; } return false; } public boolean transfer(Account destination, int amount) { Account first; Account second; if (this.order < destination.order) { first = this; second = destination; } else { first = destination; second = this; } synchronized(first) { synchronized(second) { if (balance >= amount) { balance -= amount; destination.balance += amount; return true; } return false; } } } public synchronized int getBalance() { return balance; } } final Account bob = new Account(200000); final Account joe = new Account(300000); class FirstTransfer extends Thread { public void run() { for (int i = 0; i < 100000; i++) { bob.transfer(joe, 2); } } } class SecondTransfer extends Thread { public void run() { for (int i = 0; i < 100000; i++) { joe.transfer(bob, 1); } } } FirstTransfer thread1 = new FirstTransfer(); SecondTransfer thread2 = new SecondTransfer(); thread1.start(); thread2.start(); thread1.join(); thread2.join(); System.out.println("Bob's balance: " + bob.getBalance()); System.out.println("Joe's balance: " + joe.getBalance()); } }

เนื่องจากข้อผิดพลาดดังกล่าวไม่สามารถคาดเดาได้ บางครั้งจึงเกิดขึ้น แต่ไม่เสมอไป และยากที่จะทำซ้ำ หากโปรแกรมทำงานอย่างคาดเดาไม่ได้ มักเกิดจากการทำงานพร้อมกันซึ่งทำให้เกิดการไม่กำหนดขึ้นโดยไม่ได้ตั้งใจ เพื่อหลีกเลี่ยงความไม่ตั้งใจโดยไม่ได้ตั้งใจ เราควรออกแบบโปรแกรมล่วงหน้าโดยคำนึงถึงความเกี่ยวข้องกันทั้งหมด

ตัวอย่างของโปรแกรมที่มีการไม่กำหนดขึ้นโดยไม่ได้ตั้งใจ

 // // NonDeteminism.java // public class NonDeterminism { public static void main(String[] args) throws InterruptedException { class Container { public String value = "Empty"; } final Container container = new Container(); class FastThread extends Thread { public void run() { container.value = "Fast"; } } class SlowThread extends Thread { public void run() { try { Thread.sleep(50); } catch(Exception e) {} container.value = "Slow"; } } FastThread fast = new FastThread(); SlowThread slow = new SlowThread(); fast.start(); slow.start(); fast.join(); slow.join(); System.out.println(container.value); } }

โปรแกรมนี้มีการไม่กำหนดขึ้นโดยไม่ได้ตั้งใจ ค่าสุดท้ายที่ป้อนในคอนเทนเนอร์จะปรากฏขึ้น

 java NonDeterminism Slow

เธรดที่ช้ากว่าจะป้อนค่าในภายหลัง และค่านี้จะถูกพิมพ์ (ช้า) แต่สิ่งนี้ไม่จำเป็นต้องเป็นอย่างนั้น จะเกิดอะไรขึ้นหากคอมพิวเตอร์รันโปรแกรมอื่นที่ต้องการทรัพยากร CPU จำนวนมากพร้อมกัน เราไม่รับประกันว่ามันจะเป็นเธรดที่ช้ากว่าที่ป้อนค่าล่าสุดเพราะถูกควบคุมโดยระบบปฏิบัติการไม่ใช่โปรแกรม เราสามารถมีสถานการณ์ที่โปรแกรมทำงานบนคอมพิวเตอร์เครื่องหนึ่งและอีกเครื่องหนึ่งทำงานแตกต่างกันได้ ข้อผิดพลาดดังกล่าวหายากและทำให้นักพัฒนาต้องปวดหัว ด้วยเหตุผลทั้งหมดนี้ โมเดลการทำงานพร้อมกันนี้จึงเป็นเรื่องยากมากที่จะทำสิ่งที่ถูกต้อง

วิธีการใช้งาน

ความเท่าเทียม

ลองดูรุ่นอื่นที่ใช้ภาษาที่ใช้งานได้ ตัวอย่างเช่น เราจะใช้ Clojure ซึ่งสามารถตีความได้โดยใช้เครื่องมือ Leiningen Clojure เป็นภาษาที่น่าสนใจมาก รองรับการทำงานพร้อมกันได้ดี โมเดลการทำงานพร้อมกันก่อนหน้านี้มีสถานะที่ไม่แน่นอนที่ใช้ร่วมกัน คลาสที่เราใช้ยังสามารถมีสถานะซ่อนเร้นที่กลายพันธุ์ที่เราไม่รู้ เพราะมันไม่ชัดเจนจาก API ของคลาสเหล่านั้น ดังที่เราได้เห็น โมเดลนี้อาจทำให้เกิดการไม่กำหนดขึ้นและการหยุดชะงักโดยไม่ได้ตั้งใจ หากเราไม่ระวัง ภาษาที่ใช้งานได้มีประเภทข้อมูลที่ไม่มีการกลายพันธุ์ จึงสามารถแบ่งปันได้อย่างปลอดภัยโดยไม่มีความเสี่ยงที่จะเปลี่ยนแปลง ฟังก์ชันมีคุณสมบัติเช่นเดียวกับชนิดข้อมูลอื่นๆ ฟังก์ชันต่างๆ สามารถสร้างได้ระหว่างการทำงานของโปรแกรม และส่งผ่านเป็นพารามิเตอร์ไปยังฟังก์ชันอื่นหรือส่งกลับจากการเรียกใช้ฟังก์ชัน

พื้นฐานเบื้องต้นสำหรับการเขียนโปรแกรมพร้อมกันคืออนาคตและสัญญา Future รันบล็อกของโค้ดในเธรดอื่นและส่งคืนอ็อบเจ็กต์สำหรับค่าในอนาคตที่จะถูกป้อนเมื่อบล็อกถูกดำเนินการ

 ; ; future.clj ; (let [a (future (println "Started A") (Thread/sleep 1000) (println "Finished A") (+ 1 2)) b (future (println "Started B") (Thread/sleep 2000) (println "Finished B") (+ 3 4))] (println "Waiting for futures") (+ @a @b))

เมื่อฉันรันสคริปต์นี้ ผลลัพธ์คือ:

 Started A Started B Waiting for futures Finished A Finished B 10

ในตัวอย่างนี้ เรามีสองช่วงตึกในอนาคตที่ดำเนินการอย่างอิสระ โปรแกรมจะบล็อกเมื่ออ่านค่าจากวัตถุในอนาคตที่ยังไม่พร้อมใช้งานเท่านั้น ในกรณีของเรา กำลังรอผลรวมของบล็อคในอนาคตทั้งสอง พฤติกรรมสามารถคาดเดาได้ (กำหนดได้) และจะให้ผลลัพธ์แบบเดียวกันเสมอเนื่องจากไม่มีสถานะที่ไม่แน่นอนที่ใช้ร่วมกัน

ดั้งเดิมอีกอย่างหนึ่งที่ใช้สำหรับการทำงานพร้อมกันคือสัญญา Promise เป็นคอนเทนเนอร์ที่สามารถใส่ค่าได้เพียงครั้งเดียว เมื่ออ่านคำสัญญา เธรดจะรอจนกว่าค่าของสัญญาจะเต็ม

 ; ; promise.clj ; (def result (promise)) (future (println "The result is: " @result)) (Thread/sleep 2000) (deliver result 42)

ในตัวอย่างนี้ อนาคตจะรอพิมพ์ผลลัพธ์ตราบเท่าที่สัญญาว่าจะไม่บันทึกค่า หลังจากสองวินาทีในสัญญาจะเก็บค่า 42 เพื่อพิมพ์ในเธรดในอนาคต การใช้คำสัญญาอาจนำไปสู่การชะงักงันเมื่อเทียบกับอนาคต ดังนั้นโปรดใช้ความระมัดระวังเมื่อใช้สัญญา

 ; ; promise-deadlock.clj ; (def promise-result (promise)) (def future-result (future (println "The result is: " + @promise-result) 13)) (println "Future result is: " @future-result) (deliver result 42)

ในตัวอย่างนี้ เรากำลังใช้ผลลัพธ์ของอนาคตและผลลัพธ์ของสัญญา ลำดับของการตั้งค่าและการอ่านค่าคือ เธรดหลักกำลังรอค่าจากเธรดในอนาคต และเธรดในอนาคตกำลังรอค่าจากเธรดหลัก ลักษณะการทำงานนี้จะคาดเดาได้ (กำหนดได้) และจะเล่นทุกครั้งที่โปรแกรมดำเนินการ ซึ่งทำให้ค้นหาและขจัดข้อผิดพลาดได้ง่ายขึ้น

การใช้อนาคตช่วยให้โปรแกรมดำเนินการฝึกต่อไปได้จนกว่าจะต้องการผลลัพธ์จากการดำเนินการในอนาคต ส่งผลให้การทำงานของโปรแกรมเร็วขึ้น หากคุณมีโปรเซสเซอร์หลายตัวในอนาคต คุณสามารถดำเนินการขนานกันของโปรแกรมที่มีพฤติกรรมคาดการณ์ได้ (กำหนดขึ้นเองได้) (แต่ละครั้งจะให้ผลลัพธ์เหมือนกัน) ด้วยวิธีนี้เราจะใช้ประโยชน์จากพลังของคอมพิวเตอร์ได้ดีขึ้น

 ; ; fibonacci.clj ; (defn fibonacci[a] (if (<= a 2) 1 (+ (fibonacci (- a 1)) (fibonacci (- a 2))))) (println "Start serial calculation") (time (println "The result is: " (+ (fibonacci 36) (fibonacci 36)))) (println "Start parallel calculation") (defn parallel-fibonacci[] (def result-1 (future (fibonacci 36))) (def result-2 (future (fibonacci 36))) (+ @result-1 @result-2)) (time (println "The result is: " (parallel-fibonacci)))

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

ผลลัพธ์ของการดำเนินการสคริปต์นี้บนแล็ปท็อปของฉัน:

 Start serial calculation The result is: 29860704 "Elapsed time: 2568.816524 msecs" Start parallel calculation The result is: 29860704 "Elapsed time: 1216.991448 msecs"

พร้อมกัน

เพื่อรองรับการทำงานพร้อมกันและความไม่แน่นอนในภาษาการเขียนโปรแกรม Clojure เราต้องใช้ชนิดข้อมูลที่เป็นตัวแปรเพื่อให้เธรดอื่นเห็นการเปลี่ยนแปลง ชนิดข้อมูลตัวแปรที่ง่ายที่สุดคืออะตอม อะตอมเป็นภาชนะที่มีค่าซึ่งสามารถแทนที่ด้วยค่าอื่นได้เสมอ ค่านี้สามารถแทนที่ได้ด้วยการป้อนค่าใหม่หรือโดยการเรียกใช้ฟังก์ชันที่ใช้ค่าเดิมและส่งกลับค่าใหม่ที่ใช้บ่อยกว่า เป็นที่น่าสนใจที่อะตอมถูกนำไปใช้โดยไม่ต้องล็อคและปลอดภัยที่จะใช้ในเธรด ซึ่งหมายความว่าเป็นไปไม่ได้ที่จะถึงขั้นชะงักงัน ภายใน atom ใช้ไลบรารี java.util.concurrent.AtomicReference มาดูตัวอย่างเคาน์เตอร์ที่ใช้อะตอมกัน

 ; ; atom-counter.clj ; (def counter (atom 0)) (def attempts (atom 0)) (defn counter-increases[] (dotimes [cnt 500000] (swap! counter (fn [counter] (swap! attempts inc) ; side effect DO NOT DO THIS (inc counter))))) (def first-future (future (counter-increases))) (def second-future (future (counter-increases))) ; Wait for futures to complete @first-future @second-future ; Print value of the counter (println "The counter is: " @counter) (println "Number of attempts: " @attempts)

ผลลัพธ์ของการดำเนินการสคริปต์บนแล็ปท็อปของฉัน:

 The counter is: 1000000 Number of attempts: 1680212

ในตัวอย่างนี้ เราใช้อะตอมที่มีค่าของตัวนับ ตัวนับเพิ่มขึ้นด้วย (swap! counter inc) ฟังก์ชันสลับทำงานดังนี้: 1. ใช้ค่าตัวนับและรักษาไว้ 2. สำหรับการเรียกค่านี้ฟังก์ชันที่กำหนดที่คำนวณค่าใหม่ 3. เพื่อบันทึกค่าใหม่ จะใช้การดำเนินการปรมาณูที่ตรวจสอบว่าค่าเก่ามีการเปลี่ยนแปลง 3a หรือไม่ หากค่าไม่เปลี่ยนแปลง ให้ป้อนค่าใหม่ 3b หากค่ามีการเปลี่ยนแปลงในระหว่างนี้ให้ไปที่ขั้นตอนที่ 1 เราจะเห็นว่าสามารถเรียกใช้ฟังก์ชันได้อีกครั้งหากค่ามีการเปลี่ยนแปลงในระหว่างนี้ ค่านี้สามารถเปลี่ยนแปลงได้จากเธรดอื่นเท่านั้น ดังนั้นจึงจำเป็นที่ฟังก์ชันที่คำนวณค่าใหม่จะต้องไม่มีผลข้างเคียง จึงไม่คำนึงว่าจะถูกเรียกซ้ำหลายครั้งหรือไม่ ข้อจำกัดหนึ่งของอะตอมคือการซิงโครไนซ์การเปลี่ยนแปลงเป็นค่าเดียว

 ; ; atom-acocunts.clj ; (def bob (atom 200000)) (def joe (atom 300000)) (def inconsistencies (atom 0)) (defn transfer [source destination amount] (if (not= (+ @bob @joe) 500000) (swap! inconsistencies inc)) (swap! source - amount) (swap! destination + amount)) (defn first-transfer [] (dotimes [cnt 100000] (transfer bob joe 2))) (defn second-transfer [] (dotimes [cnt 100000] (transfer joe bob 1))) (def first-future (future (first-transfer))) (def second-future (future (second-transfer))) @first-future @second-future (println "Bob has in account: " @bob) (println "Joe has in account: " @joe) (println "Inconsistencies while transfer: " @inconsistencies)

เมื่อฉันรันสคริปต์นี้ ฉันจะได้รับ:

 Bob has in account: 100000 Joe has in account: 400000 Inconsistencies while transfer: 36525

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

1. ใส่ค่ามากขึ้นในหนึ่งอะตอม
2. ใช้ข้อมูลอ้างอิงและหน่วยความจำธุรกรรมของซอฟต์แวร์ ตามที่เราจะได้เห็นในภายหลัง

 ; ; atom-accounts-fixed.clj ; (def accounts (atom {:bob 200000, :joe 300000})) (def inconsistencies (atom 0)) (defn transfer [source destination amount] (let [deref-accounts @accounts] (if (not= (+ (get deref-accounts :bob) (get deref-accounts :joe)) 500000) (swap! inconsistencies inc)) (swap! accounts (fn [accs] (update (update accs source - amount) destination + amount))))) (defn first-transfer [] (dotimes [cnt 100000] (transfer :bob :joe 2))) (defn second-transfer [] (dotimes [cnt 100000] (transfer :joe :bob 1))) (def first-future (future (first-transfer))) (def second-future (future (second-transfer))) @first-future @second-future (println "Bob has in account: " (get @accounts :bob)) (println "Joe has in account: " (get @accounts :joe)) (println "Inconsistencies while transfer: " @inconsistencies)

เมื่อฉันเรียกใช้สคริปต์นี้บนคอมพิวเตอร์ของฉัน ฉันจะได้รับ:

 Bob has in account: 100000 Joe has in account: 400000 Inconsistencies while transfer: 0

ในตัวอย่าง มีการแก้ไขการประสานงานเพื่อให้เราเพิ่มมูลค่าโดยใช้แผนที่ เมื่อเราโอนเงินจากบัญชี เราจะเปลี่ยนทุกบัญชี ณ เวลานั้น เพื่อไม่ให้เกิดยอดเงินไม่เท่ากัน

ชนิดข้อมูลตัวแปรถัดไปคือเอเจนต์ เอเจนต์ทำงานเหมือนอะตอมเท่านั้นโดยที่ฟังก์ชันที่เปลี่ยนค่าจะถูกดำเนินการในเธรดอื่น ดังนั้นจึงต้องใช้เวลาระยะหนึ่งกว่าที่การเปลี่ยนแปลงจะมองเห็นได้ ดังนั้น เมื่ออ่านค่าของเอเจนต์ จำเป็นต้องเรียกใช้ฟังก์ชันที่รอจนกว่าฟังก์ชันทั้งหมดที่เปลี่ยนค่าของเอเจนต์จะถูกดำเนินการ ต่างจากฟังก์ชันอะตอมที่เปลี่ยนค่าถูกเรียกเพียงครั้งเดียว จึงสามารถมีผลข้างเคียงได้ ประเภทนี้ยังสามารถซิงโครไนซ์หนึ่งค่าและไม่สามารถชะงักงันได้

 ; ; agent-counter.clj ; (def counter (agent 0)) (def attempts (atom 0)) (defn counter-increases[] (dotimes [cnt 500000] (send counter (fn [counter] (swap! attempts inc) (inc counter))))) (def first-future (future (counter-increases))) (def second-future (future (counter-increases))) ; wait for futures to complete @first-future @second-future ; wait for counter to be finished with updating (await counter) ; print the value of the counter (println "The counter is: " @counter) (println "Number of attempts: " @attempts)

เมื่อฉันเรียกใช้สคริปต์นี้บนแล็ปท็อป ฉันจะได้รับ:

 The counter is: 1000000 Number of attempts: 1000000

ตัวอย่างนี้เหมือนกับการใช้งานตัวนับกับอะตอม ข้อแตกต่างเพียงอย่างเดียวคือที่นี่เรากำลังรอให้การเปลี่ยนแปลงเอเจนต์ทั้งหมดเสร็จสิ้นก่อนที่จะอ่านค่าสุดท้ายโดยใช้การรอ

ชนิดข้อมูลตัวแปรสุดท้ายคือข้อมูลอ้างอิง การอ้างอิงสามารถซิงโครไนซ์การเปลี่ยนแปลงกับค่าหลายค่าต่างจากอะตอม การดำเนินการอ้างอิงแต่ละรายการควรอยู่ในธุรกรรมโดยใช้ dosync วิธีการเปลี่ยนข้อมูลนี้เรียกว่าหน่วยความจำทรานแซคชันซอฟต์แวร์หรือ STM แบบย่อ มาดูตัวอย่างการโอนเงินในบัญชีกัน

 ; ; stm-accounts.clj ; (def bob (ref 200000)) (def joe (ref 300000)) (def inconsistencies (atom 0)) (def attempts (atom 0)) (def transfers (agent 0)) (defn transfer [source destination amount] (dosync (swap! attempts inc) ; side effect DO NOT DO THIS (send transfers inc) (when (not= (+ @bob @joe) 500000) (swap! inconsistencies inc)) ; side effect DO NOT DO THIS (alter source - amount) (alter destination + amount))) (defn first-transfer [] (dotimes [cnt 100000] (transfer bob joe 2))) (defn second-transfer [] (dotimes [cnt 100000] (transfer joe bob 1))) (def first-future (future (first-transfer))) (def second-future (future (second-transfer))) @first-future @second-future (await transfers) (println "Bob has in account: " @bob) (println "Joe has in account: " @joe) (println "Inconsistencies while transfer: " @inconsistencies) (println "Attempts: " @attempts) (println "Transfers: " @transfers)

เมื่อฉันเรียกใช้สคริปต์นี้ ฉันจะได้รับ:

 Bob has in account: 100000 Joe has in account: 400000 Inconsistencies while transfer: 0 Attempts: 330841 Transfers: 200000

น่าสนใจ มีการพยายามมากกว่าจำนวนธุรกรรมที่ทำ เนื่องจาก STM ไม่ได้ใช้การล็อก ดังนั้นหากมีข้อขัดแย้ง (เช่น สองเธรดที่พยายามเปลี่ยนค่าเดียวกัน) ธุรกรรมจะถูกดำเนินการอีกครั้ง ด้วยเหตุนี้การทำธุรกรรมจึงไม่ควรมีผลข้างเคียง เราจะเห็นได้ว่าตัวแทนที่มูลค่าเปลี่ยนแปลงภายในธุรกรรมมีพฤติกรรมที่คาดเดาได้ ฟังก์ชันที่เปลี่ยนค่าของตัวแทนจะได้รับการประเมินหลายครั้งตามที่มีธุรกรรม เหตุผลก็คือตัวแทนรับรู้ธุรกรรม หากการทำธุรกรรมต้องมีผลข้างเคียงก็ควรนำไปใช้งานภายในตัวแทน ด้วยวิธีนี้ โปรแกรมจะมีพฤติกรรมที่คาดเดาได้ คุณอาจคิดว่าคุณควรใช้ STM เสมอ แต่โปรแกรมเมอร์ที่มีประสบการณ์มักจะใช้อะตอมเพราะอะตอมนั้นง่ายกว่าและเร็วกว่า STM แน่นอนว่าถ้าเป็นไปได้ที่จะสร้างโปรแกรมในลักษณะนั้น หากคุณมีผลข้างเคียง ก็ไม่มีทางเลือกอื่นนอกจากการใช้ STM และตัวแทน

นักแสดง นายแบบ

แบบจำลองการทำงานพร้อมกันต่อไปนี้เป็นแบบจำลองนักแสดง หลักการของรุ่นนี้คล้ายกับโลกแห่งความจริง หากเราทำข้อตกลงเพื่อสร้างบางสิ่งกับผู้คนจำนวนมาก เช่น อาคาร ผู้ชายทุกคนที่สถานที่ก่อสร้างจะมีบทบาทของตนเอง ฝูงชนจำนวนมากถูกควบคุมโดยผู้บังคับบัญชา หากคนงานได้รับบาดเจ็บจากการทำงาน หัวหน้างานจะมอบหมายงานของผู้บาดเจ็บให้กับผู้อื่นที่มีอยู่ หากจำเป็นเขาอาจนำคนใหม่ไปยังไซต์ บนไซต์เรามีผู้คนที่ทำงานพร้อมกันมากขึ้น (พร้อม ๆ กัน) แต่ยังพูดคุยกันเพื่อซิงโครไนซ์ หากเราใส่งานในไซต์ก่อสร้างลงในโปรแกรม ทุกคนก็จะเป็นนักแสดงที่มีสถานะและดำเนินการตามกระบวนการของตนเอง และการพูดคุยก็จะถูกแทนที่ด้วยข้อความ ภาษาโปรแกรมยอดนิยมที่ใช้โมเดลนี้คือ Erlang ภาษาที่น่าสนใจนี้มีประเภทข้อมูลและฟังก์ชันที่ไม่เปลี่ยนรูปซึ่งมีคุณสมบัติเหมือนกับประเภทข้อมูลอื่นๆ ฟังก์ชันต่างๆ สามารถสร้างได้ระหว่างการทำงานของโปรแกรม และส่งผ่านเป็นอาร์กิวเมนต์ไปยังฟังก์ชันอื่น หรือส่งกลับจากการเรียกใช้ฟังก์ชัน ฉันจะยกตัวอย่างในภาษา Elixir ที่ใช้เครื่องเสมือน Erlang ดังนั้นฉันจะมีรูปแบบการเขียนโปรแกรมเดียวกันกับ Erlang เพียงรูปแบบที่แตกต่างกัน พื้นฐานที่สำคัญที่สุดสามประการใน Elixir คือการวางไข่ ส่ง และรับ วางไข่เรียกใช้ฟังก์ชันในกระบวนการใหม่ ส่งข้อความไปยังกระบวนการ และรับข้อความที่ส่งไปยังกระบวนการปัจจุบัน

ตัวอย่างแรกกับโมเดลนักแสดงจะตอบโต้เพิ่มขึ้นพร้อมกัน ในการสร้างโปรแกรมด้วยโมเดลนี้ จำเป็นต้องทำให้นักแสดงมีค่าของตัวนับและรับข้อความเพื่อตั้งค่าและรับค่าของตัวนับและมีนักแสดงสองคนที่จะเพิ่มมูลค่าของตัวนับพร้อม ๆ กัน

 # # Counting.exs # defmodule Counting do def counter(value) do receive do {:get, sender} -> send sender, {:counter, value} counter value {:set, new_value} -> counter(new_value) end end def counting(sender, counter, times) do if times > 0 do send counter, {:get, self} receive do {:counter, value} -> send counter, {:set, value + 1} end counting(sender, counter, times - 1) else send sender, {:done, self} end end end counter = spawn fn -> Counting.counter 0 end IO.puts "Starting counting processes" this = self counting1 = spawn fn -> IO.puts "Counting A started" Counting.counting this, counter, 500_000 IO.puts "Counting A finished" end counting2 = spawn fn -> IO.puts "Counting B started" Counting.counting this, counter, 500_000 IO.puts "Counting B finished" end IO.puts "Waiting for counting to be done" receive do {:done, ^counting1} -> nil end receive do {:done, ^counting2} -> nil end send counter, {:get, self} receive do {:counter, value} -> IO.puts "Counter is: #{value}" end

เมื่อฉันรันตัวอย่างนี้ ฉันจะได้รับ:

 Starting counting processes Counting A started Waiting for counting to be done Counting B started Counting A finished Counting B finished Counter is: 516827

เราจะเห็นได้ว่าในที่สุดตัวนับคือ 516827 และไม่ใช่ 1000000 อย่างที่เราคาดไว้ เมื่อฉันรันสคริปต์ในครั้งต่อไป ฉันได้รับ 511010 สาเหตุของพฤติกรรมนี้คือตัวนับได้รับข้อความสองข้อความ: ดึงค่าปัจจุบันและตั้งค่าใหม่ ในการเพิ่มตัวนับ โปรแกรมต้องได้รับค่าปัจจุบัน เพิ่มขึ้น 1 และตั้งค่าที่เพิ่มขึ้น สองกระบวนการอ่านและเขียนค่าของตัวนับพร้อมกันโดยใช้ข้อความที่ส่งไปยังกระบวนการตัวนับ ลำดับของข้อความที่ตัวนับจะได้รับนั้นคาดเดาไม่ได้ และโปรแกรมไม่สามารถควบคุมได้ เราสามารถจินตนาการถึงสถานการณ์นี้:

1. ค่าตัวนับคือ115
2. กระบวนการ A อ่านค่าของตัวนับ (115)
3. กระบวนการ B อ่านค่าของตัวนับ (115)
4. กระบวนการ B เพิ่มมูลค่าในเครื่อง (116)
5. กระบวนการ B ตั้งค่าที่เพิ่มขึ้นเป็นตัวนับ (116)
6. กระบวนการ A เพิ่มมูลค่าของตัวนับ (116)
7. กระบวนการ A ตั้งค่าที่เพิ่มขึ้นให้กับตัวนับ (116)
8. ค่าตัวนับคือ116

ถ้าเราดูที่สถานการณ์สมมติ สองกระบวนการจะเพิ่มตัวนับ 1 และตัวนับจะเพิ่มขึ้นในตอนท้าย 1 และไม่ใช่ 2 การพันกันดังกล่าวสามารถเกิดขึ้นได้ในจำนวนที่คาดเดาไม่ได้ ดังนั้นมูลค่าของตัวนับจึงคาดเดาไม่ได้ เพื่อป้องกันพฤติกรรมนี้ การดำเนินการเพิ่มต้องทำโดยข้อความเดียว

 # # CountingFixed.exs # defmodule Counting do def counter(value) do receive do :increase -> counter(value + 1) {:get, sender} -> send sender, {:counter, value} counter value end end def counting(sender, counter, times) do if times > 0 do send counter, :increase counting(sender, counter, times - 1) else send sender, {:done, self} end end end counter = spawn fn -> Counting.counter 0 end IO.puts "Starting counting processes" this = self counting1 = spawn fn -> IO.puts "Counting A started" Counting.counting this, counter, 500_000 IO.puts "Counting A finished" end counting2 = spawn fn -> IO.puts "Counting B started" Counting.counting this, counter, 500_000 IO.puts "Counting B finished" end IO.puts "Waiting for counting to be done" receive do {:done, ^counting1} -> nil end receive do {:done, ^counting2} -> nil end send counter, {:get, self} receive do {:counter, value} -> IO.puts "Counter is: #{value}" end

เมื่อเรียกใช้สคริปต์นี้ ฉันจะได้รับ:

 Starting counting processes Counting A started Waiting for counting to be done Counting B started Counting A finished Counting B finished Counter is: 1000000

เราจะเห็นได้ว่าตัวนับมีค่าที่ถูกต้อง สาเหตุของพฤติกรรมที่คาดการณ์ได้ (กำหนดขึ้นเองได้) คือ ค่าของตัวนับเพิ่มขึ้นหนึ่งข้อความ ดังนั้นลำดับของข้อความที่จะเพิ่มตัวนับจะไม่ส่งผลต่อค่าสุดท้าย การทำงานกับโมเดลนักแสดง เราต้องให้ความสนใจกับวิธีที่ข้อความสามารถเชื่อมโยงและออกแบบข้อความและการดำเนินการอย่างระมัดระวังกับข้อความเพื่อหลีกเลี่ยงความคาดเดาไม่ได้โดยไม่ได้ตั้งใจ (ไม่ใช่การกำหนด)

เราจะโอนเงินระหว่างสองบัญชีกับรุ่นนี้ได้อย่างไร?

 # # Accounts.exs # defmodule Accounts do def accounts(state) do receive do {:transfer, source, destination, amount} -> accounts %{state | source => state[source] - amount , destination => state[destination] + amount} {:amounts, accounts, sender } -> send sender, {:amounts, for account <- accounts do {account, state[account]} end} accounts(state) end end def transfer(sender, accounts, source, destination, amount, times, inconsistencies) do if times > 0 do send accounts, {:amounts, [source, destination], self} receive do {:amounts, amounts} -> if amounts[source] + amounts[destination] != 500_000 do Agent.update(inconsistencies, fn value -> value + 1 end) end end send accounts, {:transfer, source, destination, amount} transfer(sender, accounts, source, destination, amount, times - 1, inconsistencies) else send sender, {:done, self} end end end accounts = spawn fn -> Accounts.accounts(%{bob: 200_000, joe: 300_000 }) end {:ok, inconsistencies} = Agent.start(fn -> 0 end) this = self transfer1 = spawn fn -> IO.puts "Transfer A started" Accounts.transfer(this, accounts, :bob, :joe, 2, 100_000, inconsistencies) IO.puts "Transfer A finished" end transfer2 = spawn fn -> IO.puts "Transfer B started" Accounts.transfer(this, accounts, :joe, :bob, 1, 100_000, inconsistencies) IO.puts "Transfer B finished" end IO.puts "Waiting for transfers to be done" receive do {:done, ^transfer1} -> nil end receive do {:done, ^transfer2} -> nil end send accounts, {:amounts, [:bob, :joe], self} receive do {:amounts, amounts} -> IO.puts "Bob has in account: #{amounts[:bob]}" IO.puts "Joe has in account: #{amounts[:joe]}" IO.puts "Inconsistencies while transfer: #{Agent.get(inconsistencies, fn x -> x end)}" end

เมื่อฉันเรียกใช้สคริปต์นี้ ฉันจะได้รับ:

 Waiting for transfers to be done Transfer A started Transfer B started Transfer B finished Transfer A finished Bob has in account: 100000 Joe has in account: 400000 Inconsistencies while transfer: 0

เราจะเห็นได้ว่าการโอนเงินทำงานโดยไม่มีความไม่สอดคล้องกัน เนื่องจากเราได้เลือกการโอนเงินเพื่อโอนเงินและจำนวนข้อความเพื่อให้ได้มูลค่าของบัญชีซึ่งทำให้เราสามารถคาดการณ์พฤติกรรมของโปรแกรมได้ ทุกครั้งที่เราโอนเงิน จำนวนเงินทั้งหมดควรเท่ากันทุกเมื่อ

โมเดลนักแสดงอาจทำให้เกิดการล็อกและทำให้หยุดชะงักได้ ดังนั้นโปรดใช้ความระมัดระวังเมื่อออกแบบโปรแกรม สคริปต์ต่อไปนี้แสดงวิธีที่คุณสามารถจำลองสถานการณ์การล็อกและการชะงักงัน

 # # Deadlock.exs # defmodule Lock do def loop(state) do receive do {:lock, sender} -> case state do [] -> send sender, :locked loop([sender]) _ -> loop(state ++ [sender]) end {:unlock, sender} -> case state do [] -> loop(state) [^sender | []] -> loop([]) [^sender | [next | tail]] -> send next, :locked loop([next | tail]) _ -> loop(state) end end end def lock(pid) do send pid, {:lock, self} receive do :locked -> nil # This will block until we receive message end end def unlock(pid) do send pid, {:unlock, self} end def locking(first, second, times) do if times > 0 do lock(first) lock(second) unlock(second) unlock(first) locking(first, second, times - 1) end end end a_lock = spawn fn -> Lock.loop([]) end b_lock = spawn fn -> Lock.loop([]) end this = self IO.puts "Locking A, B started" spawn fn -> Lock.locking(a_lock, b_lock, 1_000) IO.puts "Locking A, B finished" send this, :done end IO.puts "Locking B, A started" spawn fn -> Lock.locking(b_lock, a_lock, 1_000) IO.puts "Locking B, A finished" send this, :done end IO.puts "Waiting for locking to be done" receive do :done -> nil end receive do :done -> nil End

When I run this script on my laptop I get:

 Locking A, B started Locking B, A started Waiting for locking to be done

From the output we can see that the processes that lock A and B are stuck. This happens because the first process waits for the second process to release B while second process waiting first process to release A. They are waiting for each other and are stuck forever. To avoid this locking, order should always be the same, or design a program so that it doesn't use lock (meaning that it doesn't wait for a specific message). The following listing always locks first A then B.

 # # Deadlock fixed # defmodule Lock do def loop(state) do receive do {:lock, sender} -> case state do [] -> send sender, :locked loop([sender]) _ -> loop(state ++ [sender]) end {:unlock, sender} -> case state do [] -> loop(state) [^sender | []] -> loop([]) [^sender | [next | tail]] -> send next, :locked loop([next | tail]) _ -> loop(state) end end end def lock(pid) do send pid, {:lock, self} receive do :locked -> nil # This will block until we receive message end end def unlock(pid) do send pid, {:unlock, self} end def locking(first, second, times) do if times > 0 do lock(first) lock(second) unlock(second) unlock(first) locking(first, second, times - 1) end end end a_lock = spawn fn -> Lock.loop([]) end b_lock = spawn fn -> Lock.loop([]) end this = self IO.puts "Locking A, B started" spawn fn -> Lock.locking(a_lock, b_lock, 1_000) IO.puts "Locking A, B finished" send this, :done end IO.puts "Locking A, B started" spawn fn -> Lock.locking(a_lock, b_lock, 1_000) IO.puts "Locking A, B finished" send this, :done end IO.puts "Waiting for locking to be done" receive do :done -> nil end receive do :done -> nil End

When I run this script on my laptop I get:

 Locking A, B started Locking A, B started Waiting for locking to be done Locking A, B finished Locking A, B finished

And now, there is no longer a deadlock.

สรุป

As an introduction to concurrent programming, we have covered a few concurrency models. We haven't covered all models, as this article would be too big. Just to name a few, channels and reactive streams are some of the other popularly used concurrency models. Channels and reactive streams have many similarities with the actor model. All of them transmit messages, but many threads can receive messages from one channel, and reactive streams transmit messages in one direction to form directed graph that receive messages from one end and send messages from the other end as a result of the processing.

Shared mutable state models can easily go wrong if we don't think ahead. It has problems of race condition and deadlock. If we have a choice between different concurrent programming models, it would be easier to implement and maintain but otherwise we have to be very careful what we do.

The functional way is a lot easier to reason about and implement. It cannot have deadlock. This model may have worse performance than shared mutable state model, but a program that works is always faster than one that does not work.

Actor model is a good choice for concurrent programming. Although there are problems of race condition and deadlock, they can happen less than in shared mutable state model since the only way for processes to communicate is via messages. With good message design between processes, that can be avoided. If a problem occurs it is then in the order or meaning of messages in communication between the processes and you know where to look.

I hope this article has given you some insight to what concurrent programming is and how it gives structure to the programs you write.