Hallo,

ich habe ein Problem mit einer sehr komplexen Thread Thematik. Die Anforderung lautet, eine Queue zu bauen (FIFO), in die alle Client-Requests aufgenommen werden, alle x-Sekunden soll dann ein Thread für das jeweils erste Element in der Queue gestartet werden. Dieser Thread soll über RemoteShell ein Unix-Command auf einem anderen Rechner absetzen und das Ergebnis wieder zurückliefern. Sollte diese Routine mehr als y-Sekunden dauern, dann läuft der Thread in einen Timeout und der Request wird negativ beantwortet. In diesem Fall sollen dann auch alle anderen Requests, die bereits in der Queue vorhanden sind, negetiv beantwortet werden und erst nach z-Sekunden wieder neue Requests zugelassen werden.
Bitte beginnt hier kein Diskussion über die Sinnhaftigkeit dieser Anforderung. Diese ist leider fix.

Ich habe die Queue mit einer LinkedBlockingQueue umgesetzt. Sobald in der run-Methode starte ich dann einen neuen Thread für die Abarbeitung und einen für die Controlling Funktionalität. Wenn der Controller einen Timeout meldet, dann reagiere ich auf die Interrupted Exception und leere die Queue aus. Wäre das so einmal korrekt? Wie kann ich dann ein Ergebnis aus dem Shell-Command zurückgeben? Woher weiß ich, aus welchem Servlet der Request gekommen ist?

Ihr seht: Fragen über Fragen. Für jeden Hinweis bin ich dankbar.

Grüße
Hasel

Hallo,

schau mal hier:

Code java:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162

/**  *  */ package de.tutorials;   import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit;   /**  * @author Tom  *  */ public class RequestProcessorExample {       // ConcurrentLinkedQueue<E>       /**      * @param args      */     public static void main(String[] args) {         final RequestProcessor requestProcessor = new RequestProcessor();           Executors.newSingleThreadExecutor().execute(new Runnable() {             @Override             public void run() {                 System.out.println("Producing requests...");                 while (true) {                     Request request = new Request();                     System.out.println("Produced request: " + request);                     requestProcessor.getRequestQueue().add(request);                     try {                         TimeUnit.MILLISECONDS.sleep(500L);                     } catch (InterruptedException e) {                         e.printStackTrace();                     }                 }             }         });           requestProcessor.start();       }       static class RequestProcessor {         ConcurrentLinkedQueue<Request> requestQueue;         ExecutorService executorService;           int pollTimeOutInSeconds = 5;         int requestProcessingTimeOutInSeconds = 2;           public RequestProcessor() {             this.requestQueue = new ConcurrentLinkedQueue<Request>();             this.executorService = Executors.newCachedThreadPool();         }           public void start() {             System.out.println("Start RequestProcessor");             while (true) {                   if (!requestQueue.isEmpty()) {                     Request request = requestQueue.poll();                     System.out.println("Handling request: " + request);                     Response response = process(request);                     // do something with response...                 }                 try {                     TimeUnit.SECONDS.sleep(pollTimeOutInSeconds);                 } catch (InterruptedException e) {                     e.printStackTrace();                 }             }         }           private Response process(Request request) {             Response response = null;             try {                 System.out.println("Process request: " + request);                 response = getExecutorService().submit(                         new RequestHandler(request)).get(                         requestProcessingTimeOutInSeconds, TimeUnit.SECONDS);             } catch (Exception e) {                 e.printStackTrace();             }             if (null == response) {                 System.out                         .println("Was not able to handle request: " + request);                                 //request failed... remove all other elements from the queue...                 System.out.println("Clearing request queue...");                 requestQueue.clear();                             } else {                 System.out.println("Successfully handled request: " + request);             }             return response;         }           public ConcurrentLinkedQueue<Request> getRequestQueue() {             return requestQueue;         }           public void setRequestQueue(ConcurrentLinkedQueue<Request> requestQueue) {             this.requestQueue = requestQueue;         }           public ExecutorService getExecutorService() {             return executorService;         }           public void setExecutorService(ExecutorService executorService) {             this.executorService = executorService;         }     }       static class RequestHandler implements Callable<Response> {           Request request;           public RequestHandler(Request request) {             super();             this.request = request;         }           @Override         public Response call() throws Exception {             try {                 TimeUnit.SECONDS.sleep((long) (Math.random() * 8));             } catch (InterruptedException e) {                 e.printStackTrace();             }             return new Response();         }           public Request getRequest() {             return request;         }           public void setRequest(Request request) {             this.request = request;         }       }       static class Request {         static int requestCount;         int requestId = requestCount++;           @Override         public String toString() {             return super.toString() + "[" + requestId + "]";         }       }       static class Response {       }   }

Siehe auch hier:
http://www.tutorials.de/forum/java/2...orservice.html

Gruß Tom

Danke. Hat super funktioniert.

Hallo,

ich habe ein ähnliches Problem.
Die Aufgabe besteht darin, dass ein "Masterthread" über einen Datenbankzugriff eine Treffermenge von IDs bekommt.
Aus diesen IDs werden Wrapper erstellt (ModelDataObjects), welche später noch mit Leben gefüllt werden sollen.
Diese Wrapper müssen dann von x Threads bearbeitet und via LPSQL Aufruf in XML umgewandelt werden.

Der eigentlich Haken an der Sache ist folgender:
Der Masterprozess schreibt die IDs in eine Art Queue (LinkedList)
Alle "WorkerThreads" laufen in einem Threadpool und holen sich die Aufgaben aus der Queue.
nachdem eine Aufgabe erfüllt wurde, wird diese aus der Queue entfernt und der Masterprozess benachrichtigt, dass der nächste Prozessabschnitt (nächster Arbeitsschritt z.B. XML in PDF umwandeln) ausgeführt werden kann

Leider habe ich mit Threads noch Nix gemacht und stocher diesbezüglich im Dunkeln herum. =(
Falls Jemand eine Schemenhafte Implementierung oder Idee liefern kann, wäre ich sehr dankbar.
Vor Allem ist mir der Mechanismus mit der Queue wichtig.

Vielen vielen Dank im Voraus!