在Java中动态创建异步消息队列

Question

我需要在Java中动态创建异步消息队列。我的用例是通过多个SMTP服务器发送电子邮件：我需要按顺序强制执行相同SMTP服务器的电子邮件，但可以同时处理发送到不同SMTP服务器的电子邮件。我以前使用过JMS，但据我所知它只允许编译时队列创建，而我需要在运行时创建队列（每个SMTP服务器一个队列）。

我是否遗漏了有关JMS的内容或是否有其他工具/建议我应该查看？

Answer 1

我同意Adam，用例听起来像JMS是开销。 Java内置功能足够：

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本身在这个问题上相当沉默。大多数实现允许您执行此操作，而不是通过JMS本身，而是使用自己的API。但是，您无法将像MDB这样的正式内容连接到动态队列。相反，你需要管理自己的连接和听众。

Answer 3

我们最后一次在WebSphere环境中查看它时，动态创建队列是非常困难/不可能的（我认为临时队列对你来说太短暂了）。虽然存在用于创建队列的API，但是之后需要重新启动服务器才能变为活动状态。然后就是MDB问题。

基于所谓的问题可以通过额外的间接级别来解决所有问题，这假设可用的打印机组相对较小。

创建队列Printer01到Printer99（或更小的数字）。有一个<！>“数据库<！>”;将队列映射到真正的打印机。随着打印机的请求出现，您可以添加到映射表。您可能会有一些MDB查看永远不会使用的队列的开销，但除非你的潜在数量很多的打印机，否则你可以负担得起吗？

Answer 4

为每个SMTP服务器和限制队列使用者（MDB或消息监听器）创建一个队列为1

Answer 5

我用activemq完成了这个 - 我当时实际上已经发布了一个问题，因为我有类似的问题（当时的JMS文档声明这不受支持），并且确信它得到了支持。 / p>