Creazione dinamica di code di messaggi asincroni in Java

Question

Ho bisogno di creare code di messaggi asincroni dinamicamente in Java. Il mio caso d'uso è l'invio di e-mail tramite più server SMTP: devo applicare le e-mail allo stesso server SMTP in modo sequenziale, ma le e-mail a diversi server SMTP possono essere elaborate contemporaneamente. Ho usato JMS in passato, ma per quanto posso vedere consente solo la creazione della coda in fase di compilazione, mentre ho bisogno di creare code in fase di runtime (una coda per ogni server SMTP).

Mi manca qualcosa per quanto riguarda JMS o c'è qualche altro strumento/proposta che dovrei dare un'occhiata?

Answer 1

Sono d'accordo con Adam, il caso d'uso suona come JMS è sovraccarico. Java built-in funzionalità sufficiente:

package de.mhaller; 

import java.util.ArrayDeque; 
import java.util.ArrayList; 
import java.util.Deque; 
import java.util.HashMap; 
import java.util.Map; 
import java.util.Queue; 
import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 
import java.util.concurrent.LinkedBlockingDeque; 

import org.junit.Assert; 
import org.junit.Test; 

public class Mailer { 

    @Test 
    public void testMailer() throws Exception { 
     ExecutorService executor = Executors.newCachedThreadPool(); 
     ArrayList<Mail> log = new ArrayList<Mail>(); 
     LinkedBlockingDeque<Mail> incoming = new LinkedBlockingDeque<Mail>(); 

     // TODO: Put mails to be sent into the incoming queue 
     incoming.offer(new Mail("foo1@localhost", "localhost")); 
     incoming.offer(new Mail("foo2@otherhost", "otherhost")); 
     incoming.offer(new Mail("foo3@otherhost", "otherhost")); 
     incoming.offer(new Mail("foo4@localhost", "localhost")); 

     Map<Mailserver, Queue<Mail>> queues = new HashMap<Mailserver, Queue<Mail>>(); 
     while (!incoming.isEmpty()) { 
      Mail mail = incoming.pollFirst(); 
      Mailserver mailserver = findMailserver(mail); 
      if (!queues.containsKey(mailserver)) { 
       ArrayDeque<Mail> serverQueue = new ArrayDeque<Mail>(); 
       queues.put(mailserver, serverQueue); 
       executor.execute(new SendMail(mailserver, serverQueue)); 
      } 
      Queue<Mail> slot = queues.get(mailserver); 
      slot.offer(mail); 
     } 

     assertMailSentWithCorrectServer(log); 
    } 

    private void assertMailSentWithCorrectServer(ArrayList<Mail> log) { 
     for (Mail mail : log) { 
      if (!mail.server.equals(mail.sentBy.mailserver)) { 
       Assert.fail("Mail sent by wrong server: " + mail); 
      } 
     } 
    } 

    private Mailserver findMailserver(Mail mail) { 
     // TODO: Your lookup logic which server to use 
     return new Mailserver(mail.server); 
    } 

    private static class Mail { 
     String recipient; 
     String server; 
     SendMail sentBy; 

     public Mail(String recipient, String server) { 
      this.recipient = recipient; 
      this.server = server; 
     } 

     @Override 
     public String toString() { 
      return "mail for " + recipient; 
     } 
    } 

    public static class SendMail implements Runnable { 

     private final Deque<Mail> queue; 
     private final Mailserver mailserver; 

     public SendMail(Mailserver mailserver, Deque<Mail> queue) { 
      this.mailserver = mailserver; 
      this.queue = queue; 
     } 

     @Override 
     public void run() { 
      while (!queue.isEmpty()) { 
       Mail mail = queue.pollFirst(); 
       // TODO: Use SMTP to send the mail via mailserver 
       System.out.println(this + " sent " + mail + " via " + mailserver); 
       mail.sentBy = this; 
      } 
     } 

    } 

    public static class Mailserver { 
     String hostname; 

     public Mailserver(String hostname) { 
      this.hostname = hostname; 
     } 

     @Override 
     public String toString() { 
      return hostname; 
     } 

     @Override 
     public int hashCode() { 
      return hostname.hashCode(); 
     } 

     @Override 
     public boolean equals(Object obj) { 
      return hostname.equals(((Mailserver) obj).hostname); 
     } 

    } 

} 

Answer 2

JMS stesso come una specifica è piuttosto silenzioso sul problema. La maggior parte delle implementazioni ti consente di farlo, non solo tramite JMS stesso, ma utilizzando la propria API. Ma non sarai in grado di collegare qualcosa di formale come un MDB a una coda dinamica. Piuttosto, dovrai gestire le tue connessioni e i tuoi ascoltatori.

Answer 3

L'ultima volta che abbiamo esaminato questo in un ambiente WebSphere è stato sorprendentemente difficile/impossibile creare code in modo dinamico (le code temporanee sono troppo transitorie per voi, credo). Sebbene esistessero API per la creazione di code, è stato necessario riavviare il server in seguito per diventare attivi. Poi c'è il problema MDB a cui si è abituati.

Che ne dite di un work-around sporco basato sul proverbio che tutti i problemi possono essere risolti da un livello aggiuntivo di riferimento indiretto, che presuppone che l'insieme di stampanti disponibili sia relativamente piccolo.

Creare code da Printer01 a Printer99 (o un numero inferiore). Avere un "database" che mappa le code alle stampanti reali. Poiché le richieste di stampanti arrivano, è possibile aggiungere alla tabella di mappatura. Potresti avere un sovraccarico di MDB che guardano le code che non saranno mai utilizzate, ma a meno che il tuo numero di stampanti sia vasto forse puoi permetterlo?

Answer 4

Creare una coda per ciascuna della vostra coda di server SMTP e il limite di consumo (MDB o un ascoltatore messaggio) a 1

Answer 5

ho fatto questo con ActiveMQ - in realtà ho postato una domanda su questo, al momento, come Avevo preoccupazioni simili (la documentazione JMS all'epoca affermava che questo non era supportato) ed era certo che fosse supportato.