JMS入门,好东西

hain

浏览: 455546 次
来自: ...

最近访客更多访客>>

kismilan

jackywu2k2

ArlenYe

xuhuanfeng

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

JMS JBoss 应用服务器 SQL

1 基本信息

摘要：本篇为JMS的学习笔记，教你一步一步使用JMS，并提供一个能运行的完整的案例，可以使人达到快速入门的目的。

作者：陈光耀

2 正文

　　JMS(Java Message Service),是Java消息服务，通过JMS，可以在Java对象之间发送消息。JMS消息支持点对点之间的消息发送，也支持主题/订阅方式的消息发送。

　　/** 注：本笔记中的代码针在jboss4.0.5下运行通过 */

　　消息服务由客户和消息代理组成。每位客户都连接到消息服务，客户可以创建消息、发送消息、接收消息、阅读消息。消息服务可以将接收到的消息转发给其他的客户。

　　消息服务的关键特点：客户只需要最少的信息就可以其他客户通信（需知道其他客户的提供的服务、服务所需信息和客户地址）。

　　消息服务使用转发－存储结构以提交异步信息。

　　JMS包含5个元素：

•　提供者：负责管理消息服务的消息代理
• 　客户： java编写的，利用提供者进行通信的应用程序和组件
• 　消息：在客户之间传输的对象
•　管理的对象：传输中使用的jms对象，分两种，目标工厂（Destination Factory）对象和连接对象，用于连接消息服务，处理发送者和接收者之间的传输。
•　本机客户：是在引入jms之前构建的应用程序，它们是采用另外一种消息系统的本机客户API

　　在引入jms之前，客户是用点对点和订阅/发布结构

　　jms的5个元素：

•　管理对象：连接工厂（Connection Factory）对象和会话对象；连接工厂对象用于创建会话对象；会话对象用于创建发送者和接收者
   ConnectionFactory(QueueConnectionFactory,TopicConnectionFactoy,XAQueueConnectionFactory,XATopicConnectionFactoy)
   Connection(QueueConnection,TopicConnection,XAQueueConnection,XATopicConnection)
• 　会话：  Session
•　消息生成者: MessageProducer(QueueSender, TopicPublisher)
•　消息使用者: MessageConsumer(QueueReciever,TopicSubscriber)
•　消息:  Message

　　jms使用步骤：

1 访问连接工厂：

InitialContext ctx=new InitialContext();
TopicConnectionFactory tcf=(TopicConnectionFactory) ctx.lookup("TopicConnectionFactory");
QueueConnectionFactory qcf=(QueueConnectionFactory) ctx.lookup("QueueConnectionFactory");

2 访问目标工厂：创建Queue对象/Topic对象

Topic mt=(Topic) ctx.lookup("topic/testTopic");
Queue mq=(Queue) ctx.lookup("queue/A");

3 创建连接，分两种：队列连接和主题连接

TopicConnection tc=tcf.createTopicConnection();
QueueConnection qc=qcf.createQueueConnection();

4. 开始接收：也可以放在后面进行

tc.start(); //以后需要调用 tc.close(); 关闭连接
qc.start(); //以后需要调用 qc.close(); 关闭连接

5. 创建会话：会话用于创建消息生产者、消息消费者、消息。

QueueSession qSession=qc.createQueueSession(true,0);
TopicSession tSession=tc.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);

参数1：是否使用事务
参数2：消息确认模式

　　当消息发送者向消息提供者（即消息代理）发送消息时，消息发送者等待消息代理的确认，没有回应则抛出异常，消息发送程序负责处理这个错误。

　　注：消息代理确认只是确认收到了消息，而不是确认消息提交给了消息接收者。

　　消息确认模式：JMS使用确认协议以保证消息的发送，使用了3种确认模式

AUTO_ACKNOWLEDGE ：指定消息提供者在每次收到消息时自动发送确认。消息只向目标发送一次，但传输过程中可能因为错误而丢失消息。
CLIENT_ACKNOWLEDGE ：由消息接收者确认收到消息，通过调用消息的acknowledge()方法（会通知消息提供者收到了消息）
DUPS_OK_ACKNOWLEDGE ：指定消息提供者在消息接收者没有确认发送时重新发送消息（这种确认模式不在乎接收者收到重复的消息）。如

