ActiveMQ5.0实战3:使用Spring发送,消费topic和queue消息

多谢楼主，终于实现了两个应用之间的消息通讯，这个问题困扰我很久了，一直不得其解，真的是感觉不尽啊!activemq官网上说支持多种语言，不知楼主是否知道c++与java之间如何通过activemq通讯呢

没接触过，菜鸟向前辈学习！

已经通过楼主的贡献代码，调试通过！谢谢！
但我还有一点不明白，楼主的listener包是要干什么用呢？

spring的事件机制

前面你也说到
transportConnector是定義鏈接activeMQ的brokerURL. 你可以為一個activeMQ server指定多個transportconnector, 比如vm, tcp, ssl, stomp, xmpp等.
那我A多定义几个，B是不是也有不同的brokerURL？那他们又是怎么链接同一个activeMq？

楼主能否给个 文档 链接或者下载，感激不尽啊

楼主你好，我看了你的例子，完全可以跑通，我改了下，现有A,B2个web应用系统，都起在一个web服务器（tomcat）下，A应用作为producer，B应用consumer，现在可以看到A应用发的消息，但是在B应用怎么看到已经得到或没有得到这个消息呢？有些不明白怎么处理，我不知道我这样写对不对，请指教，配置如下：
A应用：applicationContext-activemq.xml

<amq:broker useJmx="false" persistent="true">
<amq:persistenceAdapter>
<amq:amqPersistenceAdapter directory="c:/amq"/>
</amq:persistenceAdapter>
<amq:transportConnectors>
<amq:transportConnector uri="tcp://localhost:0" />
</amq:transportConnectors>
</amq:broker>

<!-- 连接外部的activeMQ -->
<amq:connectionFactory id="jmsConnectionFactory" brokerURL="tcp://localhost:61616" />

<!--  ActiveMQ destinations  -->
<!--  使用topic方式-->
<amq:topic name="TOPIC" physicalName="JMS-TEST-TOPIC" />

<!--  Spring JmsTemplate config -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory">
<!--  lets wrap in a pool to avoid creating a connection per send -->
<bean class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="jmsConnectionFactory" />
</bean>
</property>
<!-- custom MessageConverter -->
<property name="messageConverter" ref="defaultMessageConverter" />
</bean>

<!-- converter  -->
<bean id="defaultMessageConverter" class="com.me.jms.DefaultMessageConverter" />

<!-- POJO which send Message uses  Spring JmsTemplate -->
<bean id="topicMessageProducer" class="com.me.jms.TopicMessageProducer">
<property name="template" ref="jmsTemplate" />
<property name="destination" ref="TOPIC" />
</bean>
A应用producer
public class TopicMessageProducer {
   
    private JmsTemplate template;

private Topic destination;

public void setTemplate(JmsTemplate template) {
this.template = template;
}

public void setDestination(Topic destination) {
this.destination = destination;
}

public void send(TestBean message) {
System.out.println("=====> message send! ");
template.convertAndSend(this.destination, message);
}
}
A应用中的Action
public class TestAction extends ActionSupport{

private TestBean tb;
private TopicMessageProducer topicMessageProducer;

public String test(){
System.out.println("===> username:" + tb.getUserName());
System.out.println("===> username:" + tb.getPassWord());
topicMessageProducer.send(tb);

return "success";
}
...geter(),seter();
}

B应用applicationContext-activemq.xml
<amq:connectionFactory id="jmsConnectionFactory"
brokerURL="tcp://localhost:61616" />

<amq:topic name="TOPIC" physicalName="JMS-TEST-TOPIC" />

<!-- converter  -->
<bean id="defaultMessageConverter"
class="com.me.jms.DefaultMessageConverter" />

<!--  Message Driven POJO (MDP) -->
    <!-- consumer1 for topic a -->
    <bean id="topicConsumerA" class="com.me.jms.TopicConsumerA" />

    <!-- consumer2 for topic a -->
    <bean id="topicConsumerB" class="com.me.jms.TopicConsumerB" />
   
    <!-- Message Listener for  -->
