ดีกว่ากัน: Java และคลาสเธรด ตอนที่ VI — ยิงออกไป!

การแนะนำ

หัวข้อเป็นสิ่งที่น่าสนใจ ในการตรวจสอบที่ผ่านมา เราดูเครื่องมือที่มีอยู่บางส่วนสำหรับการนำมัลติเธรดไปใช้ มาดูกันว่ามีอะไรน่าสนใจอีกบ้างที่เราสามารถทำได้ ณ จุดนี้เรารู้มาก ตัวอย่างเช่น จาก " Better together: Java and the Thread class. Part I — Threads ofexecution ", เรารู้ว่าคลาส Thread เป็นตัวแทนของเธรดของการดำเนินการ เรารู้ว่าเธรดทำงานบางอย่าง หากเราต้องการให้งานของเราสามารถทำได้runเราจะต้องทำเครื่องหมายเธรดRunnableด้วย เพื่อให้จำไว้ เราสามารถใช้Tutorialspoint Online Java Compiler :

public static void main(String[] args){
	Runnable task = () -> {
 		Thread thread = Thread.currentThread();
		System.out.println("Hello from " + thread.getName());
	};
	Thread thread = new Thread(task);
	thread.start();
}

เรายังรู้ว่าเรามีสิ่งที่เรียกว่าล็อค เราได้เรียนรู้เกี่ยวกับสิ่งนี้ใน " Better together: Java and the Thread class. Part II — Synchronization . หากเธรดหนึ่งได้รับการล็อค เธรดอื่นที่พยายามรับการล็อคจะถูกบังคับให้รอให้ปลดล็อค:

import java.util.concurrent.locks.*;

public class HelloWorld{
	public static void main(String []args){
		Lock lock = new ReentrantLock();
		Runnable task = () -> {
			lock.lock();
			Thread thread = Thread.currentThread();
			System.out.println("Hello from " + thread.getName());
			lock.unlock();
		};
		Thread thread = new Thread(task);
		thread.start();
	}
}

ฉันคิดว่าถึงเวลาแล้วที่จะพูดคุยเกี่ยวกับสิ่งอื่น ๆ ที่น่าสนใจที่เราสามารถทำได้

สัญญาณ

วิธีที่ง่ายที่สุดในการควบคุมจำนวนเธรดที่สามารถทำงานได้พร้อมกันคือสัญญาณ มันเหมือนกับสัญญาณรถไฟ สีเขียว หมายถึง ดำเนินการต่อ สีแดงหมายถึงรอ รออะไรจากสัญญาณ? เข้าถึง. ในการเข้าถึง เราต้องได้รับมัน และเมื่อไม่ต้องการการเข้าถึงอีกต่อไป เราจะต้องให้หรือปล่อยมันไป ลองดูวิธีการทำงาน เราจำเป็นต้องนำเข้าjava.util.concurrent.Semaphoreคลาส ตัวอย่าง:

