使用 ActiveMQ 实现 Jms 到 Jms 的桥接

在 Windows XP 上安装 JDK 非常简单，只需下载 MSI 安装程序，然后将其安装到“Program Files”默认位置或直接安装到“C:\”。安装 JDK 后，建议配置系统变量

1. 创建系统环境变量“JAVA_HOME”，并将其指向 JDK 的基本目录（即 C:\Program Files\Java\jdk-1.5.0_17 或 C:\jdk-1.5.0_17）。

配置系统变量后，打开命令提示符并输入“java -version”。输出将显示系统中安装的 JDK 版本。这应该有助于验证 JDK 安装的成功。验证后，关闭命令提示符。

安装 OpenJMS 和 ActiveMQ

安装 OpenJMS 也非常简单，从 Open JMS 网站（http://openjms.sourceforge.net/downloads.html）下载二进制可执行归档文件。然后将 openjms-0.7.7-beta-1.zip 归档文件解压到“C:\”。这将把归档文件解压到“C:\ ”。“C:\ openjms-0.7.7-beta-1”将是 OpenJMS 的基本目录。

安装 ActiveMQ 也非常简单，从 Apache ActiveMQ 网站（https://activemq.apache.ac.cn/activemq-510-release.html）下载二进制可执行归档文件。然后将 apache-activemq-5.1.0-bin.zip 归档文件解压到“C:\”。这将把归档文件解压到“C:\ ”。“C:\ apache-activemq-5.1.0”将是 ActiveMQ 的基本目录。

创建一个简单的 Java 类用于发布消息

import java.util.Hashtable;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class SimpleMessageSender {

	private Context jndiContext;
	
	private QueueConnectionFactory factory;
	
	private QueueConnection queueConnection;
	
	private static final String URL = "tcp://:3035";
	
	private static final String CONNECTION_FACTORY = "ConnectionFactory";
	
	private static final String QUEUE_NAME = "jmstojmsBridgeQueue";
	
	/**
	 * Constructor initialize queue connection factory and connection.
	 * @throws NamingException 
	 * @throws JMSException 
	 */
	public SimpleMessageSender() throws NamingException, JMSException
	{
		Hashtable environment = new Hashtable();
		environment.put(Context.INITIAL_CONTEXT_FACTORY, org.exolab.jms.jndi.InitialContextFactory.class.getName());
		environment.put(Context.PROVIDER_URL, URL);
		jndiContext = new InitialContext(environment);
		factory = (QueueConnectionFactory)jndiContext.lookup(CONNECTION_FACTORY);
		queueConnection = factory.createQueueConnection();
	}
	/**
	 * @param message
	 */
	public void sendMessage(String message){
		 
		try {
		
			QueueSession queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
			Queue queue = queueSession.createQueue(QUEUE_NAME);
			QueueSender sender = queueSession.createSender(queue);
			TextMessage txtMessage = queueSession.createTextMessage("OPEN_JMS :"+message);
			sender.send(txtMessage);
			System.out.println("Message has been send successfully");
			close();
		} catch (JMSException e) {
			e.printStackTrace();
		} catch (NamingException e) {
			
			e.printStackTrace();
		}
	}
	private void close() throws JMSException, NamingException
	{
	   if(queueConnection!=null){
		   queueConnection.close();
	   }
	   if(jndiContext!=null){
		   jndiContext.close();
	   }
	   factory=null;
	}
	
	
	public static void main(String args[]) throws Exception{
		if (args.length<1){
			System.out.println("Usage : java SimpleMessageSender ");
			return;
		}
		SimpleMessageSender simpleMsgSender = new SimpleMessageSender();
		simpleMsgSender.sendMessage(args[0]);
		
	}
}

		

创建一个简单的 Java 类用于消费消息

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueReceiver;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class SimpleMessageListener  implements MessageListener{
   
	private static final String QUEUE_NAME = "jmstojmsBridgeQueue";
	
	public void onMessage(Message message) {
		
		if (message!=null && message instanceof TextMessage){
			TextMessage txtMessage = (TextMessage)message;
			try {
				System.out.println("Message :: "+txtMessage.getText());
			} catch (JMSException e) {
				
				e.printStackTrace();
			}
		}
	}

 public SimpleMessageListener() throws JMSException {
	 
	 ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
	 QueueConnection queueConnection = factory.createQueueConnection();
	 queueConnection.start();
	 QueueSession queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
	 Queue queue  = queueSession.createQueue(QUEUE_NAME);
	 QueueReceiver receiver = queueSession.createReceiver(queue);
	 receiver.setMessageListener(this); 
	
 }
 public static void main(String args[])throws Exception
 {
	 new SimpleMessageListener();
	 System.out.println("Waiting for Message");
 }
}