　　消息提供者在试图向非持久化的消息接收者发送消息失败时，消息会丢失。在向持久化消息接收者发送消息时，会等待消息接收者确认，未收到确认，则重新发送消息（消息提供者需要设置JMSRedilivered=true,消息接收者需要调用Message对象的getJMSRedelivered()方法确认JMSRedilivered标记是否为true)

CLIENT_ACKNOWLEDGE 确认收到消息代码：

public void onMessage(Message msg){
try{
//Process incoming messages
msg.acknowledge();
}catch(Exception e){
//handle error
}
}

　　消息事务：包含一组消息，要么全部发送，要不全部都不发送给消息提供者。

　　消息提供者缓存消息，如果消息发送者有一个消息发送失败，则调用session.rollback()方法，则消息提供者会放弃前面发送成功的消息；如果全部发送成功，调用session.commit()方法，将消息全部发送给消息接收者。

6 创建消息生产者：

QueueSender qSender=qSession.createSender(mq);
TopicPublisher tPublisher=tSession.createPublisher(mt);
qSender.send(msg);
tPublisher.publish(msg);

7 创建消息使用者：在接收任何消息之前，客户必须注册到JMS提供者表明希望接收消息。注册后，JMS提供者就负责向客户发送消息。

QueueReceiver qr=qSession.createReceiver(mq);
qc.start();
Message msg1=qr.receive();
TopicSubscriber tSubscriber=tSession.createSubscriber(mt);
tc.start();
Message msg2=tSubscriber.receive(1500); //每隔1500毫秒从主题接收一次

8 创建消息监听器（实现MessageListener接口onMessage()方法）：消息使用者通过消息监听器异步接收消息。

QueueListener qListener=new QueueListener(); //QueueListener为自定义类，实现MessageListener接口
qr.setMessageListener(qListener);
TopicListener tListener=new TopicListener(); //TopicListener为自定义类，实现MessageListener接口
tSubscriber.setMessageListener(tListener);

9 消息(Message)：三部分，头（必须的），属性和正文（二者为可选）。

　　消息头读写方法： getXXX(),setXXX()

　　XXX是字段的名字，许多消息字段是由send()和publish()方法自动设置的，其他自动则由客户或者JMS程序设置。

1）由send() or publish()设置的：

JMSDestination
JMSDeliveryMode
JMSExpiration
JMSPriority
JMSMessageID
JMSTimestamp
JMSCorrelationID: 关联消息ID
JMSReplyTo
JMSType
JMSRedelivered

2）由客户设置的：

JMSCorrelationID: 关联消息ID
JMSReplyTo
JMSType

3) 由jms提供者设置的：

JMSRedelivered

## 属性读写方法： getXXX(name)， setXXX(name,value)

　　消息正文：包含消息，JMS支持6种消息格式，称为消息类型

TextMessage：  有文本组成的String对象
MapMessage：  可以是按顺序或随机访问的 key-value对，key为String， value为primitive
BytesMessage：字节信息（如存放图像）
StreamMessage：包含顺序读取值的流
ObjectMessage：可以序列化的java对象
Message：  无消息正文时可以使用

创建TextMessage消息：

TextMessage msg=qSession.createTextMessage();
msg.setText("myMessage");
qSender.send(msg);

读取TextMessage消息：

Message msg=qr.receive();
if(msg instanceof TextMessage){
TextMessage txtMsg=(TextMessage) msg;
System.out.println("Imcoming message: "+txtMsg.getText());
}else{
//handle error
}

10. 消息选择器：消息使用者使用消息选择器（message selector）选择收到的消息。消息选择器使用条件表达式（符合WHERE子句的SQL-92标准）作为选择条件。
创建并使用消息选择器：

String criteria="Customer='1234'";
TopicSubscriber tSubscriber=ts.createSubscriber(myTopic,criteria,false); //只从主题中接收Customer='1234'的消息

### 向队列发送消息

