JBoss ESB学习笔记10——第九个ESB应用JMS Topic

续上篇介绍了第八个ESB应用，本文介绍第九个ESB应用——JMS Topic。

 

说明：本文及后续文章虽非百分百的原创，但毕竟包含本人的努力和付出，所以希望大家转载时务请注明出处：http://yarafa.iteye.com，谢谢合作。

 

1 概述
本实例主要演示了如何配置ESB服务使其从JMS消息主题中获取消息。本实例的主要操作过程是配置了两个ESB服务从JMS消息主题获取消息，并同时将接收到的消息转发给另一个ESB服务进行处理。

 

2 新建ESB工程
操作过程略。

 

3 ESB配置
3.1 创建消息队列和消息主题

<?xml version="1.0" encoding="UTF-8"?>
<server>
	<mbean code="org.jboss.jms.server.destination.TopicService"		name="jboss.messaging.destination:service=Topic,name=jmstopicTopic" xmbean-dd="xmdesc/Topic-xmbean.xml">
		<depends optional-attribute-name="ServerPeer">
jboss.messaging:service=ServerPeer</depends>
		<depends>jboss.messaging:service=PostOffice</depends>
	</mbean>
	<mbean code="org.jboss.jms.server.destination.QueueService"		name="jboss.esb.quickstart.destination:service=Queue,name=jmstopicQueue1" xmbean-dd="xmdesc/Queue-xmbean.xml">
		<depends optional-attribute-name="ServerPeer">
jboss.messaging:service=ServerPeer</depends>
		<depends>jboss.messaging:service=PostOffice</depends>
	</mbean>
	<mbean code="org.jboss.jms.server.destination.QueueService"		name="jboss.esb.quickstart.destination:service=Queue,name=jmstopicQueue2" xmbean-dd="xmdesc/Queue-xmbean.xml">
		<depends optional-attribute-name="ServerPeer">
jboss.messaging:service=ServerPeer</depends>
		<depends>jboss.messaging:service=PostOffice</depends>
	</mbean>
	<mbean code="org.jboss.jms.server.destination.QueueService"		name="jboss.esb.quickstart.destination:service=Queue,name=jmstopicQueueResponseGw" 	xmbean-dd="xmdesc/Queue-xmbean.xml">
		<depends optional-attribute-name="ServerPeer">
jboss.messaging:service=ServerPeer</depends>
		<depends>jboss.messaging:service=PostOffice</depends>
	</mbean>
	<mbean code="org.jboss.jms.server.destination.QueueService"		name="jboss.esb.quickstart.destination:service=Queue,name=jmstopicQueueResponseEsb" xmbean-dd="xmdesc/Queue-xmbean.xml">
		<depends optional-attribute-name="ServerPeer">
jboss.messaging:service=ServerPeer</depends>
		<depends>jboss.messaging:service=PostOffice</depends>
	</mbean>
</server>

配置说明：这里需要配置一个消息主题用于接收客户端发来的消息。

 

3.2 定义Provider
这里将定义一个JMS Provider，并定义5个消息通道，内容如下：

<jms-provider connection-factory="ConnectionFactory" name="JMS">
	<jms-bus busid="jmstopicTopic">
		<jms-message-filter dest-name="topic/jmstopicTopic"
			dest-type="TOPIC" />
	</jms-bus>
	<jms-bus busid="jmstopicQueue1">
		<jms-message-filter dest-name="queue/jmstopicQueue1"
			dest-type="QUEUE" />
	</jms-bus>
	<jms-bus busid="jmstopicQueue2">
		<jms-message-filter dest-name="queue/jmstopicQueue2"
			dest-type="QUEUE" />
	</jms-bus>
	<jms-bus busid="jmstopicQueueResponseGw">
		<jms-message-filter dest-name="queue/jmstopicQueueResponseGw"
			dest-type="QUEUE" />
	</jms-bus>
	<jms-bus busid="jmstopicQueueResponseEsb">
<jms-message-filter dest-type="QUEUE"
dest-name="queue/jmstopicQueueResponseEsb" />
	</jms-bus>
</jms-provider>

 

3.3 定义Action类
3.3.1 MessageAugmentor

该类是第一个服务和第二个服务的处理类，该类在实例化时，从各自服务配置信息中获取第三个服务的类别和名称，然后创建第三个服务的实例。当接收到消息后，设置消息“addition”属性的值，同时将消息转发到第三个服务进行处理。

/***********************************************************************
 * <p>Project Name: jmstopic</p>
 * <p>File Name: com.thu.afa.esb.jbossesb.action.MessageAugmentor.java</p>
 * <p>Copyright: Copyright (c) 2010</p>
 * <p>Company: <a href="http://afa.thu.com">http://afa.thu.com</a></p>
 ***********************************************************************/
