बैकअप के साथ काम करने के लिए एक बहु-थ्रेडेड एप्लिकेशन बनाने में मेरा अनुभव

अभी, आपने किसी को मल्टीथ्रेड एप्लिकेशन के साथ आश्चर्यचकित नहीं किया है, लेकिन मुझे लगता है कि इस लेख में आप कुछ दिलचस्प विचार पा सकते हैं। जावा का मेरा अध्ययन इस परियोजना के साथ शुरू हुआ, इसलिए शायद कुछ जगहों पर मैं बहुत गलत हो जाऊंगा या एक बड़ी बाइक का निर्माण करूंगा, लेकिन मुझे उम्मीद है कि किसी को जावा में शुरुआत के अनुभव में दिलचस्पी होगी। मैं आवेदन की कई विशेषताएं दूंगा:

• यह बैकअप के साथ विशेष रूप से मेमोरी में काम करता है, बैकअप के आकार की परवाह किए बिना
• पूरे बैकअप को मेमोरी में लोड नहीं करता है
• बैकअप / पुनर्स्थापना संचालन रद्द किया जा सकता है

कटौती के तहत आवेदन की वास्तुकला, साथ ही साथ मुख्य समस्याओं का सामना करना पड़ा और उनके समाधान पर विचार किया जाएगा।

आवेदन अवलोकन

एप्लिकेशन के साथ संचार वेब UI के माध्यम से होता है, लेकिन भविष्य में यदि आवश्यक हो तो REST API जोड़ना संभव होगा।

आवेदन कर सकते हैं:

1. बैकअप बनाएं और उन्हें एक या अधिक स्टोरेज पर अपलोड करें
2. बैकअप को स्टोरेज से लोड करके रिस्टोर करें
3. सभी स्टोरेज से बैकअप हटाएं
4. समय-समय पर बैकअप बनाएं

वर्तमान में समर्थित रिपॉजिटरी:

• स्थानीय फ़ाइल सिस्टम (डॉकर से समर्थित नहीं)
• ड्रॉपबॉक्स

वर्तमान में समर्थित डेटाबेस:

• PostgreSQL

एक विशेष एप्लिकेशन से, मैं नोट कर सकता हूं:

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

नीचे वेब UI के स्क्रीनशॉट हैं जो स्पष्ट रूप से एप्लिकेशन की विशेषताओं का वर्णन करते हैं।

आर्किटेक्चर

मुख्य कार्य 3 सेवाओं में होगा - डेटाबेसबैक , प्रोसेसर , स्टोरेज , और हम कार्यों की अवधारणा का उपयोग करके उन्हें एक साथ जोड़ देंगे। इस सब के बारे में आगे।

DatabaseBackup

यह सेवा सादे-पाठ बैकअप बनाने और पुनर्स्थापित करने के लिए ज़िम्मेदार है।

सेवा इंटरफ़ेस:

public interface DatabaseBackup { InputStream createBackup(DatabaseSettings databaseSettings, Integer id); void restoreBackup(InputStream in, DatabaseSettings databaseSettings, Integer id); } 

दोनों इंटरफ़ेस मेथड इनपुटस्ट्रीम इंस्टेंस पर काम करते हैं, क्योंकि हमें मेमोरी में लोड नहीं होने के लिए पूरे बैकअप की आवश्यकता होती है, जिसका अर्थ है कि बैकअप को स्ट्रीमिंग मोड में पढ़ा / लिखा जाना चाहिए। DatabaseSettings इकाई वेब UI से पूर्व-निर्मित है और डेटाबेस तक पहुँचने के लिए आवश्यक विभिन्न सेटिंग्स संग्रहीत करता है। यह पैरामीटर क्या है - id - आगे थोड़ा समझाया जाएगा।

सेवा आवश्यकताएँ इस प्रकार हैं:

1. दोनों तरीकों को पूरे बैकअप को मेमोरी में नहीं पढ़ना चाहिए।
2. restoreBackup() विधि को एकल लेनदेन में बैकअप को पुनर्स्थापित करना चाहिए, ताकि त्रुटि के मामले में डेटाबेस को असंगत स्थिति में न छोड़ें।

प्रोसेसर

