Creazione dinamica di code di messaggi asincrone in Java

Question

Devo creare code di messaggi asincrone in modo dinamico in Java. Il mio caso d'uso è l'invio di e-mail tramite più server SMTP: devo imporre che le e-mail allo stesso server SMTP siano processi in sequenza, ma le e-mail a server SMTP diversi possono essere elaborate contemporaneamente. Ho usato JMS in passato, ma per quanto posso vedere consente solo la creazione di code in fase di compilazione, mentre ho bisogno di creare code in fase di esecuzione (una coda per ciascun server SMTP).

Mi manca qualcosa riguardo a JMS o c'è qualche altro strumento / proposta a cui dovrei dare un'occhiata?

Answer 1

Sono d'accordo con Adam, il caso d'uso sembra che JMS sia sovraccarico. Funzionalità integrata Java 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 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 l'abbiamo visto in un ambiente WebSphere è stato sorprendentemente difficile / impossibile creare code dinamicamente (penso che le code temporanee siano troppo transitorie per te). Sebbene esistessero API per la creazione di code, per attivarsi è stato necessario riavviare successivamente il server. Poi c'è il problema MDB a cui si fa riferimento.

Che ne dici di una sporca soluzione alternativa basata sull'adagio secondo cui tutti i problemi possono essere risolti con un ulteriore livello di riferimento indiretto, il che presuppone che il set di stampanti disponibili sia relativamente piccolo.

Crea code da Printer01 a Printer99 (o un numero più piccolo). Avere un " database " che mappa le code su stampanti reali. Man mano che arrivano le richieste di stampanti, è possibile aggiungere alla tabella di mappatura. Potresti avere un certo sovraccarico di MDB che guardano le code che non verranno mai utilizzate, ma a meno che il tuo numero significativo di stampanti sia vasto, forse te lo puoi permettere?

Answer 4

Crea una coda per ciascuno dei tuoi server SMTP e limita il consumatore della coda (MDB o un listener di messaggi) a 1

Answer 5

L'ho fatto con activemq: al momento avevo pubblicato una domanda al riguardo, dato che avevo preoccupazioni simili (la documentazione JMS all'epoca affermava che non era supportata) e mi assicurava che era supportata.