基于spring和weblogic的jms

clh899

　　环境



　　1.weblogic 9.2.3
　　已经创建JMS Connection Factory：jms/connectionFactory
　　JMS Queue:jms/TestQueue
　　JMS Topic:jms/TestTopic
　　实现一



　　通过spring的JmsTemplate发送消息到Queue,然后通过messageListener实现消息的异步接收.
　　完整的springl配置文件applicationContext-jms.xml如下:

<beans>
<bean id="jmsConnectionFactory" class="org.springframework.jndi.JndiObjectFactoryBean">
<property name="jndiName">
<value>jms/connectionFactory</value>
</property>
<property name="jndiTemplate">
<ref bean="jmsJndiTemplate"></ref>
</property>
</bean>
<!-- define jms queue jndi -->
<bean id="jmsDestination" class="org.springframework.jndi.JndiObjectFactoryBean">
<property name="jndiName">
<value>jms/TestQueue</value>
</property>
<property name="jndiTemplate">
<ref bean="jmsJndiTemplate"></ref>
</property>
</bean>
<!-- define jms queue url -->
<bean id="jmsJndiTemplate" class="org.springframework.jndi.JndiTemplate">
<property name="environment">
<props>
<!-- com.sun.enterprise.naming.SerialInitContextFactory -->
<prop key="java.naming.factory.initial">weblogic.jndi.WLInitialContextFactory</prop>
<prop key="java.naming.provider.url">t3://16.158.51.221:7001</prop>
</props>
</property>
</bean>
<!-- JMS Queue Send Template  -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory">
<ref local="jmsConnectionFactory" />
</property>
<property name="defaultDestination">
<ref local="jmsDestination" />
</property>
</bean>
<bean id="jmsProducer" class="org.springframework.samples.jpetstore.jms.JmsProducer">
<constructor-arg ref="jmsTemplate" />
</bean>
<bean id="messageListener" class="org.springframework.samples.jpetstore.jms.JpetstoreJmsListener" />
<bean id="listenerContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<!--property name="concurrentConsumers" value="5" /-->
<property name="connectionFactory" ref="jmsConnectionFactory" />
<property name="destination" ref="jmsDestination" />
<property name="messageListener" ref="messageListener" />
<property name="sessionAcknowledgeModeName" value="AUTO_ACKNOWLEDGE"/>
</bean>
</beans>
　　消息发送JmsProducer代码如下

package org.springframework.samples.jpetstore.jms;
import org.springframework.jms.core.JmsTemplate;
public class JmsProducer {
private JmsTemplate jmsTemplate;
public JmsProducer(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public void sendMessage() {
jmsTemplate.convertAndSend("Hello world!("+System.currentTimeMillis()+")");
}
}
　　消息接收JpetstoreJmsListener代码如下

package org.springframework.samples.jpetstore.jms;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class JpetstoreJmsListener implements MessageListener {
public void onMessage(Message message) {
if (message instanceof TextMessage) {
try {
System.out.println("ok--"+((TextMessage)message).getText());
} catch (JMSException e) {
throw new RuntimeException(e);
}
} else {
throw new IllegalArgumentException(
"Message must be of type TestMessage");
}
}
}
 　　测试代码

public class JmsConsole {
public static void main(String[] args) throws IOException {
ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext-jms.xml");
JmsProducer sender = (JmsProducer) context.getBean("jmsProducer");
sender.sendMessage();
}
}
　　运行JmsConsole，控制台输入如下：

ok--Hello world!(1250560961109)
　　将测试程序修改一下,如下

public static void main(String[] args) throws Exception {
ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext-jms.xml");
JmsProducer sender = (JmsProducer) context.getBean("jmsProducer");
int i = 1;
System.out.println(System.currentTimeMillis());
while(i<10){
System.out.println("line:"+i);
sender.sendMessage(i);
i++;
}
}
　　同时修改一下JpetstoreJmsListener

public void onMessage(Message message) {
if (message instanceof TextMessage) {
try {
System.out.println("ok--"+((TextMessage)message).getText());
} catch (JMSException e) {
throw new RuntimeException(e);
}
} else {
throw new IllegalArgumentException(
"Message must be of type TestMessage");
}
System.out.println(System.currentTimeMillis());
}
　　将spring配置中的

<property name="concurrentConsumers" value="1" />

　　设置为1，运行测试程序,从开始往Queue发送消息开始到消息全部处理完毕结束平均耗时大概有600ms左右。
　　将spring配置中的


<property name="concurrentConsumers" value="9" />

　　
设置为9，运行测试程序,从开始往Queue发送消息开始到消息全部处理完毕结束平均耗时大概也在600ms左右。
　　将循环加到100，设置为并发消费者数为1和为9的时间有10倍左右的差距。
　　因此对于大并发的Queue处理，设置concurrentConsumers是非常有必要的。
　　实现二



　　上面的例子是传输简单类型的消息，下面让我们看一下对象消息如何传输。
　　传输的对象如下

package org.springframework.samples.jpetstore.domain;
import java.io.Serializable;

public class Product implements Serializable {
/* Private Fields */
private String productId;
private String categoryId;
private String name;
private String description;
/* JavaBeans Properties */
public String getProductId() { return productId; }
public void setProductId(String productId) { this.productId = productId.trim(); }
public String getCategoryId() { return categoryId; }
public void setCategoryId(String categoryId) { this.categoryId = categoryId; }
public String getName() { return name; }
public void setName(String name) { this.name = name; }
public String getDescription() { return description; }
public void setDescription(String description) { this.description = description; }
/* Public Methods*/
public String toString() {
return getName();
}
}

