JMS est-il la réponse à la nécessité d'une file d'attente de blocage persistante?

Question

Je crée une bibliothèque composée d'un applicateur Log4J qui envoie des événements de manière asynchrone à un serveur distant. Lorsqu'une instruction de journal est créée, l'appender enregistre l'événement de manière asynchrone dans une file d'attente locale qu'un groupe de consommateurs récupère ensuite et envoie à la télécommande.

La solution complètement en mémoire serait de créer un BlockingQueue qui gérerait le problème de la concurrence. Cependant, j'aimerais que la file d'attente soit conservée de façon à ce que, si le serveur distant n'est pas disponible, je ne développe pas la file d'attente sans limite ou commence à rejeter les messages dans le cas d'une file d'attente limitée. Je pensais à utiliser une base de données H2 intégrée pour stocker les événements localement, puis utiliser un mécanisme d'interrogation pour récupérer les événements et les envoyer à la télécommande. Je préfèrerais utiliser un BlockingQueue plutôt que d'interroger une table de base de données.

Est-ce que JMS est la réponse?

EDIT:

Si JMS est la réponse, et il semble aller de cette façon, quelqu'un at-il des recommandations sur une solution JMS léger, intégrable qui peut être configuré pour accepter uniquement les messages en cours de traitement? En d'autres termes, je ne veux pas, et je ne le ferai peut-être pas, ouvrir un socket TCP sur lequel écouter.

EDIT:

J'ai ActiveMQ intégrés maintenant et il semble fonctionner. Merci a tous.

Answer 1

Vous pouvez utiliser JMS pour envoyer de manière asynchrone des messages à un ordinateur distant (en supposant qu'il puisse les recevoir bien sûr), Log4j dispose d'un Appender JMS que vous pouvez utiliser pour cela.

Answer 2

Vous pouvez certainement utiliser JMS à cet effet. Autant que je comprends, vous utilisez l'appender Log4J JMS. Ce composant envoie des messages à destination JMS préconfigurée (généralement en file d'attente). Vous pouvez configurer cette file d'attente pour qu'elle soit persistante. Dans ce cas, tous les messages insérés dans la file d'attente seront automatiquement stockés dans un magasin persistant (généralement une base de données.). Malheureusement, cette configuration est spécifique au fournisseur (dépend du fournisseur JMS), mais est généralement très simple. Veuillez vous reporter à la documentation de votre fournisseur JMS.

Answer 3

Bob Lee a ouvert depuis peu une file d'attente très simple, https://github.com/square/retrofit/blob/master/modules/android/src/retrofit/io/QueueFile.java - peut être utile, et est certainement beaucoup plus facile à introduire que JMS si vous pouvez accepter la durabilité locale.

Cette classe est autonome - elle peut être copiée et collée.

Answer 4

Voir si cela fonctionne

