JDK समवर्ती पैकेज

वर्तमान में जावा में मौजूद मेमोरी मॉडल इस कोड में थ्रेड रेसिंग की अनुपस्थिति में बहु-थ्रेडेड कोड के अपेक्षित निष्पादन आदेश की गारंटी देता है। और अपने कोड को रेसिंग से बचाने के लिए, उनके बीच डेटा को सिंक्रनाइज़ और एक्सचेंज करने के विभिन्न तरीकों का आविष्कार किया गया है।

HotSpot JDK के साथ शामिल java.util.concurrent पैकेज में बहु-थ्रेडेड कोड लिखने के लिए निम्नलिखित टूल दिए गए हैं:

• परमाणु
• ताले
• संग्रह
• तुल्यकालन अंक
• निष्पादकों
• संचायक _jdk 1.8_

परमाणु

java.util.concurrent.atomic चाइल्ड पैकेज में आदिम प्रकार के साथ परमाणु कार्य के लिए कक्षाओं का एक सेट होता है। इन वर्गों का अनुबंध "प्रोसेसर समय की 1 इकाई" के लिए compare-and-set ऑपरेशन के प्रदर्शन की गारंटी देता है। इस चर के लिए एक नया मान सेट करते समय, आप इसके पुराने मूल्य (आशावादी लॉकिंग दृष्टिकोण) को भी पास करते हैं। यदि, जिस क्षण से विधि को बुलाया जाता है, चर का मान अपेक्षित से भिन्न होता है, तो निष्पादन का परिणाम false होगा।

उदाहरण के लिए, long चर [1,2,3,4,5] और [-1,-2,-3,-4,-5] । प्रत्येक थ्रेड क्रमिक रूप से सरणी पर पुनरावृत्त होगा और तत्वों को एक एकल चर में सारांशित करेगा। एक निराशावादी ताला के साथ कोड ( ग्रूवी ) इस तरह दिखता है:

 class Sum { static monitor = new Object() static volatile long sum = 0 } class Summer implements Callable { long[] data Object call() throws Exception { data.each { synchronized (Sum.monitor) { println("${Thread.currentThread().name}: add ${it} to ${Sum.sum}") Sum.sum += it } } } } Executors.newFixedThreadPool(2).invokeAll([ new Summer(data: [1,2,3,4,5]), new Summer(data: [-1,-2,-3,-4,-5]) ]) print("Sum: ${Sum.sum}") 

निष्पादन के परिणाम की उम्मीद की जाएगी:

 pool-1-thread-1: add 1 to 0 pool-1-thread-2: add -1 to 1 pool-1-thread-1: add 2 to 0 pool-1-thread-2: add -2 to 2 pool-1-thread-1: add 3 to 0 pool-1-thread-2: add -3 to 3 pool-1-thread-1: add 4 to 0 pool-1-thread-1: add 5 to 4 pool-1-thread-2: add -4 to 9 pool-1-thread-2: add -5 to 5 Sum: 0 

हालांकि, इस दृष्टिकोण में महत्वपूर्ण प्रदर्शन नुकसान हैं। इस मामले में, अधिक बेकार काम हमें उपयोगी से अधिक संसाधन लेता है:

• मॉनिटर को लॉक करने का प्रयास
• प्रवाह अवरुद्ध
• मॉनिटर अनलॉक
• धागा अनलॉक

एक ही राशि की गणना करते समय आशावादी लॉकिंग को लागू करने के लिए AtomicLong का उपयोग करने पर विचार करें:

 class Sum { static volatile AtomicLong sum = new AtomicLong(0) } class Summer implements Callable { long[] data Object call() throws Exception { data.each { while(true) { long localSum = Sum.sum.get() if (Sum.sum.compareAndSet(localSum, localSum + it)) { println("${Thread.currentThread().name}: add ${it} to ${Sum.sum}") break; } else { println("[MISS!] ${Thread.currentThread().name}: add ${it} to ${Sum.sum}") } } } } } Executors.newFixedThreadPool(2).invokeAll([ new Summer(data: [1,2,3,4,5]), new Summer(data: [-1,-2,-3,-4,-5]) ]) print("Sum: ${Sum.sum}") 

