我有一个充当订阅者的代码。我创建了持久订阅者。所以由于这个原因,我得到了例外
Exception in thread "main" javax.jms.JMSException: Error registering consumer: org.wso2.andes.AMQTimeoutException: Server did not respond in a timely fashion [error code 408: Request Timeout]
at org.wso2.andes.client.AMQSession$4.execute(AMQSession.java:2054)
at org.wso2.andes.client.AMQSession$4.execute(AMQSession.java:1997)
at org.wso2.andes.client.AMQConnectionDelegate_8_0.executeRetrySupport(AMQConnectionDelegate_8_0.java:305)
at org.wso2.andes.client.AMQConnection.executeRetrySupport(AMQConnection.java:621)
at org.wso2.andes.client.failover.FailoverRetrySupport.execute(FailoverRetrySupport.java:102)
at org.wso2.andes.client.AMQSession.createConsumerImpl(AMQSession.java:1995)
at org.wso2.andes.client.AMQSession.createConsumer(AMQSession.java:993)
at org.wso2.andes.client.AMQSession.createDurableSubscriber(AMQSession.java:1142)
at org.wso2.andes.client.AMQSession.createDurableSubscriber(AMQSession.java:1042)
at org.wso2.andes.client.AMQTopicSessionAdaptor.createDurableSubscriber(AMQTopicSessionAdaptor.java:73)
at xml.parser.Parser.subscribe(Parser.java:62)
at xml.parser.Parser.main(Parser.java:34)
但是当我创建普通订阅者时,我的代码运行良好并且没有错误。为什么我收到此错误?还有一个问题-我如何退订该主题?
我的订阅者代码是:
package xml.parser;
import org.w3c.dom.*;
import javax.xml.xpath.*;
import javax.xml.namespace.NamespaceContext;
import javax.xml.parsers.*;
import java.io.IOException;
import org.xml.sax.SAXException;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.util.Properties;
public class Parser {
public static final String QPID_ICF = "org.wso2.andes.jndi.PropertiesFileInitialContextFactory";
private static final String CF_NAME_PREFIX = "connectionfactory.";
private static final String CF_NAME = "qpidConnectionfactory";
String userName = "admin";
String password = "admin";
private static String CARBON_CLIENT_ID = "carbon";
private static String CARBON_VIRTUAL_HOST_NAME = "carbon";
private static String CARBON_DEFAULT_HOSTNAME = "localhost";
private static String CARBON_BROKER_PORT = "5673";
String topicName = "myTopic";
public static void main(String[] args) throws NamingException,
JMSException, XPathExpressionException,
ParserConfigurationException, SAXException, IOException {
Parser queueReceiver = new Parser();
String message = queueReceiver.subscribe();
System.out.println("Got message from Queue ==> " + message);
}
public String subscribe() throws NamingException, JMSException {
String messageContent = "";
Properties properties = new Properties();
properties.put(Context.INITIAL_CONTEXT_FACTORY, QPID_ICF);
properties.put(CF_NAME_PREFIX + CF_NAME,
getTCPConnectionURL(userName, password));
properties.put("topic." + topicName, topicName);
System.out.println("getTCPConnectionURL(userName,password) = "
+ getTCPConnectionURL(userName, password));
InitialContext ctx = new InitialContext(properties);
// Lookup connection factory
TopicConnectionFactory connFactory = (TopicConnectionFactory) ctx
.lookup(CF_NAME);
TopicConnection topicConnection = connFactory.createTopicConnection();
topicConnection.start();
TopicSession topicSession = topicConnection.createTopicSession(false,
QueueSession.AUTO_ACKNOWLEDGE);
// Send message
// Topic topic = topicSession.createTopic(topicName);
Topic topic = (Topic) ctx.lookup(topicName);
javax.jms.TopicSubscriber topicSubscriber = topicSession
.createDurableSubscriber(topic,"topicQueue");
Message message = topicSubscriber.receive();
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
System.out.println("textMessage.getText() = "
+ textMessage.getText());
messageContent = textMessage.getText();
}
topicSubscriber.close();
topicSession.close();
topicConnection.stop();
topicConnection.close();
return messageContent;
}
public String getTCPConnectionURL(String username, String password) {
return new StringBuffer().append("amqp://").append(username)
.append(":").append(password).append("@")
.append(CARBON_CLIENT_ID).append("/")
.append(CARBON_VIRTUAL_HOST_NAME).append("?brokerlist='tcp://")
.append(CARBON_DEFAULT_HOSTNAME).append(":")
.append(CARBON_BROKER_PORT).append("'").toString();
}
}