1

我在我的应用程序中使用 ActiveMQ。它有时表现得很奇怪，当我启动 Tomcat 服务器并让它闲置几分钟时，我得到以下异常：

在 org.apache.activemq.store.kahadb.MessageDatabase.checkpointUpdate(MessageDatabase.java: 1349) 在 org.apache.kahadb.page.Transaction.execute(Transaction.java:769) 在 org.apache.activemq 的 org.apache.activemq.store.kahadb.MessageDatabase$10.execute(MessageDatabase.java:814)。 store.kahadb.MessageDatabase.checkpointCleanup(MessageDatabase.java:812) 在 org.apache.activemq.store.kahadb.MessageDatabase$3.run(MessageDatabase.java:324)

在 org.apache.activemq.broker.region.Queue.expireMessages(Queue. java:816) 在 org.apache.activemq.broker.region.Queue.access$100(Queue.java:96) 在 org.apache.activemq.broker.region.Queue$2.run(Queue.java:136) 在 org .apache.activemq.thread.SchedulerTimerTask.run(SchedulerTimerTask.java:33) 在 java.util.TimerThread.mainLoop(Timer.java:512) 在 java.util.TimerThread.run(Timer.java:462)

在此之后，当再次调用队列时，它会引发异常：

捕获：无法创建传输。原因：javax.management.InstanceAlreadyExistsException：org.apache.activemq:BrokerName=localhost,Type=Broker javax.jms.JMSException：无法创建传输。原因：javax.management.InstanceAlreadyExistsException：org.apache.activemq:BrokerName=localhost,Type=Broker

我使用了队列生产者和消费者，如下所示：

制片人：

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                "vm://localhost");

        // Create a Connection
        Connection connection = connectionFactory.createConnection();
        connection.start();

        // Create a Session
        Session session = connection.createSession(false,
                Session.AUTO_ACKNOWLEDGE);

        // Create the destination (Topic or Queue)
        Destination destination = session.createQueue(queuename);

        // Create a MessageProducer from the Session to the Topic or Queue
        MessageProducer producer = session.createProducer(destination);
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

        // Create a messages
        Iterator it = mapMessage.entrySet().iterator();
        MapMessage message = session.createMapMessage();
        while (it.hasNext()) {
            Map.Entry pairs = (Map.Entry)it.next();
            message.setString(""+pairs.getKey(), ""+pairs.getValue());
            System.out.println(pairs.getKey() + " = " + pairs.getValue());
            it.remove(); // avoids a ConcurrentModificationException
        }

        // Tell the producer to send the message
        System.out.println("Sent message: "+ message);
        producer.send(message);

        // Clean up
        session.close();
        connection.close();

消费者：

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                "vm://localhost");

        // Create a Connection
        Connection connection = connectionFactory.createConnection();
        connection.start();

        // Create a Session
        Session session = connection.createSession(false,
                Session.AUTO_ACKNOWLEDGE);

        // Create the destination (Topic or Queue)
        Destination destination = session.createQueue("register");

        // Create a MessageConsumer from the Session to the Topic or Queue
        MessageConsumer consumer = session.createConsumer(destination);
        consumer.setMessageListener(this);

ActiveMQ 集成：

PushRegisterConsumer prc = new PushRegisterConsumer();
    prc.start();

PushQueueProducer pmp = new PushQueueProducer();
pmp.queueProducer(AppConstants.QUEUE_NAME,registerDetails);

生产者和消费者已经整合如上图

请帮我解决这个问题。

谢谢，

卡西凯扬

4

1 回答 1

0

创建ActiveMQConnectionFactory一次，并将其传递给PushRegisterConsumerand的构造函数PushRegisterProducer。目前，似乎正在使用该名称创建 2 个嵌入式代理localhost- 您希望确保这只发生一次。

于 2013-08-09T09:41:40.283 回答