BlazeDS JMS适配器断开连接

Question

I将Blazeds消息系统与ActiveMQ:集成。

我将BlazeDS配置为使用flex.messaging.services.messaging.adapters.JMSAdapter创建一个持久的flex目的地（配置如下）。

我能够创建一个注册到主题并接收消息的小型应用程序。由于我需要接收脱机时发送的消息，所以我创建了一个持久的查询。

只要我正确地取消订阅弹性消费者，一切都很好。

我的问题是当弹性消费者不打电话取消订阅时。例如当我关闭浏览器。

在这种情况下，主题仍然是活动的（我可以从ActiveMQ web控制台中看到它），并且它使用消息。

当我再次连接到flex应用程序的新实例时，连接正常，但我没有收到任何消息。也不是我不在的时候寄来的，也不是新来的。我甚至无法使用ActiveMQ web控制台删除主题：javax.jms.JMSException: Durable consumer is in use。

删除该主题的唯一解决方案是重新启动包含BlazeDS代理的Web应用程序。

有人能给我另一个解决方案吗？

这是我的火焰配置

  <adapters>
    <adapter-definition id="actionscript" class="flex.messaging.services.messaging.adapters.ActionScriptAdapter" default="true" />
    <adapter-definition id="jms" class="flex.messaging.services.messaging.adapters.JMSAdapter"/>
</adapters>
...
  <destination id="warehouse-topic-jms">
    <properties>
        <jms>
            <destination-type>Topic</destination-type>
            <message-type>javax.jms.ObjectMessage</message-type>
            <connection-factory>java:comp/env/jms/flex/TopicConnectionFactory</connection-factory>
            <destination-jndi-name>java:comp/env/jms/warehouse</destination-jndi-name>
            <delivery-mode>PERSISTENT</delivery-mode>
            <message-priority>DEFAULT_PRIORITY</message-priority>
            <acknowledge-mode>AUTO_ACKNOWLEDGE</acknowledge-mode>
            <initial-context-environment>
                <property>
                    <name>Context.INITIAL_CONTEXT_FACTORY</name>
                    <value>org.apache.activemq.jndi.ActiveMQInitialContextFactory</value>
                </property>
                <property>
                    <name>Context.PROVIDER_URL</name>
                    <value>tcp://localhost:61616</value>
                </property>
            </initial-context-environment>
        </jms>
        <server>
            <durable>true</durable>             
        </server>
    </properties>
    <adapter ref="jms"/>
</destination>

这是公开JNDI资源的Tomcat上下文

<Resource name="jms/flex/TopicConnectionFactory"
    type="org.apache.activemq.ActiveMQConnectionFactory"
    description="JMS Connection Factory"
    factory="org.apache.activemq.jndi.JNDIReferenceFactory"
    brokerURL="tcp://localhost:61616"
    brokerName="myBroker"/>
<Resource name="jms/warehouse"
    type="org.apache.activemq.command.ActiveMQTopic"
    description="warehouse.topic"
    factory="org.apache.activemq.jndi.JNDIReferenceFactory"
    physicalName="warehouse.topic"/>

这是我的弹性消费者

<mx:Consumer id="consumer"   
                 channelConnect="consumer_channelConnectHandler(event)" 
                 channelFault="consumer_channelFaultHandler(event)" destination="warehouse-topic-jms"
                 fault="consumer_faultHandler(event)" message="consumer_messageHandler(event)"/>

Answer 1

我找到的唯一解决方案是使用BlazeDS公开的JMX MBEAN来强制JMSAdapter删除ActiveMQ Topic。通过这种方式，我可以用相同的clientID重新创建一个新的Topic并接收消息。

这里是我实现的Java方法的代码

public boolean deleteTopicQueue(String clientId) throws Exception {
        if(clientId==null){
            throw new Exception("Error removing topic: null name provided");
        }
        clientId=clientId.trim();
        JMXServiceURL url = new JMXServiceURL(DEFAULT_JMX_ADAPTER_URL);
        try (JMXConnector jmxc = JMXConnectorFactory.connect(url, null)){
            MBeanServerConnection mbsc = jmxc.getMBeanServerConnection();
            ObjectName pattern = new ObjectName(DEFAULT_JMSADAPTER_MEAN_QUERY);
            Set<ObjectName>  names =
                new TreeSet<ObjectName>(mbsc.queryNames(pattern, null));
            if(names.size()<=0){
                logger.info("Error Removig topic "+clientId+": No JMSAdapter found ");
                throw new Exception("Error Removig topic "+clientId+": No JMSAdapter found ");
            }
            while(names.iterator().hasNext()){
                ObjectName ob =  names.iterator().next();
                JMSAdapterControlMBean obProxy =  JMX.newMXBeanProxy(mbsc, ob,JMSAdapterControlMBean.class);
                String[] consumerList=  obProxy.getTopicConsumerIds();
                for(String consumer :consumerList){
                    if(consumer.trim().equals(clientId)){
                        logger.info("Removing "+consumer+" from "+ob.getCanonicalName());
                        obProxy.removeConsumer(consumer);
                        return true;
                    }
                }
            }
            logger.debug("No consumer with ID "+clientId+" Found");
            return false;
        } catch (Exception e) {
            logger.info("Error Removig topic "+clientId+": No JMSAdapter found ");
            throw new Exception("Error Removig topic "+clientId+" :"+e.getMessage());
        }
    }