<bean id="topicListenerA" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg ref="topicConsumerA" />
<!--  may be other method -->
<property name="defaultListenerMethod" value="receive" />
<!-- custom MessageConverter define -->
<property name="messageConverter" ref="defaultMessageConverter" />
</bean>

<bean id="topicListenerB" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg ref="topicConsumerB" />
<!--  may be other method -->
<property name="defaultListenerMethod" value="receive" />
<!-- custom MessageConverter define -->
<property name="messageConverter" ref="defaultMessageConverter" />
</bean>

<!--  listener container，MDP无需实现接口 -->
<bean id="topicListenerContainerA" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="jmsConnectionFactory" />
<property name="destination" ref="TOPIC" />
<property name="messageListener" ref="topicListenerA" />
</bean>

    <bean id="topicListenerContainerB" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="jmsConnectionFactory" />
<property name="destination" ref="TOPIC" />
<property name="messageListener" ref="topicListenerB" />
</bean>
B应用的consumerA和consumerB都一样
public class TopicConsumerA {
public void receive(TestBean message) {
System.out.println("************************************** Topic A userName: " + message.getUserName());
System.out.println("************************************** Topic A passWord: " + message.getPassWord());
}
}
目前是在tomcat控制台和http://127.0.0.1:8161/admin/中可以看消息已发送了，但是在tomcat控制台中和http://127.0.0.1:8161/admin/上看不到是否已经接受到了消息，我该如何查看或修改，谢谢！！！

http://127.0.0.1:8161/admin/这个界面中你可以查看queue和topic的信息列表

Name  topic或者queue的名字 
Number Of Pending Messages   pend消息的数目
Number Of Consumers   消费者的数目
Messages Sent   发送的消息数目
Messages Received  接收的消息数目

你要看有多少消息被接受了, 看Messages Received  条目

谢谢楼主，已解决，原因是我的TestBean忘了序列化接口(我用了defaultMessageConvertor)，在我加上LOG后才发现问题，实现了序列化接口就好了，谢谢！！

博主你好，
我怎么在http://127.0.0.1:8161/admin/topics.jsp中看不到JMS-TEST-TOPIC消息传递的效果呢，消息数量和名称等都看不到
附加本例子所需要的jar包
activemq-all-5.2.0.jar
commons-logging-1.1.jar
log4j-1.2.14.jar
spring2.5.5.jar
xbean-spring-2.8.jar(2.8以上)
jdk
jee

这些jar包都可以在activemq的解压包中找到

给个贴图看看

谢谢博主，可能是我运行的有问题，或者对ActiveMQ理解不够，只能看到消息发送、接收在控制台的输出，如果不单独启动ActiveMQ,连http://localhost:8161/admin都无法访问。
我的理解是Spring整合ActiveMQ后，运行TestMain后，ActiveMQ的broker会自动启动，但是为什么不能通过http://localhost:8161/admin访问。

谢谢博主，我的问题解决了，要想在http://127.0.0.1:8161/admin/queues.jsp中看到，必须把<!--  embedded ActiveMQ Broker --> 
    <amq:broker useJmx="false" persistent="true"> 
        <amq:persistenceAdapter> 
            <amq:amqPersistenceAdapter directory="d:/amq"/> 
        </amq:persistenceAdapter> 
        <amq:transportConnectors> 
            <amq:transportConnector uri="tcp://localhost:61616" /> 
                       <amq:transportConnector uri="vm://localhost:0" /> 
        </amq:transportConnectors> 
    </amq:broker> 这一段中的<amq:transportConnectors> 
            <amq:transportConnector uri="tcp://localhost:61616" /> 
                       <amq:transportConnector uri="vm://localhost:0" /> 
        </amq:transportConnectors>注释掉，即使用外部的ActiveMQ服务

我的消息每次都会重复被接受，这是为什么呢？
结果如下
13:43:43,921 INFO  [STDOUT] ****************** Topic B : ID001BUG管理需求和设计.
doc
13:43:43,921 INFO  [STDOUT] ****************** Topic B : ID001BUG管理需求和设计.
doc
13:43:43,921 INFO  [STDOUT] ************** Topic A : ID001BUG管理需求和设计.doc
13:43:43,921 INFO  [STDOUT] ************** Topic A : ID001BUG管理需求和设计.doc

