Creación dinámica de colas de mensajes asíncronos en Java

Question

Necesito crear colas de mensajes asíncronos dinámicamente en Java. Mi caso de uso es el envío de correo electrónico a través de múltiples servidores SMTP: necesito hacer cumplir que los correos electrónicos al mismo servidor SMTP son procesos secuencialmente, pero los correos electrónicos a diferentes servidores SMTP se pueden procesar de manera concurrente. He usado JMS en el pasado, pero hasta donde puedo ver, solo permite la creación de colas en tiempo de compilación, mientras que necesito crear colas en el tiempo de ejecución (una cola para cada servidor SMTP).

¿Me falta algo con respecto a JMS o hay alguna otra herramienta/propuesta que debería tener en cuenta?

Answer 1

Estoy de acuerdo con Adam, el caso de uso parece que JMS está sobrecargado. Funcionalidad incorporada de Java suficiente:

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 en sí mismo como una especificación es bastante silencioso sobre el tema. La mayoría de las implementaciones le permiten hacer esto, pero no a través de JMS, sino con su propia API. Pero no podrá conectar algo formal como un MDB a una cola dinámica. En cambio, necesitarás administrar tus propias conexiones y oyentes.

Answer 3

La última vez que observamos esto en un entorno de WebSphere fue sorprendentemente difícil/imposible crear colas de forma dinámica (creo que las colas temporales son demasiado transitorias para usted). Aunque las API para crear colas existían, requerían un reinicio posterior del servidor para activarse. Luego está el problema de MDB al que se usa.

¿Qué tal un trabajo sucio basado en el adagio de que todos los problemas se pueden resolver con un nivel adicional de indirección, que asume que el conjunto de impresoras disponibles es comparativamente pequeño.

Cree colas Printer01 a Printer99 (o un número más pequeño). Tener una "base de datos" que mapea las colas en impresoras reales. A medida que aparecen las solicitudes de impresoras, puede agregarlas a la tabla de asignación. Es posible que tengas algunos gastos generales de los bancos multilaterales de desarrollo que buscan colas que nunca se utilizarán, pero a menos que tu número potencial de impresoras sea grande, ¿quizás te lo puedes permitir?

Answer 4

Crear una cola para cada uno de su Sever SMTP y límite de la cola de los consumidores (MDB o un receptor de mensajes) a 1

Answer 5

que he hecho esto con activemq - En realidad publicado una pregunta sobre esto en el tiempo, como Tenía preocupaciones similares (la documentación de JMS en ese momento indicaba que esto no era compatible) y me aseguraron que era compatible.