我在进行JMS测试用例时发现处理是顺序的。当我向一个使用JMS发送消息的Servlet发送200个请求,并且接收器(messageListner)依次接收请求时,如何接收并发请求?我们有哪些参数需要设置?我阅读了JMS教程和API,发现在同一会话中消息是按顺序传递的,即使我为每个发送请求创建一个新会话并在接收端使用10个会话仍然是顺序处理。
public class ProducerServlet extends javax.servlet.http.HttpServlet implements
javax.servlet.Servlet {
// Defines the JNDI context factory.
public final static String JNDI_FACTORY = "weblogic.jndi.WLInitialContextFactory";
// Defines the JMS context factory.
public final static String JMS_FACTORY = "jms/TestConnectionFactory";
// Defines the queue.
public final static String QUEUE = "jms/TestJMSQueue";
public final static String TOPIC = "jms/TestTopic";
TestJMSListener jms = new TestJMSListener();
ConnectionFactory connectionFactory = null;
Queue dest1 = null;
Topic dest =null;
Connection connection = null;
MessageProducer producer = null;
protected void doGet(HttpServletRequest request,
HttpServletResponse response) throws ServletException, IOException {
try {
connection = connectionFactory.createConnection();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
producer = session.createProducer(dest1);
TextMessage message = session.createTextMessage();
message.setText("This is message from JMSSECOND DEMO "
+ request.getParameter("Num"));
System.out.println("Sending message: " + message.getText());
producer.send(message);
producer.send(session.createMessage());
} catch (Exception e) {
System.out.println("Exception occurred: " + e.toString());
}
}
@Override
public void init(ServletConfig arg0) throws ServletException {
Context jndiContext = null;
try {
Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
env.put(Context.PROVIDER_URL, "http://localhost:7001");
jndiContext = new InitialContext(env);
} catch (NamingException e) {
System.out.println("Could not create JNDI API context: "
+ e.toString());
}
try {
connectionFactory = (ConnectionFactory) jndiContext
.lookup(JMS_FACTORY);
dest1 = (Queue) jndiContext.lookup(QUEUE);
} catch (Exception e) {
System.out.println("JNDI API lookup failed: " + e.toString());
e.printStackTrace();
}
}
}
一个 Listner 的实现,在接收到消息后,我会睡眠(进行一些操作一秒钟)。
public class TestJMSListener implements MessageListener {
// Defines the JNDI context factory.
public final static String JNDI_FACTORY = "weblogic.jndi.WLInitialContextFactory";
// Defines the JMS context factory.
public final static String JMS_FACTORY = "jms/TestConnectionFactory";
// Defines the queue.
public final static String QUEUE = "jms/TestJMSQueue";
public final static String TOPIC = "jms/TestTopic";
public TestJMSListener() {
System.out.println("********* Consumer check **********");
Context jndiContext = null;
ConnectionFactory connectionFactory = null;
Connection connection[] = null;
Session session[] = null;
Queue dest1 = null;
Topic dest = null;
MessageConsumer consumer[] = null;
// TextMessage message = null;
try {
Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
env.put(Context.PROVIDER_URL, "http://localhost:7001");
jndiContext = new InitialContext(env);
} catch (NamingException e) {
System.out.println("Could not create JNDI API context: "
+ e.toString());
System.exit(1);
}
try {
connectionFactory = (ConnectionFactory) jndiContext
.lookup(JMS_FACTORY);
dest1 = (Queue) jndiContext.lookup(QUEUE);
} catch (Exception e) {
System.out.println("JNDI API lookup failed: " + e.toString());
System.exit(1);
}
connection = new Connection[10];
session = new Session[10];
consumer = new MessageConsumer[10];
for (int i = 0; i < 10; i++) {
try {
connection[i] = connectionFactory.createConnection();
session[i] = connection[i].createSession(false,
Session.AUTO_ACKNOWLEDGE);
consumer[i] = session[i].createConsumer(dest);
consumer[i].setMessageListener(this);
connection[i].start();
} catch (JMSException e) {
System.out.println("Exception occurred: " + e.toString());
}
}
}
@Override
public void onMessage(Message m) {
if (m instanceof TextMessage) {
TextMessage message = (TextMessage) m;
try {
System.out.println("Reading message from Listener: "
+ new Date() + message.getText());
Thread.sleep(1000);
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
我使用的是Weblogic 11g,使用默认的ConnectionFactory和Queue配置。当我使用Topic时,它实际上只能每秒传递一条消息(即第一条消息完成后),而对于Queue,它每秒钟传递2到3条消息。如何使我的监听器支持并发处理。
最终解决方案
增加更多的监听器对象,而不是在监听器中使用多个会话/消费者,就可以解决这个问题。以下是修改后的代码。
public class ProducerServlet extends javax.servlet.http.HttpServlet implements
javax.servlet.Servlet {
// Defines the JNDI context factory.
public final static String JNDI_FACTORY = "weblogic.jndi.WLInitialContextFactory";
// Defines the JMS context factory.
public final static String JMS_FACTORY = "jms/TestConnectionFactory";
// Defines the queue.
public final static String QUEUE = "jms/TestJMSQueue";
public final static String TOPIC = "jms/TestTopic";
TestJMSListener listeners[] = new TestJMSListener[20];
ConnectionFactory connectionFactory = null;
Queue dest1 = null;
Topic dest =null;
Connection connection = null;
MessageProducer producer = null;
protected void doGet(HttpServletRequest request,
HttpServletResponse response) throws ServletException, IOException {
try {
connection = connectionFactory.createConnection();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
producer = session.createProducer(dest1);
TextMessage message = session.createTextMessage();
message.setText("This is message from JMSSECOND DEMO "
+ request.getParameter("Num"));
System.out.println("Sending message: " + message.getText());
producer.send(message);
producer.send(session.createMessage());
} catch (Exception e) {
System.out.println("Exception occurred: " + e.toString());
}
}
@Override
public void init(ServletConfig arg0) throws ServletException {
Context jndiContext = null;
try {
Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
env.put(Context.PROVIDER_URL, "http://localhost:7001");
jndiContext = new InitialContext(env);
} catch (NamingException e) {
System.out.println("Could not create JNDI API context: "
+ e.toString());
}
try {
connectionFactory = (ConnectionFactory) jndiContext
.lookup(JMS_FACTORY);
dest1 = (Queue) jndiContext.lookup(QUEUE);
for(int i=0;i<listeners.length;i++ ){
listeners[i]=new TestJMSListener(Integer.toString(i+1));
}
} catch (Exception e) {
System.out.println("JNDI API lookup failed: " + e.toString());
e.printStackTrace();
}
}
}
public class TestJMSListener implements MessageListener {
// Defines the JNDI context factory.
public final static String JNDI_FACTORY = "weblogic.jndi.WLInitialContextFactory";
// Defines the JMS context factory.
public final static String JMS_FACTORY = "jms/TestConnectionFactory";
// Defines the queue.
public final static String QUEUE = "jms/TestJMSQueue";
public final static String TOPIC = "jms/TestTopic";
public String listnerNum = "";
public TestJMSListener(String listerNo) {
super();
System.out.println("********* Consumer check **********");
listnerNum = listerNo;
Context jndiContext = null;
ConnectionFactory connectionFactory = null;
Connection connection = null;
Session session = null;
Queue dest1 = null;
Topic dest = null;
MessageConsumer consumer = null;
// TextMessage message = null;
try {
Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
env.put(Context.PROVIDER_URL, "http://localhost:7001");
jndiContext = new InitialContext(env);
} catch (NamingException e) {
System.out.println("Could not create JNDI API context: "
+ e.toString());
System.exit(1);
}
try {
connectionFactory = (ConnectionFactory) jndiContext
.lookup(JMS_FACTORY);
dest1 = (Queue) jndiContext.lookup(QUEUE);
} catch (Exception e) {
System.out.println("JNDI API lookup failed: " + e.toString());
System.exit(1);
}
try{
connection = connectionFactory.createConnection();
session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
consumer = session.createConsumer(dest1);
consumer.setMessageListener(this);
connection.start();
} catch (JMSException e) {
System.out.println("Exception occurred: " + e.toString());
}
}
@Override
public void onMessage(Message m) {
if (m instanceof TextMessage) {
TextMessage message = (TextMessage) m;
try {
System.out.println("Reading message from Listener: "+listnerNum+ " : "
+ new Date() + message.getText());
Thread.sleep(1000);
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
MessageConsumer#receive
手动触发10个线程来消费消息,那么消息消费会如何表现? - ewernli