=====================spring配置文件如下============================

<beans
		xmlns="http://www.springframework.org/schema/beans"
		xmlns:amq="http://activemq.org/config/1.0"
		xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
		xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd http://activemq.org/config/1.0 http://activemq.apache.org/schema/core/activemq-core-5.0-SNAPSHOT.xsd" >
 <!-- 
    	推荐版本，使用spring的listenerContainer，消息用数据库持久化保存，服务器重启不会丢失
     -->

	<!--  embedded ActiveMQ Broker -->
	<amq:broker useJmx="false" persistent="true">
		<amq:persistenceAdapter>
			<amq:jdbcPersistenceAdapter id="jdbcAdapter" dataSource="#dataSource" createTablesOnStartup="true"
										useDatabaseLock="false"/>
			<!-- 
				Mysql can setup useDatabaseLock="true",this is defualt
				HSQLDB,MSSQL plz setup useDatabaseLock="false",
				if u setup useDatabaseLock="true",u will catch error:
				MSSQL Error Info:FOR UPDATE clause allowed only for DECLARE CURSOR 
				HSQLDB Error Info:FOR in statement [SELECT * FROM ACTIVEMQ_LOCK FOR UPDATE]

				see http://www.nabble.com/ActiveMQ-JDBC-Persistence-with-SQL-Server-tf2022248.html#a5560296
			-->
		</amq:persistenceAdapter>
		<amq:transportConnectors>
			<amq:transportConnector uri="tcp://localhost:0"/>
		</amq:transportConnectors>
	</amq:broker>
     <!-- 连接外部的activeMQ
	<amq:connectionFactory id="jmsConnectionFactory" brokerURL="tcp://localhost:61616?wireFormat.maxInactivityDuration=0" />
	  ActiveMQ connectionFactory  -->
	  
	<!--  ActiveMQ connectionFactory  连接内部的-->
	<amq:connectionFactory id="jmsConnectionFactory" brokerURL="vm://localhost"/>

	<!--  ActiveMQ destinations  使用Queue方式 -->
	<amq:queue name="destination" physicalName="org.apache.activemq.spring.Test.spring.embedded"/>
	<!--  使用topic方式-->
	<amq:topic name="TOPIC" physicalName="JMS-TEST-TOPIC" />

	<!-- The msSQL Datasource that will be used by the Broker
	<bean id="mssql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
		<property name="driverClassName" value="net.sourceforge.jtds.jdbc.Driver"/>
		<property name="url">
			<value>jdbc:jtds:sqlserver://localhost:1433/wmjqxgl;SelectMethod=cursor;charset=GBK;tds=8.0;lastupdatecount=true</value>
		</property>
		<property name="username" value="sa"/>
		<property name="password" value="sa"/>
		<property name="poolPreparedStatements" value="true"/>
	</bean>	
 	-->
	<!--  Spring JmsTemplate config -->
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory">
			<!--  lets wrap in a pool to avoid creating a connection per send -->
			<bean class="org.springframework.jms.connection.SingleConnectionFactory">
				<property name="targetConnectionFactory" ref="jmsConnectionFactory"/>
			</bean>
		</property>
		<!-- custom MessageConverter -->
		<property name="messageConverter" ref="orderMessageConverter"/>
	</bean>

	<!--  OrderMessage converter  -->
	<bean id="orderMessageConverter" class="com.work.activemq.OrderMessageConverter"/>

	<!-- POJO which send Message uses  Spring JmsTemplate -->
	<bean id="orderMessageProducer" class="com.work.activemq.OrderMessageProducer">
		<property name="template" ref="jmsTemplate"/>
		<property name="destination" ref="destination"/>
	</bean>
	<!-- topic 方式信息发送者 -->
	<bean id="topicMessageProducer" class="com.work.activemq.TopicMessageProducer">
		<property name="template" ref="jmsTemplate" />
		<property name="destination" ref="TOPIC" />
	</bean>
		
    <!-- consumer1 for topic a 消息消费者 -->
    <bean id="topicConsumerA" class="com.work.activemq.TopicConsumerA" />

    <!-- consumer2 for topic a -->
    <bean id="topicConsumerB" class="com.work.activemq.TopicConsumerB" />	
    
    
    <!-- Message Listener for  -->
	<bean id="topicListenerA" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
		<constructor-arg ref="topicConsumerA" />
		<!--  may be other method -->
		<property name="defaultListenerMethod" value="receive" />
		<!-- custom MessageConverter define -->
		<property name="messageConverter" ref="orderMessageConverter" />
	</bean>

	<bean id="topicListenerB" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
		<constructor-arg ref="topicConsumerB" />
		<!--  may be other method -->
		<property name="defaultListenerMethod" value="receive" />
		<!-- custom MessageConverter define -->
		<property name="messageConverter" ref="orderMessageConverter" />
	</bean>
    
	<!--  Message Driven POJO (MDP)  通过queue的方式发送消息，一个发送一个接收-->
	<bean id="messageListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
		<constructor-arg>
			<bean class="com.work.activemq.OrderMessageConsumer">
				<!-- <property name="mailService" ref="mailService"/>  -->
			</bean>
		</constructor-arg>
		<!--  may be other method -->
		<property name="defaultListenerMethod" value="sendEmail"/>
		<!-- custom MessageConverter define -->
		<property name="messageConverter" ref="orderMessageConverter"/>
	</bean>

	<!--  listener container，MDP无需实现接口 -->
	<bean id="listenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="jmsConnectionFactory"/>
		<property name="destination" ref="destination"/>
		<property name="messageListener" ref="messageListener"/>
	</bean>
	
	<!--  listener container，MDP无需实现接口 -->
	<bean id="topicListenerContainerA" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="jmsConnectionFactory" />
		<property name="destination" ref="TOPIC" />
		<property name="messageListener" ref="topicListenerA" />
	</bean>

    <bean id="topicListenerContainerB" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="jmsConnectionFactory" />
		<property name="destination" ref="TOPIC" />
		<property name="messageListener" ref="topicListenerB" />
	</bean>
	
	<bean id="orderNotify" class="com.work.activemq.OrderNotifyImpl">
		<property name="orderMessageProducer" ref="orderMessageProducer" />
		<property name="topicMessageProducer" ref="topicMessageProducer" />
	</bean>
	<!--  -->

	