package com.thu.afa.esb.jbossesb.action;

import org.jboss.soa.esb.ConfigurationException;
import org.jboss.soa.esb.Service;
import org.jboss.soa.esb.actions.AbstractActionLifecycle;
import org.jboss.soa.esb.client.ServiceInvoker;
import org.jboss.soa.esb.helpers.ConfigTree;
import org.jboss.soa.esb.message.Message;

/**
 * <p>Class Name: MessageAugmentor</p>
 * <p>Description: </p>
 * @author Afa
 * @date 2010-9-13
 * @version 1.0
 */
public class MessageAugmentor extends AbstractActionLifecycle
{
	private String addition;
    private Service targetService;
    
    public MessageAugmentor(ConfigTree configTree) throws Exception
    {
    	addition = configTree.getRequiredAttribute("addition");
    	String target = configTree.getRequiredAttribute("target");
    	String[] targetTokens = target.split(":");
    	
    	if(targetTokens.length != 2) {
            throw new ConfigurationException("Action not configured properly - 'target' service property must be in format 'category:name'.");
        }
    	
    	targetService = new Service(targetTokens[0], targetTokens[1]);
    }
    
    public Message process(Message message) throws Exception
    {
    	message.getProperties().setProperty("addition", addition);
    	System.out.println("Received message on Topic.  Sending '" + addition + "' to " + targetService);
    	try
		{
    		getInvoker().deliverAsync(message);
		}
		catch (Exception e)
		{
			throw new Exception("Failed to deliver message: " + e.getMessage());
		}
		
		return message;
    }
    
    private ServiceInvoker getInvoker() throws Exception
    {
    	try
		{
			return new ServiceInvoker(targetService);
		}
		catch (Exception e)
		{
			throw new Exception("Failed to create ServiceInvoker: " + e.getMessage());
		}
    }
}

 

3.3.2 TargetServiceAction

第三个服务的处理类，当第三个服务接收到消息后，将消息中“addition”属性打印出来，以验证消息的来源是第一个服务还是第二个服务。

/***********************************************************************
 * <p>Project Name: jmstopic</p>
 * <p>File Name: com.thu.afa.esb.jbossesb.action.TargetServiceAction.java</p>
 * <p>Copyright: Copyright (c) 2010</p>
 * <p>Company: <a href="http://afa.thu.com">http://afa.thu.com</a></p>
 ***********************************************************************/
package com.thu.afa.esb.jbossesb.action;

import org.jboss.soa.esb.actions.AbstractActionLifecycle;
import org.jboss.soa.esb.helpers.ConfigTree;
import org.jboss.soa.esb.message.Message;

/**
 * <p>Class Name: TargetServiceAction</p>
 * <p>Description: </p>
 * @author Afa
 * @date 2010-9-13
 * @version 1.0
 */
public class TargetServiceAction extends AbstractActionLifecycle
{
	protected ConfigTree configTree;
	
	public TargetServiceAction(ConfigTree configTree)
	{
		this.configTree = configTree;
	}
	
	public Message process(Message message) throws Exception
	{
		String addition = (String) message.getProperties().getProperty("addition");
		System.out.println("MyService3 Received message: '" + addition + "'");

        return message;
	}
}

 

3.4 定义第一个Service
第一个Service监听消息主题，在接收到消息后，将消息转发给第三个Service。

<service category="JmsTopic" description="Jms Topic"
name="JmsTopicService1">
	<listeners>
		<jms-listener busidref="jmstopicTopic" is-gateway="true"
			name="JmsGwListener" />
		<jms-listener busidref="jmstopicQueue1"
name="JmsEsbListener" />
	/listeners>
	<actions mep="OneWay">
		<action name="augmentor" process="process"
class="com.thu.afa.esb.jbossesb.action.MessageAugmentor">
		<property name="addition" value="Service1-Received" />
		<property name="target" value="JmsTopic:JmsTopicService3" />
		</action>
	</actions>
</service>

 

3.5定义第二个Service
第二个Service监听消息主题，在接收到消息后，将消息转发给第三个Service。与第一个服务一样监听同一个消息主题的消息。

<service category="JmsTopic" description="Jms Topic"
name="JmsTopicService2">
	<listeners>
		<jms-listener busidref="jmstopicTopic" is-gateway="true"
			name="JmsGwListener" />
		<jms-listener busidref="jmstopicQueue1"
name="JmsEsbListener" />
	/listeners>
	<actions mep="OneWay">
		<action name="augmentor" process="process"
class="com.thu.afa.esb.jbossesb.action.MessageAugmentor">
		<property name="addition" value="Service2-Received" />
		<property name="target" value="JmsTopic:JmsTopicService3" />
		</action>
	</actions>
</service>

 

3.6 定义第三个Service
第三个Service接收并处理由前两个Service转发来的消息。

