如何为Jboss 5上的主题创建持久订阅者？

Question

我正在尝试使用持久订户编写主题。我得到了我的基本主题：

<?xml version="1.0" encoding="UTF-8"?> 
<server> 
    <mbean code="org.jboss.jms.server.destination.TopicService" name="jboss.messaging.destination:service=Topic,name=durableTopic" xmbean-dd="xmdesc/Topic-xmbean.xml"> 
    <depends optional-attribute-name="ServerPeer">jboss.messaging:service=ServerPeer</depends> 
    <depends>jboss.messaging:service=PostOffice</depends> 
</mbean> 
</server> 

而且我得到了我的订阅MDB：

@MessageDriven(activationConfig = { 
@ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Topic"), 
@ActivationConfigProperty(propertyName = "destination", propertyValue = "topic/durableTopic"), 
@ActivationConfigProperty(propertyName = "subscriptionDurability", propertyValue = "Durable") }) 
public class DurableSubscriberOne implements MessageListener { 
// ... 

但是，当我进入了JMX的控制台或管理控制台我看到我的话题一个非持久订阅和无持久订阅。

我是否犯了一些错字或一些小错误，还是比这更棘手？我正在使用JBoss 5.1.0.GA.

Answer 1

有过同样的问题，我最终设法使MDB通过增加两个@ActivationConfigProperty创建一个持久订阅：

@ActivationConfigProperty(propertyName = "subscriptionName", propertyValue ="SomeSubscriptionName") 
@ActivationConfigProperty(propertyName = "clientId", propertyValue ="SomeClientId") 

Answer 2

我真的找不到任何错误。代码只是为我工作。我包括我的代码示例以及截图供您参考。

package com.jboss.example; 

import javax.ejb.ActivationConfigProperty; 
import javax.ejb.MessageDriven; 
import javax.jms.Message; 
import javax.jms.MessageListener; 

/** 
* Message-Driven Bean implementation class for: DurableMessageListener 
* 
*/ 
@MessageDriven(activationConfig = { 
     @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Topic"), 
     @ActivationConfigProperty(propertyName = "destination", propertyValue = "topic/durableTopic"), 
     @ActivationConfigProperty(propertyName = "subscriptionDurability", propertyValue = "Durable") }) 
//, mappedName = "durableTopic") 
public class DurableMessageListener implements MessageListener { 

    /** 
    * Default constructor. 
    */ 
    public DurableMessageListener() { 
     // TODO Auto-generated constructor stub 
    } 

    /** 
    * @see MessageListener#onMessage(Message) 
    */ 
    public void onMessage(Message message) { 
     // TODO Auto-generated method stub 
     System.out.println("Received Message " + message); 
    } 

} 

主题订阅者样本

package com.jboss.example; 

import java.util.Properties; 

import javax.jms.JMSException; 
import javax.jms.Message; 
import javax.jms.Topic; 
import javax.jms.TopicConnection; 
import javax.jms.TopicConnectionFactory; 
import javax.jms.TopicSession; 
import javax.jms.TopicSubscriber; 
import javax.naming.InitialContext; 
import javax.naming.NamingException; 

public class DurableTopicSubscriber { 
    TopicConnection conn = null; 
    TopicSession session = null; 
    Topic topic = null; 

    public void setupPubSub() throws JMSException, NamingException { 

     Properties env = new Properties(); 
     env.setProperty("java.naming.factory.initial", 
       "org.jnp.interfaces.NamingContextFactory"); 
     env.setProperty("java.naming.factory.url.pkgs", 
       "org.jboss.naming:org.jnp.interfaces"); 
     env.setProperty("java.naming.provider.url", "jnp://localhost:1099"); 
     InitialContext iniCtx = new InitialContext(env); 
     Object tmp = iniCtx.lookup("ConnectionFactory"); 

     TopicConnectionFactory tcf = (TopicConnectionFactory) tmp; 
     conn = tcf.createTopicConnection("guest", "guest"); 
     conn.setClientID("Dirabla"); 
     topic = (Topic) iniCtx.lookup("topic/durableTopic"); 

     session = conn.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE); 
     conn.start(); 
    } 