</beans>

==================

TO: wmj2003

把你的java代码也贴出来看看

package com.work.activemq;


import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.HashMap;
import java.util.Map;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.command.ActiveMQObjectMessage;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.jms.support.converter.MessageConverter;

/**
 * 订单消息转换类. 使用数据库持久化消息的时候使用。
 * <p>实现MessageConverter使得JMS的发送与接收者可以直接发送POJO而不是JMS Message.</p>
 *
 * @author wangmj
 * @see MessageConverter
 */

public class OrderMessageConverter implements MessageConverter {
	private static final Log log = LogFactory.getLog(OrderMessageConverter.class);
	/*
	 * (non-Javadoc)
	 *
	 * @see org.springframework.jms.support.converter.MessageConverter#toMessage(java.lang.Object,
	 *      javax.jms.Session)
	 */
	public Message toMessage(Object obj, Session session) throws JMSException {
		//check Type
		if (obj instanceof Order) {
			ActiveMQObjectMessage objMsg = (ActiveMQObjectMessage) session.createObjectMessage();
			   HashMap<String, byte[]> map = new HashMap<String, byte[]>();  
			try{
				// POJO must implements Seralizable
				ByteArrayOutputStream bos = new ByteArrayOutputStream();
				ObjectOutputStream oos = new ObjectOutputStream(bos);
				oos.writeObject(obj);
				map.put("Order", bos.toByteArray());
				objMsg.setObjectProperty("Map", map);
			} catch (IOException e) {
				log.error("toMessage(Object, Session)", e);
			}
			return objMsg;
		} else {
			throw new JMSException("Object:[" + obj + "] is not Order");
		}

	}