public static void main(String[] args) throws InterruptedException {
	Semaphore semaphore = new Semaphore(0);
	Runnable task = () -> {
		try {
			semaphore.acquire();
			System.out.println("Finished");
			semaphore.release();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	};
	new Thread(task).start();
	Thread.sleep(5000);
	semaphore.release(1);
}

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

สลักนับถอยหลัง

กลไกต่อไปCountDownLatchคือ ไม่น่าแปลกใจเลยที่นี่คือสลักที่มีการนับถอยหลัง ที่นี่เราต้องการคำสั่งนำเข้าที่เหมาะสมสำหรับjava.util.concurrent.CountDownLatchชั้นเรียน มันเหมือนกับการแข่งขันวิ่งด้วยเท้าที่ทุกคนมารวมกันที่เส้นสตาร์ท และเมื่อทุกคนพร้อมทุกคนจะได้รับสัญญาณสตาร์ทพร้อมกันและสตาร์ทพร้อมกัน ตัวอย่าง:

public static void main(String[] args) {
	CountDownLatch countDownLatch = new CountDownLatch(3);
	Runnable task = () -> {
		try {
			countDownLatch.countDown();
			System.out.println("Countdown: " + countDownLatch.getCount());
			countDownLatch.await();
			System.out.println("Finished");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	};
	for (int i = 0; i < 3; i++) {
		new Thread(task).start();
 	}
}

ก่อนอื่น เราบอกสลักไปcountDown()ที่ Google นิยามการนับถอยหลังว่าเป็น "การนับเลขในลำดับย้อนกลับเป็นศูนย์" จากนั้นเราบอกสลักไปที่await()คือรอจนกว่าตัวนับจะกลายเป็นศูนย์ ที่น่าสนใจนี่คือเคาน์เตอร์แบบครั้งเดียว เอกสาร Java ระบุว่า "เมื่อเธรดต้องนับถอยหลังซ้ำๆ ด้วยวิธีนี้ ให้ใช้ CyclicBarrier แทน" กล่าวอีกนัยหนึ่ง หากคุณต้องการตัวนับที่ใช้ซ้ำได้ คุณต้องมีตัวเลือกอื่นCyclicBarrier:

CyclicBarrier

CyclicBarrierเป็นอุปสรรค "ใช้ซ้ำได้" ตามชื่อ เราจะต้องนำเข้าjava.util.concurrent.CyclicBarrierคลาส ลองดูตัวอย่าง:

public static void main(String[] args) throws InterruptedException {
	Runnable action = () -> System.out.println("On your mark!");
	CyclicBarrier barrier = new CyclicBarrier(3, action);
	Runnable task = () -> {
		try {
			barrier.await();
			System.out.println("Finished");
		} catch (BrokenBarrierException | InterruptedException e) {
			e.printStackTrace();
		}
	};
	System.out.println("Limit: " + barrier.getParties());
	for (int i = 0; i < 3; i++) {
		new Thread(task).start();
	}
}

อย่างที่คุณเห็น เธรดรันเมธอดawaitนั่นคือรอ ในกรณีนี้ ค่าสิ่งกีดขวางจะลดลง สิ่งกีดขวางจะถือว่าพัง ( barrier.isBroken()) เมื่อการนับถอยหลังถึงศูนย์ ในการรีเซ็ตสิ่งกีดขวางคุณต้องเรียกreset()เมธอดซึ่งCountDownLatchไม่มี

เครื่องแลกเปลี่ยน

กลไกต่อไปคือ Exchanger ในบริบทนี้ Exchange คือจุดซิงโครไนซ์ที่มีการแลกเปลี่ยนหรือเปลี่ยนสิ่งที่เปลี่ยนแปลง อย่างที่คุณคาดไว้ an Exchangerคือคลาสที่ทำการแลกเปลี่ยนหรือแลกเปลี่ยน ลองดูตัวอย่างที่ง่ายที่สุด:

public static void main(String[] args) {
	Exchanger<String> exchanger = new Exchanger<>();
	Runnable task = () -> {
		try {
			Thread thread = Thread.currentThread();
			String withThreadName = exchanger.exchange(thread.getName());
			System.out.println(thread.getName() + " exchanged with " + withThreadName);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	};
	new Thread(task).start();
	new Thread(task).start();
}

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

public static void main(String[] args) throws InterruptedException {
	SynchronousQueue<String> queue = new SynchronousQueue<>();
	Runnable task = () -> {
		try {
			System.out.println(queue.take());
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	};
	new Thread(task).start();
	queue.put("Message");
}

ตัวอย่างแสดงให้เห็นว่าเมื่อเธรดใหม่เริ่มต้น มันจะรอ เนื่องจากคิวจะว่างเปล่า จากนั้นเธรดหลักจะใส่สตริง "ข้อความ" ลงในคิว ยิ่งไปกว่านั้น มันจะหยุดจนกว่าจะได้รับสตริงนี้จากคิว คุณยังสามารถอ่าน " SynchronousQueue vs Exchanger " เพื่อหาข้อมูลเพิ่มเติมเกี่ยวกับหัวข้อนี้

เฟสเซอร์

เราได้บันทึกสิ่งที่ดีที่สุดไว้เป็นครั้งสุดท้ายPhaser— เราจะต้องนำเข้าjava.util.concurrent.Phaserคลาส ลองดูตัวอย่างง่ายๆ:

public static void main(String[] args) throws InterruptedException {
        Phaser phaser = new Phaser();
        // By calling the register method, we register the current (main) thread as a party
        phaser.register();
        System.out.println("Phasecount is " + phaser.getPhase());
        testPhaser(phaser);
        testPhaser(phaser);
        testPhaser(phaser);
        // After 3 seconds, we arrive at the barrier and deregister. Number of arrivals = number of registrations = start
        Thread.sleep(3000);
        phaser.arriveAndDeregister();
        System.out.println("Phasecount is " + phaser.getPhase());
    }

    private static void testPhaser(final Phaser phaser) {
        // We indicate that there will be a +1 party on the Phaser
        phaser.register();
        // Start a new thread
        new Thread(() -> {
            String name = Thread.currentThread().getName();
            System.out.println(name + " arrived");
            phaser.arriveAndAwaitAdvance(); // The threads register arrival at the phaser.
            System.out.println(name + " after passing barrier");
        }).start();
    }

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

สรุป

ดังที่คุณเห็นจากตัวอย่างเหล่านี้ มีหลายวิธีในการซิงโครไนซ์เธรด ก่อนหน้านี้ ฉันพยายามนึกถึงลักษณะของการทำงานแบบมัลติเธรด ฉันหวังว่างวดที่แล้วในซีรีส์นี้จะมีประโยชน์ บางคนกล่าวว่าเส้นทางสู่มัลติเธรดเริ่มต้นจากหนังสือ "Java Concurrency in Practice" แม้ว่าจะเปิดตัวในปี 2549 แต่ผู้คนกล่าวว่าหนังสือเล่มนี้ค่อนข้างเป็นพื้นฐานและยังคงมีความเกี่ยวข้องในปัจจุบัน ตัวอย่างเช่น คุณสามารถอ่านการสนทนาได้ที่นี่: "Java Concurrency In Practice" ยังคงใช้ได้อยู่หรือไม่ . นอกจากนี้ยังเป็นประโยชน์ในการอ่านลิงก์ในการสนทนา ตัวอย่างเช่น มีลิงก์ไปยังหนังสือ The Well - Grounded Java Developerและเราจะกล่าวถึงบทที่ 4 โดยเฉพาะ การทำงานพร้อมกันสมัยใหม่ นอกจากนี้ยังมีบทวิจารณ์ทั้งหมดเกี่ยวกับหัวข้อนี้:"การทำงานพร้อมกันของ Java ในทางปฏิบัติ" ยังคงใช้ได้ในยุคของ Java 8 หรือไม่ บทความนี้ยังเสนอคำแนะนำเกี่ยวกับสิ่งอื่นที่ควรอ่านเพื่อทำความเข้าใจหัวข้อนี้อย่างแท้จริง หลังจากนั้น คุณสามารถดูหนังสือดีๆ เช่น OCA/ OCP Java SE 8 Programmer Practice Tests เราสนใจตัวย่อที่สอง: OCP (Oracle Certified Professional) คุณจะพบการทดสอบใน "บทที่ 20: Java Concurrency" หนังสือเล่มนี้มีทั้งคำถามและคำตอบพร้อมคำอธิบาย ตัวอย่างเช่น หลายคนอาจเริ่มพูดว่าคำถามนี้เป็นอีกตัวอย่างหนึ่งของการท่องจำวิธีการ ในแง่หนึ่งใช่ ในทางกลับกัน คุณสามารถตอบคำถามนี้ได้โดยระลึกว่าExecutorServiceเป็น "การอัปเกรด" ชนิดหนึ่งExecutorของ และExecutorมีวัตถุประสงค์เพื่อซ่อนวิธีการสร้างเธรด แต่ไม่ใช่วิธีหลักในการดำเนินการ นั่นคือ เริ่มวัตถุRunnableบนเธรดใหม่ นั่นเป็นเหตุผลที่ไม่มีexecute(Callable)เพราะในจะเพิ่มExecutorServiceเมธอดที่สามารถส่งคืนวัตถุ ได้ แน่นอน เราสามารถจำรายการวิธีการได้ แต่ง่ายกว่ามากที่จะสร้างคำตอบของเราตามความรู้ของเราเกี่ยวกับธรรมชาติของชั้นเรียน และนี่คือเนื้อหาเพิ่มเติมในหัวข้อ: Executorsubmit()Future

• แพ็คเกจ java.util.concurrent
• Java Synchronizers
• แนวคิดการทำงานพร้อมกันใน Java โดย Douglas Hawkins

ดีกว่ากัน: Java และคลาสเธรด ส่วนที่ 1 — เธรดของการดำเนินการ ดีขึ้นด้วยกัน: Java และคลาสเธรด ส่วนที่ II — การซิงโครไนซ์ เข้าด้วยกันได้ดีขึ้น: Java และคลาสเธรด ส่วนที่ 3 — ปฏิสัมพันธ์ ร่วมกันได้ดีขึ้น: Java และคลาสเธรด ตอนที่ IV — Callable, Future และผองเพื่อน อยู่ด้วยกันดีกว่า: Java และคลาส Thread ส่วนที่ V — Executor, ThreadPool, Fork/Jin