在编译代码之前，您必须将所需的 jar 文件设置到类路径中

c:\jms>set CLASSPATH=.;<OPEN_JMS_INSTALL_DIR>\lib\jms-1.1.jar; <OPEN_JMS_INSTALL_DIR>\lib\openjms-0.7.7-beta-1.jar;%CLASSPATH%

c:\jms>javac SimpleMessageSender.java

D:\jms>set CLASSPATH=<ACTIVEMQ_INSTALL_DIR>\activemq-all-5.1.0.jar;%CLASSPATH%

添加 JMS 桥接连接器

打开 <ACTIVEMQ_INSTALLED_DIR>/conf/activmq.xml 文件。

在 </transportConnectors> 元素之后添加以下代码。

<jmsBridgeConnectors>
        <jmsQueueConnector name="OpenJMSBridge-Inbound"
           jndiOutboundTemplate="#remoteJndi"
           outboundQueueConnectionFactoryName="ConnectionFactory"
           localQueueConnectionFactory="#localFactory">
          <inboundQueueBridges>
            <inboundQueueBridge inboundQueueName="jmstojmsBridgeQueue"/>
          </inboundQueueBridges>
          </jmsQueueConnector>
</jmsBridgeConnectors>

Add the following code after the  </broker> element

 <bean id="remoteJndi" class="org.springframework.jndi.JndiTemplate">
      <property name="environment">
        <props>
          <prop     key="java.naming.factory.initial">org.exolab.jms.jndi.InitialContextFactory
          <prop key="java.naming.provider.url">tcp://localhost:3035
        </props>
      </property>
    </bean>
  <bean id="localFactory"
    class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="tcp://:61616" />
  </bean>

打开 <OPENJMS_INSTALLED_DIR>/config/Openjms.xml 文件。

注释掉以下代码。

<!-- Connector scheme="rmi">
      <ConnectionFactories>
        <QueueConnectionFactory name="JmsQueueConnectionFactory" />
        <TopicConnectionFactory name="JmsTopicConnectionFactory" />
        <ConnectionFactory name="RMIConnectionFactory"/>
      </ConnectionFactories>
    </Connector -->

注意：如果您不想注释掉上述代码，则应修改 ActiveMQ 中的 JMX 监听端口，因为默认情况下 activemq 监听 1099 端口。openjms rmi 连接器也监听相同的端口号。

运行示例代码的步骤

步骤 1：将 Open-JMS jar 文件复制到 <ActiveMQ-INSTALLED_DIR>/lib 目录。

步骤 2：使用以下批处理文件启动 Open-JMS。

<openjms_install_dir>\bin>start startup.bat

步骤 3：使用以下批处理文件启动 ActiveMQ-JMS。

<activemq_install_dir>\bin\start activemq.bat

步骤 4：使用以下命令运行 SimpleMessageSender 类

java SimpleMessageSender “This is a Sample Message”

SimpleMessageSender 在控制台上的输出

步骤 5

打开另一个 cmd 窗口，设置类路径和 Java 路径，然后运行 SimpleMessageListener 类。

java SimpleMessageListener

SimpleMessageListener 在控制台上的输出：

以上是您需要编写的所有 Java 代码来发布和消费 Java 消息。让我们检查一下 activemq.xml jmsBridgeConnectors 元素

jmsBridgeConnectors：是 jms 桥接连接器的根元素。

jmsQueueConnector: 桥接到其他 JMS 队列提供商。

jndiOutboundTemplate: 如果未设置 localTopicConnection 或 localTopicConnectionFactory，则用于定位 ActiveMQ Connection 的 Connection Factory。

outboundQueueConnectionFactoryName: 用于初始化外部 JMS Connection（如果未设置 localQueueConnection）。

localQueueConnectionFactory: 用于初始化 ActiveMQ JMS Connection（如果未设置 localQueueConnection）。

inboundQueueBridge: 创建一个入站队列桥接。

inboundQueueName: 要从中接收的外部队列名称。有关这些元素的更多信息，请访问以下链接： ActiveMQ。

历史

版本 1.0