यह सेवा प्रोसेसर और रिवर्स बैकअप रीप्रोसेसिंग के उपयोग के लिए जिम्मेदार है। भंडारण करने से पहले या भंडारण से उतारने के बाद प्रोसेसर का उपयोग किया जाता है। प्रोसेसर उदाहरण: कंप्रेसर, एन्क्रिप्शन।

सेवा इंटरफ़ेस:

 public interface Processor { InputStream process(InputStream in); InputStream deprocess(InputStream in); ProcessorType getType(); // ProcessorType -  Enum,     int getPrecedence(); //   } 

प्रत्येक प्रोसेसर में प्राथमिकता होती है - यदि कई प्रोसेसर निर्दिष्ट हैं, तो उन्हें प्राथमिकता के अवरोही क्रम में लागू किया जाएगा। उलटा फ़ंक्शन उसी क्रम में लागू करना जिसमें प्रोसेसर लगाए गए थे, हमें मूल बैकअप मिलता है।

भंडारण

यह सेवा बैकअप को लोड करने और उतारने के साथ-साथ भंडारण से हटाने के लिए जिम्मेदार है। भंडारण उदाहरण: ड्रॉपबॉक्स, स्थानीय फ़ाइल सिस्टम।

सेवा इंटरफ़ेस:

 public interface Storage { void uploadBackup(InputStream in, StorageSettings storageSettings, String backupName, Integer id); InputStream downloadBackup(StorageSettings storageSettings, String backupName, Integer id); void deleteBackup(StorageSettings storageSettings, String backupName, Integer id); } 

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

कार्य अवधारणा

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

क्रम से चलते हैं। सबसे पहले, हमें कार्यों को लॉन्च करने, बैकअप बनाने, पुनर्स्थापित करने और हटाने के लिए एक सेवा की आवश्यकता है।

कार्य का शुभारंभ

एक बैकअप बनाना:

 public Task startBackupTask(@NotNull Task.RunType runType, @NotNull List<String> storageSettingsNameList, @Nullable List<ProcessorType> processors, @NotNull DatabaseSettings databaseSettings) { Objects.requireNonNull(runType); Objects.requireNonNull(storageSettingsNameList); Objects.requireNonNull(processors); Objects.requireNonNull(databaseSettings); BackupProperties backupProperties = backupPropertiesManager.initNewBackupProperties(storageSettingsNameList, processors, databaseSettings.getName()); Task task = tasksManager.initNewTask(Task.Type.CREATE_BACKUP, runType, backupProperties.getId()); Integer taskId = task.getId(); Future future = tasksStarterExecutorService.submit(() -> { tasksManager.updateTaskState(taskId, Task.State.CREATING); logger.info("Creating backup..."); try (InputStream backupStream = databaseBackupManager.createBackup(databaseSettings, taskId)) { if (Thread.interrupted()) { throw new InterruptedException(); } tasksManager.updateTaskState(taskId, Task.State.APPLYING_PROCESSORS); logger.info("Applying processors on created backup. Processors: {}", processors); try (InputStream processedBackupStream = backupProcessorManager.process(backupStream, processors)) { if (Thread.interrupted()) { throw new InterruptedException(); } tasksManager.updateTaskState(taskId, Task.State.UPLOADING); logger.info("Uploading backup..."); backupLoadManager.uploadBackup(processedBackupStream, backupProperties, taskId); if (Thread.interrupted()) { throw new InterruptedException(); } tasksManager.updateTaskState(taskId, Task.State.COMPLETED); logger.info("Creating backup completed. Backup properties: {}", backupProperties); } } catch (IOException ex) { logger.error("Error occurred while closing input stream of created backup", ex); } catch (RuntimeException ex) { logger.error("Error occurred while creating backup. Backup properties: {}", backupProperties, ex); errorTasksManager.addErrorTask(taskId); } catch (InterruptedException ex) { tasksManager.setInterrupted(taskId); logger.error("Backup creating task was interrupted. Task ID: {}", taskId); } finally { futures.remove(taskId); } }); futures.put(taskId, future); return task; } 

एक बैकअप बनाना निम्नलिखित क्रम में 3 मुख्य चरणों से गुजरता है: एक बैकअप -> प्रोसेसर का अनुप्रयोग -> स्टोरेज पर अपलोड करें। लगभग सभी सेवा विधियों में, हम वर्तमान कार्य की आईडी को अग्रेषित करते हैं ताकि सेवा पृष्ठभूमि में काम करने वाले थ्रेड से एक त्रुटि की रिपोर्ट कर सके। त्रुटि से निपटने के बारे में, क्यों यहाँ InterruptedException है और एक RuntimeException प्राप्त करने के बाद एक त्रुटि के साथ क्या होता है, इस पर बाद में चर्चा की जाएगी।

और यहां बताया गया है कि हम बैकअप बनाने का कार्य कैसे करेंगे:

 tasksStarterService.startBackupTask(Task.RunType.USER, storageSettingsNameList, processors, databaseSettings); 

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

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

डेटाबेस में कार्य निम्न रूप में संग्रहीत है:

 @Entity @Table(name = "backup_tasks") public class Task { /** * Identifier of each backup task. Identifier is generated by PostgreSQL database after saving of entity. */ @Id @Column(insertable = false, updatable = false) @GeneratedValue(strategy = GenerationType.IDENTITY) private Integer id; /** * Backup task type. * <p> * Type is set at the very start of any task and can't be changed. * * @see Type */ @Enumerated(EnumType.STRING) @Column(updatable = false) private Type type; /** * Who initiated a task: user or server. * <p> * We need to know it to show on front only these tasks that was started by user. * * @see RunType */ @Enumerated(EnumType.STRING) @Column(updatable = false) private RunType runType; /** * Backup task state. * <p> * State is updated with every new step in task being executed. * * @see Task.State */ @Enumerated(EnumType.STRING) private State state; /** * Whether task has been interrupted or not. * <p> * Default is {@literal false}. */ @Column(insertable = false) private boolean interrupted; /** * Identifier of {@link BackupProperties}. * <p> * We need to know backup ID to be able to handle occurred errors. */ @Column(updatable = false) private Integer backupPropertiesId; /** * Start time of the task. */ @Column(updatable = false) private LocalDateTime date; public enum RunType { USER, INTERNAL } public enum State { PLANNED, CREATING, RESTORING, DELETING, APPLYING_PROCESSORS, APPLYING_DEPROCESSORS, DOWNLOADING, UPLOADING, COMPLETED, } public enum Type { CREATE_BACKUP { @Override public String toString() { return "CREATE BACKUP"; } }, RESTORE_BACKUP { @Override public String toString() { return "RESTORE BACKUP"; } }, DELETE_BACKUP { @Override public String toString() { return "DELETE BACKUP"; } } } // getters & setters... } 

इस प्रकार, आप चित्र के रूप में एक बैकअप बनाने की प्रक्रिया का वर्णन इस प्रकार कर सकते हैं:

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

 public Task startRestoreTask(@NotNull Task.RunType runType, @NotNull BackupProperties backupProperties, @NotNull String storageSettingsName, @NotNull DatabaseSettings databaseSettings) { Objects.requireNonNull(runType); Objects.requireNonNull(backupProperties); Objects.requireNonNull(storageSettingsName); Objects.requireNonNull(databaseSettings); Task task = tasksManager.initNewTask(Task.Type.RESTORE_BACKUP, runType, backupProperties.getId()); Integer taskId = task.getId(); Future future = tasksStarterExecutorService.submit(() -> { tasksManager.updateTaskState(taskId, Task.State.DOWNLOADING); logger.info("Downloading backup..."); try (InputStream downloadedBackup = backupLoadManager.downloadBackup(backupProperties.getBackupName(), storageSettingsName, taskId)) { if (Thread.interrupted() || downloadedBackup == null) { throw new InterruptedException(); } tasksManager.updateTaskState(taskId, Task.State.APPLYING_DEPROCESSORS); logger.info("Deprocessing backup..."); try (InputStream deprocessedBackup = backupProcessorManager.deprocess(downloadedBackup, backupProperties.getProcessors())) { if (Thread.interrupted()) { throw new InterruptedException(); } tasksManager.updateTaskState(taskId, Task.State.RESTORING); logger.info("Restoring backup..."); databaseBackupManager.restoreBackup(deprocessedBackup, databaseSettings, taskId); if (Thread.interrupted()) { throw new InterruptedException(); } tasksManager.updateTaskState(taskId, Task.State.COMPLETED); logger.info("Restoring backup completed. Backup properties: {}", backupProperties); } } catch (IOException ex) { logger.error("Error occurred while closing input stream of downloaded backup", ex); } catch (RuntimeException ex) { logger.info("Error occurred while restoring backup. Backup properties: {}", backupProperties, ex); errorTasksManager.addErrorTask(taskId); } catch (InterruptedException ex) { tasksManager.setInterrupted(taskId); logger.error("Task was interrupted. Task ID: {}", taskId); } finally { futures.remove(taskId); } }); futures.put(taskId, future); return task; } 

 tasksStarterService.startRestoreTask(Task.RunType.USER, backupProperties, storageSettingsName, databaseSettings); 

 public Task startDeleteTask(@NotNull Task.RunType runType, @NotNull BackupProperties backupProperties) { Objects.requireNonNull(runType); Objects.requireNonNull(backupProperties); Task task = tasksManager.initNewTask(Task.Type.DELETE_BACKUP, runType, backupProperties.getId()); Integer taskId = task.getId(); Future future = tasksStarterExecutorService.submit(() -> { try { logger.info("Deleting backup started. Backup properties: {}", backupProperties); tasksManager.updateTaskState(taskId, Task.State.DELETING); backupLoadManager.deleteBackup(backupProperties, taskId); if (Thread.interrupted()) { throw new InterruptedException(); } tasksManager.updateTaskState(taskId, Task.State.COMPLETED); logger.info("Deleting backup completed. Backup properties: {}", backupProperties); } catch (RuntimeException ex) { logger.error("Error occurred while deleting backup. Backup properties: {}", backupProperties, ex); errorTasksManager.addErrorTask(taskId); } catch (InterruptedException ex) { tasksManager.setInterrupted(taskId); logger.error("Task was interrupted. Task ID: {}", taskId); } finally { futures.remove(taskId); } }); futures.put(taskId, future); return task; } 

 tasksStarterService.startDeleteTask(Task.RunType.USER, backupProperties); 

कार्य रद्द करें

कार्य निरस्तीकरण क्या है? बेशक, यह एक थ्रेड समाप्ति से ज्यादा कुछ नहीं है। आप देख सकते हैं कि फ्यूचर में चल रहे सभी मुख्य कोड निम्नलिखित ट्राइ-कैच निर्माण में लिपटे हुए हैं:

 try { ... } catch (InterruptedException ex) { ... tasksManager.setInterrupted(taskId); } 

और प्रत्येक महत्वपूर्ण विधि के बाद, निष्पादन प्रवाह, जिसे हम नियंत्रित करते हैं, निम्नलिखित निर्माण स्थापित है:

 if (Thread.interrupted()) { throw new InterruptedException(); } 

आगे बढ़ने से पहले, जेवीएम थ्रेड्स के रुकावटों और राज्यों का एक संक्षिप्त सिद्धांत दिया जाना चाहिए।

JVM में थ्रेड्स निम्न अवस्थाएँ हो सकती हैं:

1. नई
2. runnable
3. समय का इंतजार किया
4. प्रतीक्षा कर रहा है
5. अवरोधित
6. समाप्त

हम केवल वेटिंग और टाइमिंग प्रतीक्षा वाले राज्यों में रुचि रखते हैं। Object.wait() वेटिंग स्थिति में Object.wait() गया है Object.wait() , Thread.join() और अन्य तरीकों से। थ्रेड को समयबद्ध प्रतीक्षा स्थिति में स्थानांतरित किया जाता है (यानी, एक प्रतीक्षा जो एक निश्चित अवधि तक चलती है) Object.wait(timeout) , Thread.join(timeout) , Thread.sleep(sleeping) और अन्य का उपयोग करते हुए।

यहां सबसे महत्वपूर्ण बात यह है कि यदि आप प्रतीक्षा या समय की प्रतीक्षा की स्थिति में प्रवेश करने से पहले धागे को बाधित करते हैं या जब धागा इस स्थिति में होता है , तो थ्रेड जाग जाता है, एक इंटरप्टेडएक्सलाइट को फेंक देता है।

लेकिन यह सब नहीं है। यह बिल्कुल भी नहीं है कि एक थ्रेड बैकअप बनाने, पुनर्स्थापित करने या हटाने के द्वारा कभी भी स्टेट डेटा में जाएगा। फिर थ्रेड को कैसे सूचित किया जाए कि यह बाधित था?

पहला तरीका यह है कि थ्रेड.interrupted Thread.interrupted() या Thread.currentThread.isInterrupted() विधियों का उपयोग करके थ्रेड के साथ इंटरप्ट फ़्लैग को स्वतंत्र रूप से देखें। दोनों के बीच का अंतर यह है कि पहले निजी देशी विधि को कॉल करता है currentThread.isInterrupted(boolean ClearInterrupted) , यह true है, यह दर्शाता है कि रुकावट का झंडा साफ हो जाएगा, और दूसरा गुजरने वाला false , बीच में आने वाले झंडे को अछूता छोड़ देगा। इन दो तरीकों के बीच का चुनाव पूरी तरह से स्थिति पर निर्भर करता है। जब एक InterruptedException को फेंक दिया जाता है, तो बाधा झंडा भी साफ़ हो जाता है - यह याद रखने योग्य है।

लेकिन एक रास्ता आसान होना चाहिए - और यह है। आवेदन में, I / O धाराओं के साथ काम की एक बड़ी मात्रा है, और इसलिए I / O विधियों के साथ। हमारा कार्य यह सुनिश्चित करना है कि आई / ओ स्ट्रीम पर read() या write(int b) तरीकों को कॉल करते समय, एक त्रुटि को रुकावट के दौरान फेंक दिया जाता है, यह सूचित करते हुए कि अवरुद्ध आई / ओ कॉल बाधित हो गया था। सौभाग्य से, जावा में ऐसा अपवाद है - InterruptedIOException । हालाँकि, सभी पठन / लेखन धारा विधियाँ धागा अवरोधों की निगरानी नहीं करती हैं, और विशेष रूप से केवल PipedInputStream इसकी निगरानी करता है। इसलिए, उन जगहों पर जहां यह धारा शामिल नहीं है, हमें रीड / राइट विधि का विस्तार करना चाहिए ताकि जब कोई रुकावट हो, तो एक InterruptedIOException को फेंक दिया जाए। वास्तव में, रीड () पद्धति का विस्तार केवल एक ही स्थान पर मेरे लिए पर्याप्त था - जब इनपुटस्ट्रीम बैकअप अपलोड विधि से वापस आया। इस तरह से हम हर जगह झंडे पर टेम्पलेट की जांच किए बिना एक रुकावट की उत्पत्ति के बारे में जान सकते हैं। हालांकि, इस अपवाद को IOException से अलग से पकड़ना और इसे अलग से संभालना महत्वपूर्ण है। बेशक, आप कुछ स्थानों पर ध्वज की टेम्प्लेट जांच की मदद के बिना नहीं कर सकते, लेकिन यह पहले से बेहतर हो गया है।

यह भी ध्यान रखना महत्वपूर्ण है कि यदि झंझट को प्रसंस्करण के दौरान झंडे को साफ किया गया था, तो अवरोधक ध्वज को फिर से सेट करना हमेशा आवश्यक होता है ताकि विधि से लौटने के बाद हम उस रुकावट के बारे में पता लगा सकें जो घटित हुई थी।

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

 backupLoadManager.uploadBackup(processedBackupStream, backupProperties, taskId); <-   ,       if (Thread.interrupted()) { //      ,      - ,    throw new InterruptedException(); } 

इसलिए, InterruptedException या InterruptedIOException को हैंडल करना अच्छा अभ्यास है:

 try { ... } catch (InterruptedException e) { //  InterruptedIOException ... // re-interrupt the thread Thread.currentThread().interrupt(); } 

ठीक है, हम रुकावट को संभाल सकते हैं, लेकिन वास्तव में थ्रेड्स को बाधित कौन करेगा?
ऐसा करने के लिए, हम एक अन्य संस्था बनाएँगे जिसे CancelTask कहा जाता है, जो रद्द करने के लिए कार्य की आईडी को संग्रहीत करेगा, और एक घड़ी भी लिखेगा जो कार्यों को बाधित करने का प्रयास करेगी। क्यों कोशिश करें? क्योंकि:

1. किसी अन्य सर्वर की स्मृति में थ्रेड को समाप्त करने में असमर्थ। कई सर्वर हमारे लिए काम कर सकते हैं, जिसका मतलब है कि फ्यूचर अलग-अलग सर्वर पर बिखरा हुआ है। इस प्रकार, जब किसी कार्य को रद्द करने का अनुरोध सर्वर में से किसी एक पर आता है, तो वांछित भविष्य दूसरे सर्वर की स्मृति में हो सकता है।
2. सर्वर क्रैश के कारण फ्यूचर के खो जाने के बाद टास्क को रद्द नहीं किया जा सकता है।

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

रद्द करें

 /** * This class scans for tasks to cancel and tries to cancel them. */ @Component class CancelTasksWatcher { private static final Logger logger = LoggerFactory.getLogger(CancelTasksWatcher.class); private static final Duration cancelTimeout = Duration.ofMinutes(10); private CancelTasksManager cancelTasksManager; private TasksStarterService tasksStarterService; private TasksManager tasksManager; // spring setters... /** * This watcher wakes up every time 10 seconds passed from the last completion, checks if there are any tasks to cancel and tries to * cancel each task. * <p> * Since there are can be working more that one instance of the program, {@literal Future} instance of task can belong to different * servers. We can't get access to {@literal Future} if it's not in memory of the server where task cancellation request was accepted. * So the purpose of this watcher is to be able cancel tasks that works in the other instance of program. Each server has this watcher * checking for available cancellation requests and if any, the watcher tries to cancel corresponding {@literal Future}. * If cancellation is successful task will be also reverted. * <p> * If task cancellation request timeout exceeded, then it means a server that had requested {@literal Future} instances has been * shutdown, so all {@literal Future} instances lost and task can't be canceled. In such case task cancellation request will be ignored. * * @see TasksStarterService#getFuture(Integer) * @see TasksManager#revertTask(Task) */ @Scheduled(fixedDelay = 10 * 1000) @Transactional(isolation = Isolation.READ_COMMITTED, propagation = Propagation.REQUIRES_NEW) public void watchTasksToCancel() { Iterable<CancelTask> cancelTasks = cancelTasksManager.findAll(); Iterable<Task> tasks = tasksManager.findAllById(StreamSupport.stream(cancelTasks.spliterator(), false) .map(CancelTask::getTaskId).collect(Collectors.toList())); Map<Integer, Task> tasksAsMap = StreamSupport.stream(tasks.spliterator(), false) .collect(Collectors.toMap(Task::getId, Function.identity())); List<Integer> taskIdsForDeleting = new ArrayList<>(); for (CancelTask cancelTask : cancelTasks) { Integer taskId = cancelTask.getTaskId(); Task task = tasksAsMap.get(taskId); if (task == null) { logger.error("Can't cancel task: no such entity with ID {}", taskId); taskIdsForDeleting.add(taskId); continue; } // timeout exceeded, that is server shutdown and lost all Future instances, so task can't be canceled if (LocalDateTime.now(ZoneOffset.UTC).isAfter(cancelTask.getPutTime().plus(cancelTimeout))) { logger.error("Can't cancel task: timeout exceed. Task ID: {}", taskId); taskIdsForDeleting.add(taskId); continue; } tasksStarterService.getFuture(taskId).ifPresent(future -> { logger.info("Canceling task with ID {}", taskId); boolean canceled = future.cancel(true); if (canceled) { try { // give time to properly handle interrupt Thread.sleep(10000); } catch (InterruptedException e) { // should not happen } tasksManager.revertTask(task); } taskIdsForDeleting.add(taskId); logger.info("Task canceled: {}. Task ID: {}", canceled, taskId); }); } cancelTasksManager.deleteByTaskIdIn(taskIdsForDeleting); } } 

हैंडलिंग में त्रुटि

, , Future, try-catch :

 try { ... } catch (RuntimeException e) { ... errorTasksManager.addErrorTask(taskId); } 

RuntimeException , Future , .

addErrorTask(taskId) , ID , .
? , , , .

:
, , . — PostgreSQL select for update , select skip locked . , , revertTask() , .

ErrorTasksWatcher :

 /** * This class scans for erroneous tasks and handles them depending on their state. */ @Component class ErrorTasksWatcher { private static final Logger logger = LoggerFactory.getLogger(ErrorTasksWatcher.class); private static final Integer nRows = 10; private TasksManager tasksManager; private ErrorTasksManager errorTasksManager; // spring setters... /** * This watcher wakes up every time 1 minute passed from the last completion, checks backup states periodically and handles erroneous * tasks if any. * <p> * The watcher handles at most N tasks as described by {@link #nRows} constant and skips already locked tasks. * When retrieving error tasks from database pessimistic lock is set. It allows safely run more than one copy of program, as no other * watcher can pick up already being handled error tasks. * <p> * If the server shutdowns while rows was locked, transaction will be rolled back and lock released, so these entities can be picked * up by the other running server. */ @Scheduled(fixedDelay = 60 * 1000) @Transactional(isolation = Isolation.READ_COMMITTED, propagation = Propagation.REQUIRES_NEW) public void watchErrorTasks() { for (ErrorTask errorTask : errorTasksManager.findFirstNAndLock(nRows)) { if (!errorTask.isErrorHandled()) { Integer backupTaskId = errorTask.getTaskId(); Optional<Task> optionalTask = tasksManager.findById(backupTaskId); if (!optionalTask.isPresent()) { logger.info("Can't handle erroneous task: no corresponding backup task entity. Backup task ID: {}", backupTaskId); continue; } tasksManager.revertTask(optionalTask.get()); errorTask.setErrorHandled(true); } } } } 

revertTask(Task) :

  /** * This function reverts erroneous task by its entity. * <p> * Use this function only after canceling related {@literal Future}. * <p> * If the task was of the type {@link Task.Type#CREATE_BACKUP} then related {@link BackupProperties} will be deleted. * * @param task the entity */ public void revertTask(@NotNull Task task) { Objects.requireNonNull(task); Task.State state = task.getState(); switch (state) { case DOWNLOADING: case APPLYING_DEPROCESSORS: case RESTORING: case DELETING: { logger.info("Handling broken operation. Operation: {}. No extra actions required", state.toString()); break; } case CREATING: case APPLYING_PROCESSORS: { logger.info("Handling broken operation. Operation: {}. Delete backup properties...", state.toString()); Integer backupPropertiesID = task.getBackupPropertiesId(); if (!backupPropertiesManager.existsById(backupPropertiesID)) { logger.error("Can't revert task: no related backup properties. Task info: {}", task); return; } backupPropertiesManager.deleteById(backupPropertiesID); break; } case UPLOADING: { logger.info("Handling broken operation. Operation: {}. Deleting backup from storage...", state); Integer backupPropertiesId = task.getBackupPropertiesId(); Optional<BackupProperties> optionalBackupProperties = backupPropertiesManager.findById(backupPropertiesId); if (!optionalBackupProperties.isPresent()) { logger.error("Can't revert task: no related backup properties. Task info: {}", task); return; } tasksStarterService.startDeleteTask(Task.RunType.INTERNAL, optionalBackupProperties.get()); backupPropertiesManager.deleteById(backupPropertiesId); break; } default: { logger.error("Can't revert task: unknown state. Task info: {}", task); } } } 

, . , ? , , Future ( 1), , InputStream ( 2). , 2, 1 2 ?

, , , . Future ( 1) :

  public void onError(@NotNull Throwable t, @NotNull Integer taskId) { logger.error("Exception caught. Task ID: {}", taskId, t); Optional<Future> optionalFuture = tasksStarterService.getFuture(taskId); if (!optionalFuture.isPresent()) { logger.error("Can't cancel the Future of task with ID {}: no such Future instance", taskId); } else { boolean canceled = optionalFuture.get().cancel(true); if (!canceled) { logger.error("Error canceling the Future of task with ID {}", taskId); } else { logger.info("Task canceled. Task ID: {}", taskId); errorTasksManager.setError(taskId); } } } 

, , ID , , Future - , ID .

, , , , , .

, :

, , , . — Future.

, , , I/O , — / . , . :

1. , . , — .
2. — Future , . , / , , ( , — IOException , , ).

, — ( ID , , ), .

, , . , , .

1. Web UI: , . ,

निष्कर्ष

:

• GitHub
• Docker Hub

, ! , GitHub!