JMS学习(四)-一个简单的聊天应用程序分析-阿里云开发者社区

一，介绍

本文介绍一个简单的聊天应用程序：生产者将消息发送到Topic上，然后由ActiveMQ将该消息Push给订阅了该Topic的消费者。示例程序来自于《JAVA 消息服务--第二版 Mark Richards著》

二，项目开发环境搭建

①使用Eclipse新建一个JAVA工程:jms_activemq

②添加依赖包：activemq-all-5.13.2.jar

③创建JNDI文件: jndi.properties，并将之放到classpath下(即eclipse项目的 src 目录)。文件内容如下：

java.naming.factory.initial = org.apache.activemq.jndi.ActiveMQInitialContextFactory
java.naming.provider.url=tcp://192.168.121.35:61616
java.naming.security.principal=system
java.naming.security.credentials=manager

# use the following property to specify the JNDI name the connection factory
# should appear as. 
#connectionFactoryNames = connectionFactory, queueConnectionFactory, topicConnectionFactry
connectionFactoryNames = TopicCF  
#TopicConnectionFactory conFactory = (TopicConnectionFactory)ctx.lookup(topicFactory)

# register some queues in JNDI using the form
# queue.[jndiName] = [physicalName]
## queue.MyQueue = example.MyQueue

# register some topics in JNDI using the form
# topic.[jndiName] = [physicalName]
## topic.MyTopic = example.MyTopic
topic.topic1=jms.topic1
#Topic chatTopic = (Topic)ctx.lookup(topicName);

注意：jndi.properties文件中的值不要拼写错误了(此外，等号右边的值的末尾，不能带有空格!!!)。否则会抛出：java.lang.ClassNotFoundException

④新建源文件。编写程序。项目截图如下：

⑤配置activemq.xml文件，在broker下添加一个 destinations，并启动ActiveMQ

配置如下：

这里详细说下如下启动ActiveMQ，一种方式是在前台启动，另一个方式是在后台启动。如果在前台终端中启动的话，当关闭终端后程序也就终止了。而在后台启动，关闭终端不影响它的启动。

前台启动： ./bin/activemq start 或者 ./bin/activemq console

后台启动：参考

 nohup ./bin/activemq start &

在后台启动后，如何关闭呢？首先找到相应的进程号： ps -ef | grep activemq，然后 kill pid 即可。

三，JNDI解释

JNDI是一个与具体实现无关的API，不管你用的是ActiveMQ还是WebSphereMQ还是Microsoft Message Queuing，它都可用作目录服务。JMS Client使用目录服务来访问ConnectionFactory和Destination（即主题和队列对象）。

ConnectionFactory及Destination 与JMS中的Producer、Consumer、Session、Connection不同，后者是用JMS API通过工厂模式直接生成的。而前者则需要JNDI查询获得。

在JMS中，JNDI主要用于命名服务来定位受管对象。受管对象就是由系统管理员创建和配置的JMS对象。受管对象包括JMS ConnectionFactory 和 Destination 对象


受管对象和命名服务中的一个名称相互绑定在一起。一个命名服务将名称和分布式对象、文件、设施关联起来，以便它们可以使用简单的名称而不是密码似的网络地址实现在网络上的定位。

如，DNS,它将域名www.cnblogs.com 转换成一个网络地址(ip)，而浏览器使用该ip连接到web服务器上


命名服务将分布式对象、JMS受管对象……和名称相互绑定，并按照和文件系统相似的层次进行组织


JNDI提供了一个隐藏命名服务细节的抽象。使用JNDI，JMS Client 能够浏览一个命名服务，并在不知道命名服务细节的情况下引用受管对象。


通常，JMS服务器会和一个标准的JNDI驱动器(也称为服务提供者)以及类似于LDAP(轻量级目录访问协议)这样的目录服务 结合使用。或者，JMS 服务器提供一个私有的JNDI服务提供者 和 目录服务。

java.naming.factory.initial = org.apache.activemq.jndi.ActiveMQInitialContextFactory
java.naming.provider.url=tcp://192.168.121.35:61616

//other jndi config......

四，建立连接的详细过程

Chat类的构造函数通过连接JMS服务器使用的JNDI命名服务来建立连接

public Chat(String topicFactory, String topicName, String userName)throws Exception {
        InitialContext ctx = new InitialContext();//使用jndi.properties 获得一个JNDI连接

InitialContex就是所有JNDI查找的起始点，相当于文件系统的根目录。它提供了一条到目录服务的网络连接，这个目录服务就发挥访问JMS受管对象的根目录的作用。

在我们的示例中，ActiveMQ就是JMS服务器，JMS受管对象就是在 conf/activemq.xml 文件中配置的destionations 和 ConnectionFactory

通过new一个InitialContex对象，就建立起了一个到ActiveMQ目录服务的连接，然后在ActiveMQ的命名服务中查找TopicConnectionFactory

TopicConnectionFactory conFactory = (TopicConnectionFactory)ctx.lookup(topicFactory);

受管对象TopicConnectionFactory的属性和行为则由系统管理员配置。比如，一个连接工厂可配置为使用特定的协议、安全模式...连接。

有了TopicConnectionFactory之后，就可以创建TopicConnection了。

TopicConnection connection = conFactory.createTopicConnection();

连接的认证则可以在jndi.properties文件中配置

java.naming.security.principal=system
java.naming.security.credentials=manager

TopicConnection表示和消息服务器(ActiveMQ)的一个连接。一个JMS Client 可以选择从同一连接工厂创建多个连接，但是由于创建连接开销很大，一般是从同一连接中创建多个Session对象。

TopicSession pubSession = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
TopicSession subSession = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);

