JMS订阅模式消息

    发布/订阅模型,消息会发送到一个名为主题(Topic)的虚拟通道中,消息生产者成为发布者(Publisher),消息的消费者成为订阅者(Subscriber);与点对点模型的最大不同,就是发布到主题的消息,能够被多个订阅者接收,类似于广播.发布/订阅模型的消息传输机制是一个基于推送(push)的方式,消息将有JMS Provider主动的向消息消费者广播,消费者客户端无需请求或者轮询,只需要保持Connection的活跃性即可.

    不过在发布/订阅消息传送模型的内部,有多种不同类型的订阅者(比如,受托管订阅者,耐久订阅者,临时订阅者,动态订阅者);临时订阅者(TemporarySubscriber)为只有它主动侦听主题时才能收到消息,JMS Provider不会为临时订阅者持久存储"离线时"的任何消息的副本;持久订阅者将接收到发布的每条消息的一个副本,即使发布消息时,订阅者处于"离线"状态.

 

###contextFactory
java.naming.factory.initial = org.apache.activemq.jndi.ActiveMQInitialContextFactory
###brokerUrl,any protocol
java.naming.provider.url = tcp://localhost:61616
##username
##java.naming.security.principal=
##password
##java.naming.security.credentials=
##connectionFactory,for building sessions
connectionFactoryNames = QueueCF,TopicCF
##topic.<topicName> = <physicalName-of-topic>
##your application should use <topicName>,such as:
## context.lookup("topic1");
##It can be more than once
topic.topic1 = jms.topic1
##queue.<topicName> = <physicalName-of-queue>
queue.queue1 = jms.queue1

 

//Topic发布者
package com.test.jms.simple.topic;

import javax.jms.DeliveryMode;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
import javax.naming.Context;
import javax.naming.InitialContext;

public class SimplePublisher {

	private TopicPublisher producer;
	private TopicSession session;
	private TopicConnection connection;
	private boolean isOpen = true;
	
	public SimplePublisher() throws Exception{
		Context context = new InitialContext();
		TopicConnectionFactory connectionFactory = (TopicConnectionFactory)context.lookup("TopicCF");
		connection = connectionFactory.createTopicConnection();
		connection.setClientID("OK111");
		session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
		Topic topic = (Topic)context.lookup("topic1");
		producer = session.createPublisher(topic);//non durable
		producer.setDeliveryMode(DeliveryMode.PERSISTENT);
		connection.start();
		
	}
	
	
	public boolean send(String text) {
		if(!isOpen){
			throw new RuntimeException("session has been closed!");
		}
		try{
			Message message = session.createTextMessage(text);
			producer.send(message);
			return true;
		}catch(Exception e){
			return false;
		}
	}
	
	public synchronized void close(){
		try{
			if(isOpen){
				isOpen = false;
			}
			session.close();
			connection.close();
		}catch (Exception e) {
			//
		}
	}
	
}

 

//Topic订阅者
package com.test.jms.simple.topic;

import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
import javax.naming.Context;
import javax.naming.InitialContext;

import com.test.jms.object.TopicMessageListener;

public class SimpleSubscriber {

	private TopicConnection connection;
	private TopicSession session;
	private TopicSubscriber consumer;
	
	private boolean isStarted;
	
	public SimpleSubscriber(String clientId) throws Exception{
		Context context = new InitialContext();
		TopicConnectionFactory connectionFactory = (TopicConnectionFactory)context.lookup("TopicCF");
		connection = connectionFactory.createTopicConnection();
		connection.setClientID(clientId);
		session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
		Topic topic = (Topic)context.lookup("topic1");
		consumer = session.createDurableSubscriber(topic, "Test-subscriber");
		consumer.setMessageListener(new TopicMessageListener());
	}
	
	
	public synchronized boolean start(){
		if(isStarted){
			return true;
		}
		try{
			connection.start();
			isStarted = true;
			return true;
		}catch(Exception e){
			return false;
		}
	}
	
	public synchronized void close(){
		isStarted = false;
		try{
			session.close();
			connection.close();
		}catch(Exception e){
			//
		}
	}
}

 

//测试类
package com.test.jms.simple.topic;




public class SimpleTestMain {

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception{
		SimpleSubscriber consumer = new SimpleSubscriber("TestClientId");
		consumer.start();
		
		SimplePublisher productor = new SimplePublisher();
		for(int i=0; i<10; i++){
			productor.send("message content:" + i);
		}
		productor.close();
		//consumer.close();
	}
	

}

 

    在session.createSubscriber(Topic topic)方法中,将会创建一个"非耐久性"主题,即只有subscriber侦听时才会收到消息,当subscriber离线时,它将错过消息.session.createDurableSubscriber(Topic topic,String name)用来创建一个耐久性订阅者,这种订阅者不会错过离线时的消息,JMS Provider将会为它保留所有的消息副本(必须符合相应的消息选择器).其中参数"name"用来表示此订阅者的名字,name的值可以任意,也不允许重复.

    其中耐久性订阅者,必须对connection设定ClientId且此ID全局不能重复,否则将会抛出:javax.jms.JMSException: You cannot create a durable subscriber without specifying a unique clientID on a Connection.

    一个session中只能创建一个耐久性订阅者,否则将抛出异常:javax.jms.JMSException: Durable consumer is in use for client;不过一个connection下可以有多个耐久性订阅者.

    如果一个connection下有多个耐久订阅者时,此时订阅者的name不能重复,否则抛出: javax.jms.JMSException: Durable consumer is in use for client: TestClientId and subscriptionName: ..

    session.unsubscribe(String name)方法为取消订阅,取消当前connection下指定name的订阅者.此后JMS Provider将不会为其保存消息副本.如果你确定一个耐久订阅者不会再次激活时,你需要"取消订阅",否则JMS Provider将会一直为它保存消息副本,而且极有可能带来存储上的风险,如果磁盘或者内存消耗完毕,将会导致JMS Provider故障.

 

    Topic中消息副本的存储模式(数据库描述):

 

++++++++++++Consumers table+++++++
++id      |       Name                      |       destinationId         |     created
1                  clientID1::name1                  testTopoc               122222222
2                  clientID1::name2                  testTopoc               122323232
//其中Name + destinationId为唯一索引.


++++++++++++Messages_handles+++++++
++id      |    messagId    |     destinationId    |    consumerId  |   delivered

1                 10010                 testTopic                      1                0
2                 10010                 testTopic                      2                0
//消息的实体将会保存在其他表中,通过messageId与其关联.
//此表中messageId + destinationId + consumerId为唯一索引.

  

    通过这个存储模式,我们能够理解出JMS Provider是如何创建消息副本的:每创建一个耐久订阅者都将会在Consumer表中新增一条记录,当"取消订阅"之后,相应的consumer记录也会被删除;当一个Topic中新的消息生成时,将会检索consumer表中此destinationId下的所有consumer,然后为每个conusmer生成消息副本--在Message_handles表中插入一条数据;如果某个consumer消费了一条消息,将会在message_handles表中删除消息副本记录.（某些JMS 实现，可能是记录每个订阅者已经消费的最后一个消息的ID，而不是消息的副本）

 

    其中createDurableSubscriber(Topic topic,String name,String selector,boolean noLocal)方法中还有一个重要的参数--noLocal,此参数主要用来控制此订阅者是否接受本地消息,所谓本地消息就是当前Connection下其他publisher发送的消息(对于JMS Provider而言,就是ClientID标识),如果noLocal = true,那么意味着将只能收到其他Client发布的消息.