 　　spring的配置不变，需要改变的是发送端代码和messageListener代码，如下
　　JmsProducer

package org.springframework.samples.jpetstore.jms;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.support.converter.SimpleMessageConverter;
import org.springframework.samples.jpetstore.domain.Product;
public class JmsProducer {
private JmsTemplate jmsTemplate;
public JmsProducer(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public void sendMessage() {   
jmsTemplate.send(new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
Product p = new Product();
p.setCategoryId("1-2-3");
p.setName("TestP");
p.setProductId("P-1-2-3");
Message message = new SimpleMessageConverter().toMessage(p, session);
message.setLongProperty("startTime", System.currentTimeMillis());
return message;
}
});
}
}
　　JpetstoreJmsListener

package org.springframework.samples.jpetstore.jms;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import org.springframework.jms.support.converter.SimpleMessageConverter;
import org.springframework.samples.jpetstore.domain.Product;
public class JpetstoreJmsListener implements MessageListener {
public void onMessage(Message message) {
if (message instanceof ObjectMessage) {
try {
Product p = (Product)(new SimpleMessageConverter().fromMessage(message));
System.out.println("ok--"+p.getProductId());
} catch (Exception e) {
throw new RuntimeException(e);
}  
} else  {
throw new IllegalArgumentException(
"Message must be of type TestMessage");
}
System.out.println(System.currentTimeMillis());
}
}
　　测试代码

public class JmsConsole {
public static void main(String[] args) throws IOException {
ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext-jms.xml");
JmsProducer sender = (JmsProducer) context.getBean("jmsProducer");
sender.sendMessage();
}
}
　　输入如下信息：

ok--P-1-2-3
1250585109375
 　　附录



　　概念


　　JMS(Java Message Service)是Java程序与企业消息系统进行交互的规范，它提供了一组API用于创建、发送、接收和读取消息。JMS定义了一组通用的消息概念和功能，旨在降低开发者访问企业消息系统的难度，同时最大化应用程序的可移植性。

　　


　　JMS消息模型




1、Pub/Sub模型：



一对多广播消息模式。在Pub/Sub模型中，消息发送方称为Publisher
，消息接收方称为Subscriber
。主题Topic
充当Publisher和Subscriber之间的中介，对某个Topic感兴趣的Subscriber在Topic上注册，Publisher将消息发送到指定的Topic，Topic上的消息将递送给所有注册的Subscriber。

Pub/Sub模型的主要特点：

对Topic上的一个消息，注册的每个Subscriber均得到该消息的一个拷贝。

Topic上的消息是自动递送给Subscriber的，不需要Subscriber去主动获取新消息，这种方式也称为推模式(Push Model)。

Publisher和Subscriber可以在运行时刻动态添加和删除。

　　2、P2P模型：


一对一消息模式。在P2P模型中，消息发送方称为Sender
，消息接收方称为Receiver
。队列Queue
充当Sender和Receiver之间的中介，Sender将消息发送到指定的Queue，Receiver从指定的Queue上获取消息。

P2P模型的主要特点：

对Queue上的每个消息，即使有多个Receiver请求该消息，有且只有一个Receiver接收到该消息。即每个消息只能被一个Receiver消费，消费之后消息就从Queue上删除了。

Receiver需要到Queue上请求消息，而不是自动递给它的，这种方式也称为拉模式(Pull Model)。

Queue上的消息是有顺序的，消息按照它发送到Queue上的顺序被Receiver取走。