关闭TopicConnection将关闭和该连接有关的所有对象，包括TopicSession，TopicPublisher、TopicSubscriber。

TopicSession是创建Message、TopicPublisher、TopicSubscriber对象的工厂，用它来创建TopicPublisher和TopicSubscriber。TopicPublisher和TopicSubscriber对象都有一定特定的Topic标识，且这两个对象专属于创建它们的TopicSession。

TopicPublisher publisher = pubSession.createPublisher(chatTopic);
TopicSubscriber subscriber = pubSession.createSubscriber(chatTopic, null, true);

TopicPublisher向Topic发送消息，TopicSubscriber从Topic订阅消息。JNDI用于定位一个Topic对象，它是类似于TopicConnectionFactory的一个受管对象。

        Topic chatTopic = (Topic)ctx.lookup(topicName);

Topic对象是消息服务器上的一个实际主题的句柄，该主题称为物理主题(physical topic)。一个物理主题就是多个Client订阅和发布消息的一条电子通道。

这样，发布者和订阅者就知道往哪里发送/接收消息了。

TopicPublisher是将消息异步发送到Topic，TopicPublisher不会阻塞直到所有的订阅者接收到该消息为止，而是只要消息服务器一接收到消息，TopicPublisher就会从publish()方法返回。

它依赖消息服务器将消息传送给该主题的所有订阅者，消息服务器将消息异步推送给TopicSubscriber。

整个完整程序代码：

 1 import java.io.BufferedReader;
 2 import java.io.InputStreamReader;
 3 
 4 import javax.jms.JMSException;
 5 import javax.jms.Message;
 6 import javax.jms.MessageListener;
 7 import javax.jms.Session;
 8 import javax.jms.TextMessage;
 9 import javax.jms.Topic;
10 import javax.jms.TopicConnection;
11 import javax.jms.TopicConnectionFactory;
12 import javax.jms.TopicPublisher;
13 import javax.jms.TopicSession;
14 import javax.jms.TopicSubscriber;
15 import javax.naming.InitialContext;
16 
17 public class Chat implements MessageListener{
18 
19     private TopicSession pubSession;
20     private TopicPublisher publisher;
21     private TopicConnection connection;
22     private String userName;
23     
24     
25     public Chat(String topicFactory, String topicName, String userName)throws Exception {
26         InitialContext ctx = new InitialContext();//使用jndi.properties 获得一个JNDI连接
27         
28         TopicConnectionFactory conFactory = (TopicConnectionFactory)ctx.lookup(topicFactory);
29         TopicConnection connection = conFactory.createTopicConnection();
30         
31         TopicSession pubSession = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
32         TopicSession subSession = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
33         
34         Topic chatTopic = (Topic)ctx.lookup(topicName);
35         
36         TopicPublisher publisher = pubSession.createPublisher(chatTopic);
37         TopicSubscriber subscriber = pubSession.createSubscriber(chatTopic, null, true);
38         
39         subscriber.setMessageListener(this);
40         
41         
42         this.connection = connection;
43         this.pubSession = pubSession;
44         this.publisher = publisher;
45         this.userName = userName;
46         
47         connection.start();
48     }
49     
50     @Override
51     public void onMessage(Message message) {
52         try{
53             TextMessage textMessage = (TextMessage)message;
54             System.out.println(textMessage.getText());
55         }catch(JMSException e){e.printStackTrace();}
56     }
57      
58     public void writeMessage(String text)throws JMSException{
59         TextMessage message = pubSession.createTextMessage();
60         message.setText(userName + ": " + text);
61         publisher.publish(message);
62     }
63     
64     public void close()throws JMSException{
65         connection.close();
66     }
67     
68     public static void main(String[] args) {
69         try{
70             if(args.length != 3)
71                 System.out.println("Factory, Topic or userName missing");
72             Chat chat = new Chat(args[0], args[1], args[2]);
73             
74             BufferedReader commandLine = new java.io.BufferedReader(new InputStreamReader(System.in));
75             
76             while(true){
77                 String s = commandLine.readLine();
78                 if(s.equalsIgnoreCase("exit")){
79                     chat.close();
80                     System.exit(0);
81                 }else
82                     chat.writeMessage(s);
83             }
84         }catch(Exception e){e.printStackTrace();}
85     }
86 }