    public void recvSync() throws JMSException, NamingException { 
     System.out.println("Begin recvSync"); 
     // Setup the pub/sub connection, session 
     setupPubSub(); 
     // Wait upto 5 seconds for the message 
     TopicSubscriber recv = session.createSubscriber(topic); 
     //TopicSubscriber recv = session.createDurableSubscriber(topic, "durableTopicName"); 
     Message msg = recv.receive(5000); 
     while (msg != null) { 
      System.out.println("DurableTopicClient.recv, msgt=" + msg); 
      msg = recv.receive(5000); 
     } 
    } 

    public void stop() throws JMSException { 
     conn.stop(); 
     session.close(); 
     conn.close(); 
    } 

    public static void main(String args[]) throws Exception { 
     System.out.println("Begin DurableTopicRecvClient, now=" 
       + System.currentTimeMillis()); 
     DurableTopicSubscriber client = new DurableTopicSubscriber(); 
     client.recvSync(); 
     client.stop(); 
     System.out.println("End DurableTopicRecvClient"); 
     System.exit(0); 
    } 

} 

主题示例发布

package com.jboss.example; 

import java.util.Properties; 

import javax.jms.JMSException; 
import javax.jms.Message; 
import javax.jms.Topic; 
import javax.jms.TopicConnection; 
import javax.jms.TopicConnectionFactory; 
import javax.jms.TopicPublisher; 
import javax.jms.TopicSession; 
import javax.naming.InitialContext; 
import javax.naming.NamingException; 

public class DurableTopicPublisher { 
    TopicConnection conn = null; 
    TopicSession session = null; 
    Topic topic = null; 

    public void setupPubSub() throws JMSException, NamingException { 

     Properties env = new Properties(); 
     env.setProperty("java.naming.factory.initial", 
       "org.jnp.interfaces.NamingContextFactory"); 
     env.setProperty("java.naming.factory.url.pkgs", 
       "org.jboss.naming:org.jnp.interfaces"); 
     env.setProperty("java.naming.provider.url", "jnp://localhost:1099"); 
     InitialContext iniCtx = new InitialContext(env); 
     Object tmp = iniCtx.lookup("ConnectionFactory"); 

     TopicConnectionFactory tcf = (TopicConnectionFactory) tmp; 
     conn = tcf.createTopicConnection("guest", "guest"); 
     conn.setClientID("Dirabla"); 
     topic = (Topic) iniCtx.lookup("topic/durableTopic"); 

     session = conn.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE); 
    } 

    public void recvSync() throws JMSException, NamingException { 
     System.out.println("Begin recvSync"); 
     setupPubSub(); 
     TopicPublisher topicPublisher = session.createPublisher(topic); 

     Message message = session.createMessage(); 

     for (int i = 0; i < 10; i++) { 
      message.setIntProperty("id", i); 
      topicPublisher.publish(message); 
     } 
    } 

    public void stop() throws JMSException { 
     conn.stop(); 
     session.close(); 
     conn.close(); 
    } 

    public static void main(String args[]) throws Exception { 
     System.out.println("Begin DurableTopicRecvClient, now=" 
       + System.currentTimeMillis()); 
     DurableTopicPublisher client = new DurableTopicPublisher(); 
     client.recvSync(); 
     client.stop(); 
     System.out.println("End DurableTopicRecvClient"); 
     System.exit(0); 
    } 

} 

主题的声明是一样的你

<?xml version="1.0" encoding="UTF-8"?> 
<server> 
    <mbean code="org.jboss.jms.server.destination.TopicService" name="jboss.messaging.destination:service=Topic,name=durableTopic" xmbean-dd="xmdesc/Topic-xmbean.xml"> 
    <depends optional-attribute-name="ServerPeer">jboss.messaging:service=ServerPeer</depends> 
    <depends>jboss.messaging:service=PostOffice</depends> 
</mbean> 
</server> 

截图