	/*
	 * (non-Javadoc)
	 *
	 * @see org.springframework.jms.support.converter.MessageConverter#fromMessage(javax.jms.Message)
	 */
	@SuppressWarnings("unchecked")
	public Object fromMessage(Message msg) throws JMSException {
		if (msg instanceof ObjectMessage) {
			 HashMap<String, byte[]> map = (HashMap<String, byte[]>) ((ObjectMessage) msg).getObjectProperty("Map"); 
			try {
				// POJO must implements Seralizable
				ByteArrayInputStream bis=new ByteArrayInputStream((byte[])map.get("Order") );
				ObjectInputStream ois = new ObjectInputStream(bis);
				Object returnObject = ois.readObject();
				return returnObject;
			} catch (IOException e) {
				log.error("fromMessage(Message)", e);

			} catch (ClassNotFoundException e) {
				log.error("fromMessage(Message)", e);
			}
			
			return null;
		} else {
			throw new JMSException("Msg:[" + msg + "] is not Map");
		}
	}

}

package com.work.activemq;

import javax.jms.Topic;

import org.springframework.jms.core.JmsTemplate;

/**
 * 奇怪啊，topic方式，消息被重复消费了。每次都消费两次。
 * 一旦更换为DefaultMessageConverter.java ，控制台就看不到打印信息了。<br>
 * 但是在activemq的监控界面，可以看到消息被消费了，而是web发送信息页面也提示成功。
 * 估计是和Serializable有关系。
 * @author wangmingjie
 * @date 2009-7-26上午11:21:32
 */
public class TopicMessageProducer {
    
    private JmsTemplate template;

	private Topic destination;

	public void setTemplate(JmsTemplate template) {
		this.template = template;
	}

	public void setDestination(Topic destination) {
		this.destination = destination;
	}

	/**
	 * 发送信息
	 * @param message
	 */
	public void send(Order message) {
		template.convertAndSend(this.destination, message);
	}
}

package com.work.activemq;
/**
 * @author wangmingjie
 * @date 2009-7-26上午11:27:00
 */

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 使用多线程接受消息
 */
public class TopicConsumerA {

	ExecutorService exec = Executors.newFixedThreadPool(10);

	public void receive(final Order message) {
		//System.out.println("线程名称"+Thread.currentThread().getName());
		exec.submit((new Runnable() {
			public void run() {
				//stem.out.println(Thread.currentThread().getName());
				System.out.println("**** Topic A : " + message.getId()+message.getName());
			}
		}));
	}

}

package com.work.activemq;
/**
 * @author wangmingjie
 * @date 2009-7-26上午11:27:05
 */
public class TopicConsumerB {

	public void receive(Order message) {
		System.out.println("**** Topic B : " + message.getId()+message.getName());
	}
}

package com.work.activemq;

public interface OrderNotifyInterface {
	/**
	 *  通知用户订单已经发送！生成者接口
	 * @param order
	 */
	public void notifyOrder(Order order);
	
	/**
	 * 通过topic的方式发送信息
	 * @param order
	 */
	public void notifyTopic(Order order);
}

实现如下：
package com.work.activemq;

/**
 * 用来测试mq的。
 * @author wangmingjie
 *
 */
public class OrderNotifyImpl implements OrderNotifyInterface {
	private OrderMessageProducer orderMessageProducer;
	public void setOrderMessageProducer(OrderMessageProducer orderMessageProducer) {
		this.orderMessageProducer = orderMessageProducer;
	}
	
	private TopicMessageProducer  topicMessageProducer;

	public void setTopicMessageProducer(TopicMessageProducer topicMessageProducer) {
		this.topicMessageProducer = topicMessageProducer;
	}
	
	/**
	 * 向顾客的邮箱发送订单通知,使用JMS发送.
	 */
	public void notifyOrder(Order order) {
		orderMessageProducer.send(order);
	}
	

	public void notifyTopic(Order order){
		topicMessageProducer.send(order);
	}
}

TestMain里面的
org.springframework.beans.factory.generic.GenericBeanFactoryAccessor;

依赖的是哪个包，我现在用的是Spring 3.0.5,并且加载了spring的所有包，还是提示
The import org.springframework.beans.factory.generic cannot be resolved，我下载的是
xbean-spring-3.5.jar 仍然找不到这个类，麻烦楼主把所有需要的jar包给列出来，谢谢

我的这个GenericBeanFactoryAccessor也是找不到，spring和activemq的所有jar都导入了，还是找不到

4年前的文章，2012才享受到，悲剧，写楼主了