topicpublisher (jms)

目的地类型JNDI名字连接工厂类型
Topic/Queuejava:/ConnectionFactoryJVM
Topic/Queuejava:/XAConnectionFactoryJVM支持XA事务
Topic/QueueRMIConnectionFactoryRMI
Topic/QueueRMIXAConnectionFactoryRMI支持XA事务
Topic/QueueConnectionFactoryOIL
Topic/QueueXAConnectionFactoryOIL支持XA事务
Topic/QueueUILConnectionFactoryUIL
Topic/QueueUILXAConnectionFactoryUIL支持XA事务
3）JBoss中高级JMS配置
在上边段落主要描述了和JbossMQ一起实行的基本配置工作。在本段来描述JMS其他配置。
(1)JMS持久性管理
JMS持久性管理(PM)负责存储消息，并且将消息标记为持久，如果服务器发生故障时，能保证消息不会丢失，并允许恢复持久性消息。持久性JMS消息可以使用不同的方法来完成。每个方法有自己优点和缺陷：
PM名字优点缺点
File比较稳定速度慢
Rollinglogged速度比较快此方法比较新，有些地方需要完善
JDBC对于稳定性和可量测性比较好必须有JDBC
Logged速度快Logfilesgrowwithoutbound,memorymanagementproblemsduringrecovery
在JBoss中缺省的持久性消息管理是File持久性管理。可以改变它，但必须对于一个JMS
Server有且仅有一个持久性管理配置。所以你在JBoss管理界面的jboss.mq–>
service=PersistenceManager只是看到一个。
持久性管理的配置文件是jbossmq-service.xml。在server\..\deploy下。
为了让大家了解持久性管理的各种方法，我下面来逐个介绍如何配置。
wFile持久性管理
File持久性管理的概念就是为每一个持久性消息建立一个文件。消息持久性方法不是全部都能使用，但它是比较稳定的。
File持久性管理在JBoss发布时，作为一个缺省的持久性管理。如果你打开jbossmq-service.xml文件，你会看到下面的XML：
<mbeancode="org.jboss.mq.pm.file.PersistenceManager"
name="jboss.mq:service=PersistenceManager">
<attributename="DataDirectory">db/jbossmq/file</attribute>
<dependsoptional-attribute-name="MessageCache">jboss.mq:service=MessageCache</depends>
</mbean>
当设置Mbean配置时，File持久性管理允许你指定下面的属性：
DataDircetory是存放持久性消息的路径，会把生成的数据文件放在此目录下。
w设置Rollinglogged持久性管理
Rollinglogged持久性管理是比较新的一种持久性消息管理方法，因为它使用日志文件来持续多个消息，所以当建立一个文件时，不需要许多的I/O。
定义Rollinglogged持久性管理：
<mbeancode="org.jboss.mq.pm.rollinglogged.PersistenceManager"
name="jboss.mq:service=PersistenceManager">
<attributename="DataDirectory">db/jbossmq/file</attribute>
<dependsoptional-attribute-name="MessageCache">jboss.mq:service=MessageCache</depends>
</mbean>
Rollinglogged持久性管理中DataDirctory存放持久性消息的路径，会把生成的数据文件放在此目录下。
w设置JDBC持久性管理
JDBC持久性管理使用数据库表来存储消息。需要一个JBoss配置的数据源来访问数据库。具体内容参考jbossmq-service.xml文件中的内容。
w设置Logged持久性管理
Logged持久性管理是比较早的一个，在JBoss2.4.1以后版本中不在建议使用。现在有其他更好的办法。
4、举例说明
当我们清楚了以后内容后，现在我们来用JBoss实现一个例子来加深对JBoss和JMS的了解。
在上面叙述中，我们知道明确使用JMSprovider有三个基本的事情要做：配置JNDI初始化上下文，连接工厂的名字和使用目的地的名字。
当编写产品的最好的事情是不受provider-specific影响，使代码能在不同的JMSprovider之间容易移植。在此这个例子没有聚焦在开发产品上，而是解释如何使用JbossMQ来工作。
1)初始化上下文
w配置JNDI的一个方法是通过属性文件jndi.properties。在这个文件中使用正确的值，并且把它所在的路径包含到classpath中，它比较容获得正确初始化上下文。
jndi.properties文件的内容如下：
java.naming.factory.initial=org.jnp.inte***ces.NamingContextFactory
java.naming.provider.url=localhost:1099
java.naming.factory.url.pkgs=org.jboss.naming:org.jnp.inte***ces
把该文件放置的路径成为你的classpath的一部分。如果你使用这种方法，在初始化上下文时，代码比较简单:Contextcontext=newIntialContext();1
w在某些情景下，可能需要手工配置JNDI；例如当运行的类文件中环境已经配置了一个初始化上下文，但不是你想用的上下文时，需要手工来配置一个上下文。设置在哈希表中的几个属性值，并且使用此哈希表来实例化一个上下文。定义语法：
Hashtableprops=newHashtable();
props.put(Context.INITIAL_CONTEXT_FACTORY,
"org.jnp.inte***ces.NamingContextFactory");
props.put(Context.PROVIDER_URL,"localhost:1099");
props.put("java.naming.rmi.security.manager","yes");
props.put(Context.URL_PKG_PREFIXES,"org.jboss.naming");
2)查找连接工厂
自有了上下文后，需要查找一个连接工厂。为了查找它，使用一个可用的名字。查找连接工厂的代码如下：
对于一个topic目的地
TopicConnectionFactorytopicFactory=(TopicConnectionFactory)context.lookup(“ConnectionFactory”)
Queue目的地：
QueueConnectionFactoryqueueFactory=(QueueConnectionFactory)context.lookup(“ConnectionFactory”)
3)建立连接和会话
在我们有了连接工厂后，建立一个连接，在此连接中建立一个会话。
对于topic代码如下：
//建立一个连接
topicConnection=topicFactory.createTopicConnection();
//建立一个会话
topicSession=topicConnection.createTopicSession(false,//不需要事务
Session.AUTO_ACKNOLEDGE//自动接收消息的收条。
);
对于queue代码如下：
//建立一个连接
queueConnection=queueFactory.createQueueConnection();
//建立一个会话
queueSession=queueConnection.createQueueSession(false,//不需要事务
Session.AUTO_ACKNOLEDGE//自动接收消息的收条。
);
一个会话建立时，配置是否调用事务
在事务Session中，当事务被提交后，自动接收，如果事务回滚，所有的被消费的消息将会被重新发送。
在非事务Session中，如果没有调用事务处理，消息传递的方式有三种：
Session.AUTO_ACKNOWLEDGE:当客户机调用的receive方法成功返回，或当MessageListenser成功处理了消息，session将会自动接收消息的收条。
Session.CLIENT_ACKNOWLEDGE：客户机通过调用消息的acknowledge方法来接收消息。接收发生在session层。接收到一个被消费的消息时，将自动接收该session已经消费的所有消息。例如：如果消息的消费者消费了10条消息，然后接收15个被传递的消息，则前面的10个消息的收据都会在这15个消息中被接收。
Session.DUPS_ACKNOWLEDGE：指示session缓慢接收消息。
4)查找目的地
现在我们来介绍建立publishes/sends或subscribles/receives消息。
下面的代码列出来查找一个目的地：
对于topic查找一个testTopic目的地
Topictopic=(Topic)context.lookup(“topic/testTopic”);
对于queue查找一个testQueue目的地
Queuequeue=(Queue)context.lookup(“queue/testQueue”);
注意：JbossM的前缀topic/(queue/)通常被放在topic(queue)名字前面。
在JMS中，当客户机扮演每种角色，象对于topic来将的publisher,subscriber或对于queue来将的sender,receiver，都有自己不同类继承和不同函数。
5)建立一个消息制造者MessageProducer(topicpublisher/queuesender)
消息制造者是一个由session创建的对象，主要工作是发送消息到目的地。
对于一个topic，需要通过TopicSession来创建一个TopicPublisher。代码如下：
TopicPublishertopicPublisher=TopicSession.createPublisher(topic);
对于一个queue，需要通过QueueSession来创建一个QueueSender。代码如下：
QueuePublisherqueuePublisher=queueSession.createSender(queue);
6)消息发送
建立一个TestMessage并且publish它，代码：
TextMessagemessage=topicSession.createTestMessage();
message.setText(msg);
topicPublishe.publish(topic,message);
建立一个TestMessage并且send它，代码：
TextMessagemessage=queueSession.createTestMessage();
message.setText(msg);
queueSender.send(queue,message);
7)下面是一个完成的topicpublisher代码，文件名HelloPublisher.java：
importjavax.naming.Context;
importjavax.naming.InitialContext;
importjavax.naming.NamingException;
importjavax.jms.TopicConnectionFactory;
importjavax.jms.TopicConnection;
importjavax.jms.TopicSession;
importjavax.jms.TopicPublisher;
importjavax.jms.Topic;
importjavax.jms.TextMessage;
importjavax.jms.Session;
importjavax.jms.JM***ception;
importjava.util.Hashtable;
publicclassHelloPublisher{
TopicConnectiontopicConnection;
TopicSessiontopicSession;
TopicPublishertopicPublisher;
Topictopic;
publicHelloPublisher(StringfactoryJNDI,StringtopicJNDI)
throwsJM***ception,NamingException{
Hashtableprops=newHashtable();
props.put(Context.INITIAL_CONTEXT_FACTORY,"org.jnp.inte***ces.NamingContextFactory");
props.put(Context.PROVIDER_URL,"localhost:1099");
props.put("java.naming.rmi.security.manager","yes");
props.put(Context.URL_PKG_PREFIXES,"org.jboss.naming");
Contextcontext=newInitialContext(props);
TopicConnectionFactorytopicFactory=
(TopicConnectionFactory)context.lookup(factoryJNDI);
topicConnection=topicFactory.createTopicConnection();
topicSession=topicConnection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);
topic=(Topic)context.lookup(topicJNDI);
topicPublisher=topicSession.createPublisher(topic);
}
publicvoidpublish(Stringmsg)throwsJM***ception{
TextMessagemessage=topicSession.createTextMessage();
message.setText(msg);
topicPublisher.publish(topic,message);
}
publicvoidclose()throwsJM***ception{
topicSession.close();
topicConnection.close();
}
publicstaticvoidmain(String[]args){
try{
HelloPublisherpublisher=newHelloPublisher(
"ConnectionFactory","topic/testTopic");
for(inti=1;i<11;i++){
Stringmsg="HelloWorldno."+i;
System.out.println("Publishingmessage:"+msg);
publisher.publish(msg);
}
publisher.close();
}catch(Exceptionex){
System.err.println(
"AnexceptionoccurredwhiletestingHelloPublisher25:"+ex);
ex.printStackTrace();
}
}
}