java - 断开与 weblogic JMS 的连接

Question

现在我与 weblogic JMS 服务器的断开连接看起来像这样

import java.util.Hashtable;
import javax.jms.*;
import javax.naming.*;
import javax.transaction.*;
import java.util.Vector;
import javax.rmi.PortableRemoteObject;
import clojure.java.api.Clojure;
import clojure.lang.IFn;
import org.apache.log4j.Logger;
import weblogic.jndi.*;

public class WebLogicListener implements MessageListener, ExceptionListener{
  public InitialContext ctx;
  public TopicConnectionFactory conFactory;
  public TopicConnection tCon;
  public TopicSession tSession;
  public TopicSubscriber tSub;
  public Boolean development;
  public Topic topic;
  /*clojure function objects*/
  public IFn publish;
  public IFn close;
  public IFn incrementMetric;
  public IFn logMessage;
  public IFn resync;

  public Object channel;
  public ExceptionListener exception;
  public String topicName;
  public String subName;
  public String username;
  public String password;
  public String clientId;
  public String factoryJNDI;
  public String topicJNDI;
  public Vector nms;
  public Hashtable<Object,Object> env;
  public boolean running = false;

  public WebLogicListener (String topicName, String host, String username, String password, String factoryJNDI,
                           String topicJNDI, String clientId, String subName, String ns, String fnName,
                           boolean development, Vector nms){
    this.username = username;
    this.password = password;
    this.clientId = clientId;
    this.topicName = topicName;
    this.subName = subName;
    this.development = development;
    this.topicJNDI = topicJNDI;
    this.factoryJNDI = factoryJNDI;
    this.nms = nms;
    /*Clojure interop handlers*/
    IFn chan = Clojure.var("clojure.core.async", "chan");
    resync = Clojure.var("cenx.baldr.api", "resync!");
    publish = Clojure.var(ns, fnName);
    incrementMetric = Clojure.var(ns, "log-metric");
    logMessage = Clojure.var (ns, "log-message");
    close = Clojure.var("clojure.core.async","close!");
    /*populate envrionment*/
    env = new Hashtable<Object,Object>();
    env.put(Context.PROVIDER_URL, host);
    env.put(Context.INITIAL_CONTEXT_FACTORY, "weblogic.jndi.WLInitialContextFactory");
    env.put(Context.SECURITY_PRINCIPAL, username);
    env.put(Context.SECURITY_CREDENTIALS, password);
    env.put("weblogic.jndi.createIntermediateContexts", "true");
    /*open communication channel for clojure daemon*/
    channel = chan.invoke();
  }

  private void initListener() throws JMSException, NamingException{
    try{
      if (!running && !development){
        ctx = new InitialContext(env);
        topic = (Topic) ctx.lookup(topicJNDI);
        conFactory =  (TopicConnectionFactory)PortableRemoteObject.narrow(ctx.lookup(factoryJNDI), TopicConnectionFactory.class);
        tCon = (TopicConnection) conFactory.createTopicConnection();
        tCon.setExceptionListener(this);
        tCon.setClientID(clientId);
        tSession = (TopicSession) tCon.createTopicSession(false, 1);
        tSub = tSession.createDurableSubscriber(topic, subName);
        tSub.setMessageListener(this);
        tCon.start();
        running = true;
      }else{
        if (running){
          logMessage.invoke("error", String.format("Listener is already running"));
        }
        if (development){
          logMessage.invoke("info", "Running in development mode, no connection established");
        }
      }
    } catch(Exception e){
      logMessage.invoke("error", String.format("Unable to start listener \n %s", e.toString()));
    }
  }

  public void startListener(){
    if (!development && env != null){
      try {
        initListener();
      }catch(Exception e){
        logMessage.invoke("error", String.format("Unable to start Listener \n %s", e.toString()));
      }
    } else {
      if (development){
        logMessage.invoke("info", "Running in development mode, no connection established");
      }
      if (env == null){
        logMessage.invoke("error", "Environment variable is null");
      }
    }
  }

  ///Closes the JMS connection and the channel
  public void stopListener(){
    if (!development){
      try{
        tSub.close();
        tSession.close();
        tCon.close();
        incrementMetric.invoke("JMS-disconnect-count");
      }catch(Exception e){
        logMessage.invoke("error", String.format("Error while stopping the listener \n %s", e.toString()));
      }finally{
        running = false;
      }
    } else {
      logMessage.invoke("info", "Listener not started, running in development mode");
    }
  }

  public Object getChannel(){
    return channel;
  }