<service category="JmsTopic" description="Jms Topic"
name="JmsTopicService3">
	<listeners>
		<jms-listener busidref="jmstopicQueueResponseGw"
			is-gateway="true" name="JmsGwListener" />
		<jms-listener busidref="jmstopicQueueResponseEsb"
name="JmsEsbListener" />
	</listeners>
	<actions mep="OneWay">
		<action name="targetAction"
class="com.thu.afa.esb.jbossesb.action.TargetServiceAction"/>
	</actions>
</service>

 

3.6 配置部署文件
部署依赖文件deployment.xml内容如下：

<jbossesb-deployment>
	<depends>jboss.messaging.destination:service=Topic,name=jmstopicTopic
	</depends>
	<depends>jboss.esb.quickstart.destination:service=Queue,name=jmstopicQueue1
	</depends>
	<depends>jboss.esb.quickstart.destination:service=Queue,name=jmstopicQueue2
	</depends>
	<depends>jboss.esb.quickstart.destination:service=Queue,name=jmstopicQueueResponseGw
	</depends>
	<depends>jboss.esb.quickstart.destination:service=Queue,name=jmstopicQueueResponseEsb
	</depends>
</jbossesb-deployment>

 

3.7 部署ESB
将整个工程导出成一个ESB文件，并保存至JBoss ESB Server的部署目录下，启动JBoss ESB Server即可。

 

4 ESB客户端
4.1 新建Java工程
这里略去操作过程以及添加所需要的Jar包，具体操作过程可参考第一个ESB实例说明。

4.2 发送消息的客户端

该客户端仅仅是向消息主题发送消息，以验证ESB服务之间的互操作。

/***********************************************************************
 * <p>Project Name: helloworldclient</p>
 * <p>File Name: com.thu.afa.esb.jbossesb.client.JmsTopicClient.java</p>
 * <p>Copyright: Copyright (c) 2010</p>
 * <p>Company: <a href="http://afa.thu.com">http://afa.thu.com</a></p>
 ***********************************************************************/
package com.thu.afa.esb.jbossesb.client;

import java.util.Properties;

import javax.jms.ObjectMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.naming.Context;
import javax.naming.InitialContext;

/**
 * <p>Class Name: JmsTopicClient</p>
 * <p>Description: </p>
 * @author Afa
 * @date 2010-9-13
 * @version 1.0
 */
public class JmsTopicClient
{
	private TopicConnectionFactory factory;
	private TopicSession session;
	private Topic topic;
	private TopicConnection connection;
	
	public void setupConnection() throws Exception
	{
		Properties properties = new Properties();
		properties.put(Context.INITIAL_CONTEXT_FACTORY, "org.jnp.interfaces.NamingContextFactory");   
		properties.put(Context.URL_PKG_PREFIXES, "org.jboss.naming:org.jnp.interfaces");   
		properties.put(Context.PROVIDER_URL, "jnp://127.0.0.1:1099");
		InitialContext context = new InitialContext(properties);
		
		factory = (TopicConnectionFactory) context.lookup("ConnectionFactory");
		topic = (Topic) context.lookup("topic/jmstopicTopic");
	}
	
	public void stop() throws Exception
	{
		if(connection != null) connection.stop();
		if(session != null) session.close();
		if(connection != null) connection.close();
	}
	
	public void sendAMessage(String text) throws Exception
	{
		connection = factory.createTopicConnection();
		session = connection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE);
		TopicPublisher publisher = session.createPublisher(topic);
		ObjectMessage message = session.createObjectMessage(text);
		publisher.publish(message);
	}

	/**
	 * <p>Title: </p>
	 * <p>Method Name: main</p>
	 * <p>Description: </p>
	 * @author: Afa
	 * @date: 2010-9-13
	 * @param args
	 */
	public static void main(String[] args) throws Exception
	{
		JmsTopicClient client = new JmsTopicClient();
		client.setupConnection();
		client.sendAMessage("Llu, miss you, afa - jms topic");
		client.stop();
	}

}

 

运行客户端程序即可看到以下输出结果：

 
 

 

-----------------------------------------------------
Stay Hungry, Stay Foolish!
http://yarafa.iteye.com
Afa
Mar 26th, 2011
-----------------------------------------------------

 

评论

1 楼 apollo211 2012-11-27  

第二个service中的 <jms-listener busidref="jmstopicQueue1"
name="JmsEsbListener" />
busidref应该设置为"jmstopicQueue2"
否则 先前定义的jmstopicQueue2没有意义
先前一共定义了5个消息通道
jmstopicTopic消息主题被service1和service2 共同监听着 作为gateway
jmstopicQueue jmstopicQueue2分别作为service1和 service2的内部esb
jmstopicQueueResponseGw  jmstopicQueueResponseEsb是service3的外部和内部esb
总共5个