Ce code devrait fonctionner pour vous - son une mémoire file d'attente de blocage persistant - a besoin des réglages de fichiers, mais devrait fonctionner

 package test; 

    import java.io.BufferedReader; 
    import java.io.BufferedWriter; 
    import java.io.File; 
    import java.io.FileReader; 
    import java.io.FileWriter; 
    import java.io.IOException; 
    import java.util.ArrayList; 
    import java.util.Collections; 
    import java.util.LinkedList; 
    import java.util.List; 

    public class BlockingQueue { 

    //private static Long maxInMenorySize = 1L; 
    private static Long minFlushSize = 3L; 

    private static String baseDirectory = "/test/code/cache/"; 
    private static String fileNameFormat = "Table-"; 

    private static String currentWriteFile = ""; 

    private static List<Object> currentQueue = new LinkedList<Object>(); 
    private static List<Object> lastQueue = new LinkedList<Object>(); 

    static{ 
     try { 
      load(); 
     } catch (IOException e) { 
      System.out.println("Unable To Load"); 
      e.printStackTrace(); 
     } 
    } 

    private static void load() throws IOException{ 
     File baseLocation = new File(baseDirectory); 
     List<String> fileList = new ArrayList<String>(); 

     for(File entry : baseLocation.listFiles()){ 
      if(!entry.isDirectory() && entry.getName().contains(fileNameFormat)){ 
       fileList.add(entry.getAbsolutePath()); 
      } 
     } 

     Collections.sort(fileList); 

     if(fileList.size()==0){ 
      //currentQueue = lastQueue = new ArrayList<Object>(); 
      currentWriteFile = baseDirectory + "Table-1"; 
      BufferedWriter writer = new BufferedWriter(new FileWriter(currentWriteFile)); 
      while (!lastQueue.isEmpty()){ 
       writer.write(lastQueue.get(0).toString()+ "\n"); 
       lastQueue.remove(0); 
      } 
      writer.close(); 
     }else{ 
      if(fileList.size()>0){ 
        BufferedReader reader = new BufferedReader(new FileReader(fileList.get(0))); 
        String line=null; 
        while ((line=reader.readLine())!=null){ 
         currentQueue.add(line); 
        } 
        reader.close(); 
        File toDelete = new File(fileList.get(0)); 
        toDelete.delete(); 
      } 

      if(fileList.size()>0){ 
       BufferedReader reader = new BufferedReader(new FileReader(fileList.get(fileList.size()-1))); 
       currentWriteFile = fileList.get(fileList.size()-1); 
       String line=null; 
       while ((line=reader.readLine())!=null){ 
        lastQueue.add(line); 
       } 
       reader.close(); 
       //lastFileNameIndex=Long.parseLong(fileList.get(fileList.size()).substring(6, 9)); 
      } 
     } 

    } 

    private void loadFirst() throws IOException{ 
     File baseLocation = new File(baseDirectory); 
     List<String> fileList = new ArrayList<String>(); 

     for(File entry : baseLocation.listFiles()){ 
      if(!entry.isDirectory() && entry.getName().contains(fileNameFormat)){ 
       fileList.add(entry.getAbsolutePath()); 
      } 
     } 

     Collections.sort(fileList); 

     if(fileList.size()>0){ 
       BufferedReader reader = new BufferedReader(new FileReader(fileList.get(0))); 
       String line=null; 
       while ((line=reader.readLine())!=null){ 
        currentQueue.add(line); 
       } 
       reader.close(); 
       File toDelete = new File(fileList.get(0)); 
       toDelete.delete(); 
     } 
    } 

    public Object pop(){ 
     if(currentQueue.size()>0) 
      return currentQueue.remove(0); 

     if(currentQueue.size()==0){ 
      try { 
       loadFirst(); 
      } catch (IOException e) { 
       // TODO Auto-generated catch block 
       e.printStackTrace(); 
      } 
     } 

     if(currentQueue.size()>0) 
      return currentQueue.remove(0); 
     else 
      return null; 
    } 

    public synchronized Object waitTillPop() throws InterruptedException{ 
     if(currentQueue.size()==0){ 
      try { 
       loadFirst(); 
      } catch (IOException e) { 
       // TODO Auto-generated catch block 
       e.printStackTrace(); 
      } 
      if(currentQueue.size()==0) 
       wait(); 
     } 
     return currentQueue.remove(0); 
    } 

    public synchronized void push(Object data) throws IOException{ 
     lastQueue.add(data); 
     this.notifyAll(); 
     if(lastQueue.size()>=minFlushSize){ 
      BufferedWriter writer = new BufferedWriter(new FileWriter(currentWriteFile)); 
      while (!lastQueue.isEmpty()){ 
       writer.write(lastQueue.get(0).toString() + "\n"); 
       lastQueue.remove(0); 
      } 
      writer.close(); 

      currentWriteFile = currentWriteFile.substring(0,currentWriteFile.indexOf("-")+1) + 
        (Integer.parseInt(currentWriteFile.substring(currentWriteFile.indexOf("-")+1,currentWriteFile.length())) + 1); 
     } 
    } 

    public static void main(String[] args) { 
     try { 
      BlockingQueue bq = new BlockingQueue(); 

      for(int i =0 ; i<=8 ; i++){ 
       bq.push(""+i); 
      } 

      System.out.println(bq.pop()); 
      System.out.println(bq.pop()); 
      System.out.println(bq.pop()); 

      System.out.println(bq.waitTillPop()); 
      System.out.println(bq.waitTillPop()); 
      System.out.println(bq.waitTillPop()); 
      System.out.println(bq.waitTillPop()); 



     } catch (Exception e) { 
      e.printStackTrace(); 
     } 
    }