SendToQueue.java:
package test.jms;
import java.util.Properties;
import javax.jms.*;
import javax.naming.*;
public class SendToQueue {
public static void main(String[] args) {
final int msgCount;
if( (args.length<1) || ( args.length>2)) {
System.out.println("Usage: java test.jms.SendToQueue queueName [sendCount]");
System.exit(1);
}
String qName=new String(args[0]);
if(args.length==2) {
msgCount=(new Integer(args[1]).intValue());
}else {
msgCount=1;
}
QueueConnection qc=null;
try {
Properties p = new Properties();
p.setProperty(Context.INITIAL_CONTEXT_FACTORY,
"org.jnp.interfaces.NamingContextFactory");
p.setProperty(Context.PROVIDER_URL, "localhost:1099");
InitialContext jc=new InitialContext(p);
QueueConnectionFactory qcf=(QueueConnectionFactory)jc.lookup("QueueConnectionFactory");
Queue q=(Queue)jc.lookup(qName);
qc=qcf.createQueueConnection();
QueueSession qs=qc.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
QueueSender qSender=qs.createSender(q);
TextMessage msg=qs.createTextMessage();
for(int i=0;i<msgCount;i++) {
msg.setText("Welcome number "+(i+1));
qSender.send(msg);
System.out.println("Send Message To "+qName+" : "+msg.getText()+"\n");
}
qSender.close();
qs.close();
}catch(Exception e) {
e.printStackTrace();
}finally {
if(qc!=null) {
try {
qc.close();
}catch(JMSException e) {}
}
}
}
}

运行： java test.jms.SendToQueue queue/A 10

### 从队列接收消息

