ActiveMQ 抛出 NullPointerException

Question

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

线程“ActiveMQ Journal Checkpoint Worker”中的异常 java.lang.NullPointerException 在 org.slf4j.impl.Log4jLoggerAdapter.debug(Log4jLoggerAdapter.java:209) 在 org.apache.activemq.store.kahadb.MessageDatabase.checkpointUpdate(MessageDatabase.java:1349) 在 org.apache.activemq.store.kahadb.MessageDatabase$10.execute(MessageDatabase.java:814) 在 org.apache.kahadb.page.Transaction.execute（Transaction.java:769） 在 org.apache.activemq.store.kahadb.MessageDatabase.checkpointCleanup(MessageDatabase.java:812) 在 org.apache.activemq.store.kahadb.MessageDatabase$3.run(MessageDatabase.java:324)

线程“ActiveMQ Broker[localhost] Scheduler”中的异常 java.lang.NullPointerException 在 org.slf4j.impl.Log4jLoggerAdapter.isDebugEnabled(Log4jLoggerAdapter.java:199) 在 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);

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

请帮我解决这个问题。

谢谢，

卡西克扬

Answer 1

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