  //re-initializes the channel in case of error
  public void initializeChannel(){
    if (channel == null){
      IFn chan = Clojure.var("clojure.core.async", "chan");
      channel = chan.invoke();
    } else {
      logMessage.invoke("info", "Channel is already initialized");
    }
  }
  //accessors for debugging

  public void closeSubscription(){
    try{
      tSub.close();
    }catch (Exception e){
      logMessage.invoke("error", "unable to close topic subscription");
      logMessage.invoke("error", e.toString());
    }
  }

  public void closeSession(){
    try{
      tSession.unsubscribe(subName);
      tSession.close();
    }catch (Exception e){
      logMessage.invoke("error", "unable to close topic session");
      logMessage.invoke("error", e.toString());
    }
  }

  public void closeConnection(){
    try{
      tCon.close();
    }catch (Exception e){
      logMessage.invoke("error", "unable to close topic connection");
      logMessage.invoke("error", e.toString());
    }
  }

  public void closeContext(){
    try {
      ctx.close();
    }catch (Exception e){
      logMessage.invoke("error", "unable to close context");
      logMessage.invoke("error", e.toString());
    }
  }

  public Boolean isRunning(){
    return running;
  }

  public Context getContext (){
    return ctx;
  }

  public TopicConnectionFactory getFactory (){
    return conFactory;
  }

  public TopicConnection getTopicConnection (){
    return tCon;
  }

  public TopicSession getTopicSession (){
    return tSession;
  }

  public Boolean getDevelopmentMode(){
    return development;
  }

  public TopicSubscriber getTopicSubscriber (){
    return tSub;
  }

  public Topic getTopic (){
    return topic;
  }

  /*Interface methods*/

  public void onMessage(Message message){
    publish.invoke(channel, message);
  }
  /*attempt a resync after an exception connection*/
  private void resync(){
    resync.invoke(nms);
  }

  private void attemptReconnect() throws Exception{
    if (!development){
      //clean up any portions of the connection that managed to establish
      stopListener();
      //incase of stopListener exceptioning out set running to false
      running = false;
      do{
        try{
          initListener();
          if (running){
            resync();
          }
        }catch(Exception e){
          logMessage.invoke("error",
                            String.format("Unable to establish connection to JMS server \n %s", e.toString()));
        }finally{
          Thread.sleep(30000);
        }
      } while (!running);
    } else {
      logMessage.invoke("info", "Running in development mode, no connection established");
    }
  }

  public void onException(JMSException e){
    logMessage.invoke("error",
                      String.format("A JMS Exception has occurred, attempting to re-establish topic connection \n %s", e.toString()));
    try{
      incrementMetric.invoke("JMS-disconnect-count");
      attemptReconnect();
    }catch(Exception g){
      logMessage.invoke("error",
                        String.format("Unable to start Listener \n %s", g.toString()));
    }
  }

  /* Test functions */
  public void testException() throws JMSException{
    onException(new JMSException("testing exception function"));
  }

  public void testChannel (String message){
    if (development){
      publish.invoke(channel, message);
    }
  }
}

当我创建连接时，我使用 netstat 检查服务器是否已连接

netstat -an | grep 8001 tcp 0 0 ip-address:59730
ip-address:8001 ESTABLISHED

然后我除了调用 .closeContext 方法外还调用了我的 .stopListener 并返回使用 netstat 再次检查我的连接，我得到了相同的结果

netstat -an | grep 8001 tcp 0 0 ip-address:59730
ip-address:8001 ESTABLISHED

为什么关闭 session 、订阅者和连接不会破坏与 JMS 服务器的连接。我找到的文档没有给我任何关于为什么我不能完全破坏连接的解释。

Answer 1

我不确定您是否正确处理了这个问题。我看到您在连接上有一个异常监听器。

在 weblogic 上，每个错误事件都会多次调用监听器，因此您不应在每次调用时都尝试重新连接。它会为您注册的每个消费者调用一次，并为它监控的每个连接调用一次。仅当异常表示 ServerConnectionLost 时才应断开连接。

此外，在错误处理程序中，您只需关闭连接即可。如果你执行了 connection.close()，那也会关闭 session 和监听器。无需像您一样以相反的顺序关闭它们。

还有一件事。您的生产代码中不应包含“开发”或“调试”或“测试”代码。

那部分说“if (!development && env != null){” ...你不应该那样做。

现在回到你的问题，为什么实际连接没有关闭。我看你在干什么

try{
    tSub.close();
    tSession.close();
    tCon.close();
    incrementMetric.invoke("JMS-disconnect-count");
} catch...

如果 tSub.close() 或 tSession.close() 出错，您的连接将永远不会关闭。将每个包装在一个独立的 try/catch 中。