ReceiveFromQueue.java:
package test.jms;
import java.io.*;
import java.util.Properties;
import javax.jms.*;
import javax.naming.*;
public class ReceiveFromQueue {
public static void doReceive(String qName) {
Message msg;
TextMessage txtMsg;
QueueConnection qc = null;
try {
Properties p = new Properties();
p.setProperty(Context.INITIAL_CONTEXT_FACTORY,
"org.jnp.interfaces.NamingContextFactory");
p.setProperty(Context.PROVIDER_URL, "localhost:1099");
InitialContext jc = new InitialContext(p);
QueueConnectionFactory qcf = (QueueConnectionFactory) jc
.lookup("QueueConnectionFactory");
Queue q = (Queue) jc.lookup(qName);
qc = qcf.createQueueConnection();
QueueSession qs = qc.createQueueSession(false,
Session.AUTO_ACKNOWLEDGE);
QueueReceiver qr = qs.createReceiver(q);
qc.start();
System.out.println("begin receive messge from " + qName + "...");
msg = qr.receive(1000); // 如果不加间隔参数，会一直等着，知道消息到来。
while (msg != null) {
if (msg instanceof TextMessage) {
txtMsg = (TextMessage) msg;
System.out.println("Receive Msg from " + qName + " : "
+ txtMsg.getText());
}
msg = qr.receive(1000);
}
System.out.println("no message available!");
qr.close();
qs.close();
} catch (Exception e) {
e.printStackTrace();
} finally {
if (qc != null) {
try {
qc.close();
} catch (JMSException e) {
}
}
}
}
public static void doListen(String qName) {
Message msg;
TextMessage txtMsg;
QueueConnection qc = null;
try {
Properties p = new Properties();
p.setProperty(Context.INITIAL_CONTEXT_FACTORY,
"org.jnp.interfaces.NamingContextFactory");
p.setProperty(Context.PROVIDER_URL, "localhost:1099");
InitialContext jc = new InitialContext(p);
QueueConnectionFactory qcf = (QueueConnectionFactory) jc
.lookup("QueueConnectionFactory");
Queue q = (Queue) jc.lookup(qName);
qc = qcf.createQueueConnection();
QueueSession qs = qc.createQueueSession(false,
Session.AUTO_ACKNOWLEDGE);
QueueReceiver qr = qs.createReceiver(q);
qc.start();
System.out.println("begin listen to messge from " + qName + "...");
TextListener tListener = new TextListener();
qr.setMessageListener(tListener);
qc.start();
System.out.println("Enter 'q' and press <return> to exit ");
InputStreamReader isr = new InputStreamReader(System.in);
char response = '\0';
while (!((response == 'q') || (response == 'Q'))) {
try {
response = (char) isr.read();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("End listening!");
qr.close();
qs.close();
} catch (Exception e) {
e.printStackTrace();
} finally {
if (qc != null) {
try {
qc.close();
} catch (JMSException e) {
}
}
}
}
public static void main(String[] args) {
if ((args.length != 1)) {
System.out
.println("Usage: java test.jms.ReceiveFromQueue queueName");
System.exit(1);
}
String qName = new String(args[0]);
doReceive(qName); //通过QueueReceiver.receive()读消息
doListen(qName); // 通过消息监听器读消息
}
}

运行： java test.jms.ReceiveFromQueue queue/A

### 向主题发送消息

PublishToTopic.java:
package test.jms;
import java.util.Properties;
import javax.jms.*;
import javax.naming.*;
public class PublishToTopic {
public static void main(String[] args) {
final int msgCount;
if( (args.length<1) || ( args.length>2)) {
System.out.println("Usage: java test.jms.PublishToTopic topicName [sendCount]");
System.exit(1);
}
String tName=new String(args[0]);
if(args.length==2) {
msgCount=(new Integer(args[1]).intValue());
}else {
msgCount=10;
}
TopicConnection tc=null;
try {
Properties p = new Properties();
p.setProperty(Context.INITIAL_CONTEXT_FACTORY,
"org.jnp.interfaces.NamingContextFactory");
p.setProperty(Context.PROVIDER_URL, "localhost:1099");
InitialContext jc=new InitialContext(p);
TopicConnectionFactory tcf=(TopicConnectionFactory)jc.lookup("TopicConnectionFactory");
Topic t=(Topic)jc.lookup(tName);
tc=tcf.createTopicConnection();
TopicSession ts=tc.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
TopicPublisher tp=ts.createPublisher(t);
TextMessage msg=ts.createTextMessage();
for(int i=0;i<msgCount;i++) {
msg.setText("Welcome number "+(i+1));
tp.publish(msg);
System.out.println("Publish Message To "+tName+" : "+msg.getText()+"\n");
}
tp.close();
ts.close();
}catch(Exception e) {
e.printStackTrace();
}finally {
if(tc!=null) {
try {
tc.close();
}catch(JMSException e) {}
}
}
}
}

运行： java -classpath d:\jboss-4.0.5\client\jnp-client.jar;d:\jboss-4.0.5\client\jbossmq-client.jar;d:\jboss-4.0.5\client\jboss-j2ee.jar;d:\jboss-4.0.5\client\jbossall-client.jar;. test.jms.PublishToTopic topic/testTopic

### 从主题接收消息

SubscribeFromTopic.java:
package test.jms;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Properties;
import javax.jms.*;
import javax.naming.*;
public class SubscribeFromTopic {
public static void doListen(String tName) {
char response='\0';
TopicConnection tc=null;
try {
Properties p = new Properties();
p.setProperty(Context.INITIAL_CONTEXT_FACTORY,
"org.jnp.interfaces.NamingContextFactory");
p.setProperty(Context.PROVIDER_URL, "localhost:1099");
InitialContext jc=new InitialContext(p);
TopicConnectionFactory tcf=(TopicConnectionFactory)jc.lookup("TopicConnectionFactory");
Topic t=(Topic)jc.lookup(tName);
tc=tcf.createTopicConnection();
TopicSession ts=tc.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
TopicSubscriber tSubscriber=ts.createSubscriber(t);
System.out.println("begin listen to messge from "+tName+"...");
TextListener tListener=new TextListener();
tSubscriber.setMessageListener(tListener);
tc.start();
System.out.println("Enter 'q' and press <return> to exit ");
InputStreamReader isr=new InputStreamReader(System.in);
while(!( (response=='q') || (response=='Q' ))) {
try {
response=(char)isr.read();
}catch(IOException e) {
e.printStackTrace();
}
}
System.out.println("End listening!");
tSubscriber.close();
ts.close();
}catch(Exception e) {
e.printStackTrace();
}finally {
if(tc!=null) {
try {
tc.close();
}catch(JMSException e) {}
}
}
}
public static void doSubscribe(String tName) {
Message msg;
TextMessage txtMsg;
TopicConnection tc=null;
try {
Properties p = new Properties();
p.setProperty(Context.INITIAL_CONTEXT_FACTORY,
"org.jnp.interfaces.NamingContextFactory");
p.setProperty(Context.PROVIDER_URL, "localhost:1099");
InitialContext jc=new InitialContext(p);
TopicConnectionFactory tcf=(TopicConnectionFactory)jc.lookup("TopicConnectionFactory");
Topic t=(Topic)jc.lookup(tName);
tc=tcf.createTopicConnection();
TopicSession ts=tc.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
TopicSubscriber tSubscriber=ts.createSubscriber(t);
System.out.println("begin listen to messge from "+tName+"...");
System.out.println("begin receive messge from " + tName + "...");
msg = tSubscriber.receive(1000);
while (msg != null) {
if (msg instanceof TextMessage) {
txtMsg = (TextMessage) msg;
System.out.println("Receive Msg from " + tName + " : "
+ txtMsg.getText());
}
msg = tSubscriber.receive(1000);
}
System.out.println("no message available!");
tSubscriber.close();
ts.close();
}catch(Exception e) {
e.printStackTrace();
}finally {
if(tc!=null) {
try {
tc.close();
}catch(JMSException e) {}
}
}
}
public static void main(String[] args) {
if( (args.length!=1)) {
System.out.println("Usage: java test.jms.SubscribeFromTopic topicName");
System.exit(1);
}
String tName=new String(args[0]);
doListen(tName); //通过消息监听器接收消息
// doSubscribe(tName); //通过receive()来接收消息
}
}

附：TextListener.java:

SubscribeFromTopic.java:
package test.jms;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Properties;
import javax.jms.*;
import javax.naming.*;
public class SubscribeFromTopic {
public static void doListen(String tName) {
char response='\0';
TopicConnection tc=null;
try {
Properties p = new Properties();
p.setProperty(Context.INITIAL_CONTEXT_FACTORY,
"org.jnp.interfaces.NamingContextFactory");
p.setProperty(Context.PROVIDER_URL, "localhost:1099");
InitialContext jc=new InitialContext(p);
TopicConnectionFactory tcf=(TopicConnectionFactory)jc.lookup("TopicConnectionFactory");
Topic t=(Topic)jc.lookup(tName);
tc=tcf.createTopicConnection();
TopicSession ts=tc.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
TopicSubscriber tSubscriber=ts.createSubscriber(t);
System.out.println("begin listen to messge from "+tName+"...");
TextListener tListener=new TextListener();
tSubscriber.setMessageListener(tListener);
tc.start();
System.out.println("Enter 'q' and press <return> to exit ");
InputStreamReader isr=new InputStreamReader(System.in);
while(!( (response=='q') || (response=='Q' ))) {
try {
response=(char)isr.read();
}catch(IOException e) {
e.printStackTrace();
}
}
System.out.println("End listening!");
tSubscriber.close();
ts.close();
}catch(Exception e) {
e.printStackTrace();
}finally {
if(tc!=null) {
try {
tc.close();
}catch(JMSException e) {}
}
}
}
public static void doSubscribe(String tName) {
Message msg;
TextMessage txtMsg;
TopicConnection tc=null;
try {
Properties p = new Properties();
p.setProperty(Context.INITIAL_CONTEXT_FACTORY,
"org.jnp.interfaces.NamingContextFactory");
p.setProperty(Context.PROVIDER_URL, "localhost:1099");
InitialContext jc=new InitialContext(p);
TopicConnectionFactory tcf=(TopicConnectionFactory)jc.lookup("TopicConnectionFactory");
Topic t=(Topic)jc.lookup(tName);
tc=tcf.createTopicConnection();
TopicSession ts=tc.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
TopicSubscriber tSubscriber=ts.createSubscriber(t);
System.out.println("begin listen to messge from "+tName+"...");
System.out.println("begin receive messge from " + tName + "...");
msg = tSubscriber.receive(1000);
while (msg != null) {
if (msg instanceof TextMessage) {
txtMsg = (TextMessage) msg;
System.out.println("Receive Msg from " + tName + " : "
+ txtMsg.getText());
}
msg = tSubscriber.receive(1000);
}
System.out.println("no message available!");
tSubscriber.close();
ts.close();
}catch(Exception e) {
e.printStackTrace();
}finally {
if(tc!=null) {
try {
tc.close();
}catch(JMSException e) {}
}
}
}
public static void main(String[] args) {
if( (args.length!=1)) {
System.out.println("Usage: java test.jms.SubscribeFromTopic topicName");
System.exit(1);
}
String tName=new String(args[0]);
doListen(tName); //通过消息监听器接收消息
// doSubscribe(tName); //通过receive()来接收消息
}
}

分享到：

struts的ActionMessages消息机制 | Java消息服务基础

2007-12-18 10:48
浏览 4769
评论(0)
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论