BlazeDS JMS 适配器断开连接

Question

我将 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 应用程序。

谁能给我另一个解决方案？

这是我的 blazeds 配置

  <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 主题。 通过这种方式，我可以使用相同的 clientID 重新创建一个新主题并接收消息。

这里是我实现的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());
        }
    }