1,导入spring相关依赖
spring-context,spring-test,junit(测试包)
activemq-all(消息队列核心包)
spring-jms(Spring提供的activeMQ模板)
2,在application.xml中配置activeMQ连接工厂,和Spring cache连接工厂
<!-- ActiveMQ 连接工厂 --> <!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供--> <!-- 如果连接网络:tcp://ip:61616;未连接网络:tcp://localhost:61616 以及用户名,密码--> <!-- <amq:connectionFactory id="amqConnectionFactory" --> <!-- brokerURL="tcp://localhost:61616" userName="admin" password="admin" /> --> <bean id="amqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="tcp://localhost:61616"></property> <property name="userName" value="admin"></property> <property name="password" value="admin"></property> </bean>
<!-- Spring Caching连接工厂 --> <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory --> <bean id="mqConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory --> <property name="targetConnectionFactory" ref="amqConnectionFactory"></property> <!-- 同上,同理 --> <!-- <constructor-arg ref="amqConnectionFactory" /> --> <!-- Session缓存数量 --> <property name="sessionCacheSize" value="1" /> </bean>
<!-- Spring JmsTemplate 的消息生产者 --> <!-- 根据不同的id,根据需要进行调用 --> <!-- 定义JmsTemplate的Queue类型 --> <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 --> <constructor-arg ref="mqConnectionFactory" /> <!-- 非pub/sub模型(发布/订阅),即队列模式 --> <property name="pubSubDomain" value="false" /> </bean> <!-- 定义JmsTemplate的Topic类型 --> <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 --> <constructor-arg ref="mqConnectionFactory" /> <!-- pub/sub模型(发布/订阅) --> <property name="pubSubDomain" value="true" /> </bean>
-------------------------
创建生产者和消费者
进行测试代码编写
1,定义topic和queue两种消息生产者
@Autowired
private TopicSender topicSender;
@Autowired
private QueueSender QueueSender;
分别调用发送消息:topicSender.send("spring-topic","你好吗?"),QueueSender.send("spring-queue","你好吗?")
2,编写消费者代码
public class QueueConsumer1 implements MessageListener{
public void onMessage(Message message){
TestMessage message = (TestMessage)message;
sout("QueueConsumer1 获取消息"+message .getText());
}
}
QueueConsumer2,TopicConsumer1,TopicConsumer2结构同QueueConsumer1
对消费者监听器进行配置.
创建测试类:执行
运行结果为:
//queue只能被消费一次
QueueConsumer1获取消息你好吗(或者QueueConsumer2获取消息你好吗)
//Topic则可以被各个配置的消费者消费
TopicConsumer1获取消息你好吗
TopicConsumer2获取消息你好吗