1

我开发了 JMS 应用程序，现在我想添加支持代理重启的功能。我有动态主题，我应该在恢复连接后重新创建它们。此外，我应该有可能知道经纪人何时关闭以及何时回来。所以我尝试使用 ActiveMQ 故障转移协议来实现这个功能。我实现了 TransportListener 并在“transportInterrupted”方法中我称之为完全断开连接

  public void disconnect() throws JMSException {
    System.out.println("!!!!!!!DISCONNECTING!!!!!!!!");
    consumer.close();
    session.close();
    session = null;
    messageProducer.close();
    messageProducer = null;
    connection = null;
    connected = false;
    System.out.println("!!!!!!!DISCONNECTED!!!!!!!!");
}

在此之后，我的应用程序挂起，就像竞争条件一样。如果我只关闭（）生产者并将连接设置为 null 一切正常，但如果我尝试关闭消费者，它仅适用于 N 个案例中的 1 个。我写了证明它的测试。我认为关闭消费者的问题，但我没有找到任何信息我做错了什么。

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class FastFailProducer {
    volatile boolean connected = false;
    private ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("failover:(tcp://localhost:61616)?timeout=5000");;
    private static FailoverListener failoverListener;
    private Connection connection;
    private Session session;
    private Queue queue;
    private MessageProducer messageProducer;
    private MessageConsumer consumer;
    private String something;

public void init() throws JMSException {
    System.out.println("!!!!!!!CONNECTING!!!!!!!!");
    connection = factory.createConnection();
    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    connection.start();
    ((ActiveMQConnection) connection).addTransportListener(failoverListener);
    queue = session.createQueue("TEST");
    messageProducer = session.createProducer(queue);
    consumer = session.createConsumer(queue);
    System.out.println("!!!!!!!CONNECTING COMPLETE!!!!!!!!");
    connected = true;
}

public void disconnect() throws JMSException {
    System.out.println("!!!!!!!DISCONNECTING!!!!!!!!");
    consumer.close();
    session.close();
    session = null;
    messageProducer.close();
    messageProducer = null;
    connection = null;
    connected = false;
    System.out.println("!!!!!!!DISCONNECTED!!!!!!!!");
}

public void run() throws Exception {
    // send messages
    for (int i = 0; i < 1000; i++) {
        if (connected) {
            if (session != null & messageProducer != null & queue != null) {
                // send a message
                messageProducer.send(session.createTextMessage(i + " message"));
                System.out.println("Sent message " + i);
            }
        } else {
            // execute your backup logic
            System.out.println("Message " + i + " not sent");

        }
        Thread.sleep(1000);
    }

    messageProducer.close();
    session.close();
    connection.close();
    System.exit(0);
}

public static void main(String[] args) throws Exception {
    FastFailProducer failoverProducer = new FastFailProducer();
    failoverProducer.something = "asdfasdf";
    failoverListener = new FailoverListener(failoverProducer);
    failoverProducer.init();
    failoverProducer.setConnected(true);
    failoverProducer.run();

}

public boolean isConnected() {
    return connected;
}

public void setConnected(boolean connected) {
    this.connected = connected;
}
}

TransportListenerImpl 类

import java.io.IOException;

import javax.jms.JMSException;

import org.apache.activemq.transport.TransportListener;

public class FailoverListener implements TransportListener {
    private FastFailProducer failProducer;

public FailoverListener(FastFailProducer failProducer) {
    super();
    this.failProducer = failProducer;
}

@Override
public void onCommand(Object arg0) {
}

@Override
public void onException(IOException arg0) {

}

@Override
public void transportInterupted() {
    try {
        failProducer.disconnect();
    } catch (JMSException e) {
        e.printStackTrace();
    }
}

@Override
public void transportResumed() {
    System.out.println("!!!!!!!TRANSPORT RESUMED!!!!!!!!");
    if (!failProducer.isConnected()) {
        try {
            failProducer.init();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}
}
4

1 回答 1

3

我认为您错过了使用故障转移协议的意义。如果您使用故障转移，则无需关闭连接及其相关资源，因为故障转移传输将负责恢复客户端上的所有内容，就像代理发生故障之前一样。在事件方法中关闭连接肯定会锁定，因为您不应该做这样的事情。如果您想在代理消失时关闭所有内容，请不要使用故障转移，而改为侦听 JMS 异常侦听器事件挂钩。

于 2013-01-05T17:53:04.177 回答