以编程方式配置 Apache ActiveMQ

Question

我正在使用 Apache ActiveMQ 对大量消息进行排队，然后在一天结束时将它们出列。不过，我对 ActiveMQ 的运作方式感到困惑。在我的 PC 上，我没有将 ActiveMQ 安装为服务，也没有在某处安装服务器。我刚刚将“activemq-all-5.14.5.jar”作为 Maven 依赖项包含在我的项目中，到目前为止我正在使用以下代码：

public static void main(String[] args) throws URISyntaxException, Exception {
    BrokerService broker = BrokerFactory.createBroker(new URI("broker:(tcp://localhost:4848)"));
    broker.start();
    Connection connection = null;
    try {
        // Producer
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:4848");
        connection = connectionFactory.createConnection();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue("customerQueue");
        String payload = "Important Task";
        Message msg = session.createTextMessage(payload);
        MessageProducer producer = session.createProducer(queue);
        System.out.println("Sending text '" + payload + "'");
        msg.setLongProperty("_AMQ_SCHED_DELIVERY", System.currentTimeMillis() + 5000L);
        producer.send(msg);

        // Consumer
        MessageConsumer consumer = session.createConsumer(queue);
        connection.start();
        QueueBrowser browser = session.createBrowser(queue);
        while (browser.getEnumeration().hasMoreElements()) {
            TextMessage textMsg = (TextMessage) consumer.receive();
            browser.getEnumeration().nextElement();
            System.out.println(textMsg);
            System.out.println("Received: " + textMsg.getText());
        }

        session.close();
    } finally {
        if (connection != null) {
            connection.close();
        }
        broker.stop();
    }
}

如您所见，我想将消息延迟 5 秒（或更多，可能会有所不同），但在我找到的每个指南中，我都被指示配置 XML 配置文件。但是，这是在将 ActiveMQ 作为服务运行时使用的文件。我目前正在使用只是 jar 库。

最初，我安装了 Glassgfish 服务器以使用 JMS 来对所有消息进行排队，但从那以后我放弃了该项目，但仍然使用 ActiveMQ (localhost:4848) 的 IP。

请注意，以下是一个完美运行的示例 - KahaDB 也被用于在服务器发生故障时存储消息。

就我而言，ActiveMQ 确实从我正在运行此代码的 STS 启动本地服务器，但配置文件在哪里？我可以以编程方式更改其属性吗？

Answer 1

这应该可以工作（适用于我们使用 ActiveMQ 5.12.3）。请务必先清理 KahaDB 存储，以避免读取以前的消息。

public static void main(String[] args) throws Exception {
    BrokerService broker = BrokerFactory.createBroker(new URI("broker:(tcp://localhost:4848)"));
    broker.setSchedulerSupport(true);
    broker.start();
    Connection connection = null;
    try {
        // Producer
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:4848");
        connection = connectionFactory.createConnection();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue("customerQueue");
        String payload = "Important Task";
        Message msg = session.createTextMessage(payload);
        MessageProducer producer = session.createProducer(queue);
        System.out.println("Sending text '" + payload + "'");
        msg.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 5000L);
        producer.send(msg);
        connection.start();

        // Consumer
        MessageConsumer consumer = null;
        consumer = session.createConsumer(queue);

        QueueBrowser browser = session.createBrowser(queue);
        while (browser.getEnumeration().hasMoreElements()) {
            TextMessage textMsg = (TextMessage) consumer.receive();
            browser.getEnumeration().nextElement();
            System.out.println(textMsg);
            System.out.println("Received: " + textMsg.getText());
        }

        session.close();
    } finally{
        if (connection != null) {
            connection.close();
        }
        broker.stop();
    }
}

第一次干净运行（KahaDB 存储为空）不应输出

“收到：重要任务”

，而第二个会，如果您不删除其间的数据文件。

删除线`

msg.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 5000L);

将使第一次干净运行输出“收到：重要任务”