जैसा कि आप "गलत" प्रयासों के परिणामों से देख सकते हैं, इतने सारे नहीं थे:

 [MISS!] pool-1-thread-1: add 1 to -1 pool-1-thread-2: add -1 to -1 pool-1-thread-2: add -2 to -3 [MISS!] pool-1-thread-1: add 1 to -3 pool-1-thread-2: add -3 to -6 pool-1-thread-1: add 1 to -5 [MISS!] pool-1-thread-2: add -4 to -5 pool-1-thread-1: add 2 to -7 pool-1-thread-2: add -4 to -7 pool-1-thread-1: add 3 to -9 pool-1-thread-2: add -5 to -9 pool-1-thread-1: add 4 to -5 pool-1-thread-1: add 5 to 0 Sum: 0 

जब आशावादी लॉकिंग का उपयोग करने का निर्णय लेते हैं, तो यह महत्वपूर्ण है कि परिवर्तनशील चर के साथ कार्रवाई में अधिक समय न लगे। यह क्रिया जितनी लंबी होगी, उतनी बार गलत compare-and-set हो जाएगा, और जितनी अधिक बार आपको इस क्रिया को फिर से करना होगा।

compare-and-set आधार पर, एक गैर-अवरुद्ध रीड लॉक भी लागू किया जा सकता है। इस मामले में, परमाणु चर संसाधित किए जा रहे ऑब्जेक्ट के संस्करण को संग्रहीत करेगा। गणना से पहले संस्करण मूल्य प्राप्त करने के बाद, हम गणना के बाद इसे सत्यापित कर सकते हैं। नियमित रूप से read-write ताले तभी प्रभावी होते हैं जब संस्करण सत्यापन विफल हो जाता है।

 class Transaction { long debit } class Account { AtomicLong version = new AtomicLong() ReadWriteLock readWriteLock = new ReentrantReadWriteLock() List<Transaction> transactions = new ArrayList<Transaction>() } long balance(Account account) { ReentrantReadWriteLock.ReadLock locked while(true) { long balance = 0 long version = account.version.get() account.transactions.each {balance += it.debit} //volatile write for JMM if (account.version.compareAndSet(version, version)) { if (locked) {locked.unlock()} return balance } else { locked = account.readWriteLock.readLock() } } } void modifyTransaction(Account account, int position, long newDebit) { def writeLock = account.readWriteLock.writeLock() account.version.incrementAndGet() account.transactions[position].debit = newDebit writeLock.unlock() } 

ताले

ReentrantLock

सिंक्रोनाइज्ड ताले के विपरीत, ReentrantLock अधिक लचीले ढंग से तालों को हटाने और प्राप्त करने के क्षणों का चयन करने की अनुमति देता है नियमित जावा कॉल का उपयोग करता है। इसके अलावा, ReentrantLock आपको लॉक की वर्तमान स्थिति के बारे में जानकारी प्राप्त करने की अनुमति देता है, आपको एक निश्चित समय के लिए लॉक के लिए "प्रतीक्षा" करने की अनुमति देता है। एक ही धागे के लिए सही पुनरावर्ती पुनर्प्राप्ति और ताले की रिहाई का समर्थन करता है। यदि आपको ईमानदार ताले की आवश्यकता होती है (मॉनिटर को कैप्चर करते समय आदेश रखना) - ReentrantLock भी इस तंत्र से लैस है।

इस तथ्य के बावजूद कि syncronized और ReentrantLock ताले बहुत समान हैं - जेवीएम स्तर पर कार्यान्वयन काफी अलग है।
जेएमएम के विवरण में जाने के बिना: जेवीएम द्वारा प्रदान किए गए सिंक्रोनाइज्ड लॉक के बजाय ReentrantLock का उपयोग करना केवल इसके लायक है अगर आपके पास मॉनिटर के लिए अक्सर थ्रेड्स की लड़ाई होती है। मामले में जब केवल एक धागा सिंकट्रॉनिक विधि में हो जाता है, तो रेन्ट्रेंटलॉक प्रदर्शन जेवीएम लॉकिंग तंत्र ReentrantLock नीच है।

ReentrantReadWriteLock

कई पढ़ने और लिखने के ताले पर कब्जा करने ReentrantLock क्षमता के ReentrantLock गुणों की आपूर्ति करता है। यदि आवश्यक हो, तो रीड लॉक से पहले राइट लॉक को "कम" किया जा सकता है।

मुद्रांकितलोक _jdk 1.8_

यह आशावादी और निराशावादी पढ़ने-लिखने के ताले को उनके आगे बढ़ने या घटने की संभावना के साथ लागू करता है। आशावादी लॉकिंग लॉक ( जावदोक ) के "स्टैम्प" के माध्यम से कार्यान्वित किया जाता है:

 double distanceFromOriginV1() { // A read-only method long stamp; if ((stamp = sl.tryOptimisticRead()) != 0L) { // optimistic double currentX = x; double currentY = y; if (sl.validate(stamp)) return Math.sqrt(currentX * currentX + currentY * currentY); } stamp = sl.readLock(); // fall back to read lock try { double currentX = x; double currentY = y; return Math.sqrt(currentX * currentX + currentY * currentY); } finally { sl.unlockRead(stamp); } } 

संग्रह

ArrayBlockingQueue

एक थ्रेड से दूसरे मेसेज भेजने के लिए एक ईमानदार कतार। अवरोधक ( put() take() ) और नॉनब्लॉकिंग ( offer() pool() offer() का समर्थन करता है। निषिद्ध मूल्यों को निषिद्ध करता है। निर्माण पर कतार की क्षमता निर्दिष्ट की जानी चाहिए।

ConcurrentHashMap

hash फ़ंक्शन के आधार पर एक कुंजी-मूल्य संरचना। पढ़ने के ताले नहीं हैं। रिकॉर्डिंग करते समय, कार्ड का केवल एक हिस्सा (खंड) अवरुद्ध होता है। सेगमेंट की संख्या निकटतम डिग्री 2 तक सीमित है।

ConcurrentSkipListMap

बैलेंस्ड मल्टीथ्रेडेड की-वैल्यू स्ट्रक्चर (O (लॉग एन))। खोज एक स्किप की गई सूची पर आधारित है। कार्ड को चाबियों की तुलना करने में सक्षम होना चाहिए।

ConcurrentSkipListSet

मूल्यों के बिना ConcurrentSkipListMap ।

CopyOnWriteArrayList

एक राइट-ब्लॉक, गैर-रीड-ब्लॉक सूची। कोई भी संशोधन स्मृति में सरणी का एक नया उदाहरण बनाता है।

CopyOnWriteArraySet

CopyOnWriteArrayList बिना CopyOnWriteArrayList ।

DelayQueue

PriorityBlockingQueue एक निश्चित देरी के बाद ही एक तत्व प्राप्त करने की अनुमति देता है (देरी को ऑब्जेक्ट के Delayed इंटरफ़ेस के माध्यम से घोषित किया जाता है)। एक अनुसूचक को लागू करने के लिए DelayQueue का उपयोग किया जा सकता है। कतार की क्षमता तय नहीं है।

LinkedBlockingDeque

कनेक्टिविटी के आधार पर द्विदिश BlockingQueue क्यू (कैश-मिस एंड कैश कोहेरेंस ओवरहेड)। कतार की क्षमता तय नहीं है।

LinkedBlockingQueue

कनेक्टिविटी के आधार पर यूनिडायरेक्शनल BlockingQueue क्यू (कैशे-मिस एंड कैश कोहेरेंस ओवरहेड)। कतार की क्षमता तय नहीं है।

LinkedTransferQueue

कनेक्टिविटी के आधार पर यूनिडायरेक्शनल `ब्लॉकिंगक्यूएयू` (कैश-मिस एंड कैश कोहेरेंस ओवरहेड)। कतार की क्षमता तय नहीं है। यह कतार आपको हैंडलर को तत्व चुनने के लिए इंतजार करने की अनुमति देती है।

PriorityBlockingQueue

यूनिडायरेक्शनल `ब्लॉकिंगक्यूएयू`, संदेशों को प्राथमिकता देने की अनुमति देता है (तत्वों की तुलना के माध्यम से)। निषिद्ध मूल्यों को निषिद्ध करता है।

SynchronousQueue

एक यूनिडायरेक्शनल `BlockingQueue` जो put() लिए transfer() तर्क को लागू करता है।

तुल्यकालन अंक

CountDownLatch

एक अवरोध ( await() ) जो countDown() लिए एक विशिष्ट (या अधिक) कॉलों की संख्या की अपेक्षा करता है। बाधा की स्थिति को रीसेट नहीं किया जा सकता है।

CyclicBarrier

एक बाधा ( await() ) जो अन्य थ्रेड्स द्वारा await() कॉल की एक विशिष्ट संख्या की await() करती है। जब थ्रेड्स की संख्या निर्दिष्ट तक पहुंच जाती है, तो एक वैकल्पिक कॉलबैक कहा जाएगा और लॉक जारी किया जाएगा। अवरोध अपने राज्य को प्रारंभिक अवस्था में ले जाता है जब लंबित प्रवाह जारी होता है और पुन: उपयोग किया जा सकता है।

एक्सचेंजर

दो थ्रेड्स को सिंक्रनाइज़ करने के लिए बैरियर (`एक्सचेंज ()`)। सिंक्रनाइज़ेशन के समय, थ्रेड्स के बीच वस्तुओं का अस्थिर स्थानांतरण संभव है।

Phaser

विस्तार `CyclicBarrier`, बाधा के प्रत्येक चक्र के लिए प्रतिभागियों को पंजीकृत करने और हटाने की अनुमति देता है।

सिकंदरा

मॉनिटर को पकड़ने के लिए केवल निर्दिष्ट थ्रेड की अनुमति देने वाला अवरोध। अनिवार्य रूप से `लॉक` की कार्यक्षमता को कई थ्रेड में ब्लॉक में होने की क्षमता का विस्तार करता है।

निष्पादकों

ExecutorService ने new Thread(runnable) साथ काम को आसान बनाने के लिए new Thread(runnable) को बदल दिया। ExecutorService मुक्त थ्रेड का पुन: उपयोग करने में मदद करता है, थ्रेड पूल के लिए कार्यों से कतारें व्यवस्थित करता है, और कार्य के परिणाम के लिए सदस्यता लेता है। Runnable इंटरफ़ेस के बजाय, पूल Runnable इंटरफ़ेस का उपयोग करता है (यह एक परिणाम वापस कर सकता है और त्रुटियों को फेंक सकता है)।

 ExecutorService pool = Executors.newFixedThreadPool(4) Future future = pool.submit(new Callable() { Object call() throws Exception { println("In thread") return "From thread" } }) println("From main") println(future.get()) try { pool.submit(new Callable() { Object call() throws Exception { throw new IllegalStateException() } }).get() } catch (ExecutionException e) {println("Got it: ${e.cause}")} pool.shutdown() 

invokeAll विधि कॉलिंग थ्रेड को सभी कार्यों के पूरा होने पर ही नियंत्रण देती है। invokeAny विधि पहले सफलतापूर्वक पूर्ण किए गए कार्य का परिणाम देती है, बाद के सभी को रद्द कर देती है।

ThreadPoolExecutor

पूल में काम करने की क्षमता और थ्रेड्स की अधिकतम संख्या निर्दिष्ट करने के लिए थ्रेड्स का एक पूल, कार्यों के लिए एक कतार।

ScheduledThreadPoolExecutor

कार्य को स्थगित या नियमित रूप से करने ThreadPoolExecutor क्षमता के ThreadPoolExecutor की कार्यक्षमता का विस्तार करता है।

ThreadPoolExecutor

आत्म-प्रतिकृति कार्यों के लिए हल्का धागा पूल। पूल को उम्मीद है कि माता-पिता में बच्चे के कार्यों के `फोर्क ()` और `ज्वाइन () तरीके शामिल हैं।

 class LNode { List<LNode> childs = [] def object } class Finder extends RecursiveTask<LNode> { LNode node Object expect protected LNode compute() { if (node?.object?.equals(expect)) { return node } node?.childs?.collect { new Finder(node: it, expect: expect).fork() }?.collect { it.join() }?.find { it != null } } } ForkJoinPool es = new ForkJoinPool() def invoke = es.invoke(new Finder( node: new LNode( childs: [ new LNode(object: "ivalid"), new LNode( object: "ivalid", childs: [new LNode(object: "test")] ) ] ), expect: "test" )) print("${invoke?.object}") 

संचायक _jdk 1.8_

बैटरियों आप कैस का उपयोग किए बिना एक बहु-थ्रेडेड वातावरण में संख्यात्मक तत्वों पर आदिम संचालन (योग / अधिकतम मूल्य खोज) करने की अनुमति देते हैं।