SpringBoot集成JmsTemplate(队列模式和主题模式)及xml和JavaConfig配置详解_Java

1.导入jar包：

 <!--jmstemplate-->
  <dependency>
   <groupid>org.springframework.boot</groupid>
   <artifactid>spring-boot-starter-activemq</artifactid>
  </dependency>
  <dependency>
   <groupid>org.apache.activemq</groupid>
   <artifactid>activemq-pool</artifactid>
  </dependency>

2.填写配置文件（application.properties）

#设置jms（amq）
spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.user=admin
spring.activemq.password=admin
spring.activemq.pool.enabled=true
#spring.jms.pub-sub-domain=true
spring.activemq.pool.max-connections=50
spring.activemq.pool.expiry-timeout=10000
spring.activemq.pool.idle-timeout=30000

上面需要注意的是，如果开启订阅者和发布者模式下面的代码会使监听器失效。

3.编写控制器代码

@restcontroller
@requestmapping("/jms")
public class producercontroller {
 
 @autowired
 private jmsproducerservice jmsproducerservice;
 
 @requestmapping("/send")
 public void contextloads() throws interruptedexception {
  destination destination = new activemqqueue("mytest.queue");
  for(int i=0; i<10; i++){
   jmsproducerservice.sendmessage(destination, "我是超人啊");
  }
  system.out.println("发送成功");
 }
 
}

4.服务层代码：

package com.zzf.finals.service.impl;
 
import com.zzf.finals.service.jmsproducerservice;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.jms.core.jmstemplate;
import org.springframework.stereotype.service;
 
import javax.jms.destination;
 
@service
public class jmsproducerserviceimpl implements jmsproducerservice {
 
 @autowired
 private jmstemplate jmstemplate;
 
 @override
 public void sendmessage(destination destination, string message) {
  this.jmstemplate.convertandsend(destination,message);
 }
}

5.最后加上监听器类

package com.zzf.finals.domain;
 
import org.springframework.jms.annotation.jmslistener;
import org.springframework.stereotype.component;
 
@component
public class consumer {
 
 @jmslistener(destination = "mytest.queue")
 public void receivequeue(string text) {
  system.out.println("message:"+text);
 }
}

ok~

但是这样有另外一个问题：如果开启了订阅者和发布者模式则无法发送和接收queue消息。

这里我提供两种写法xml和java配置：

首先贴上我的xml配置代码

<?xml version="1.0" encoding="utf-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
  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.xsd">
 
 <!--连接池，内部引入一个连接工厂-->
 <bean id="jmsfactory" class="org.apache.activemq.pool.pooledconnectionfactory"
   destroy-method="stop">
  <property name="connectionfactory">
   <bean class="org.apache.activemq.activemqconnectionfactory">
    <property name="brokerurl">
     <value>tcp://localhost:61616</value>
    </property>
   </bean>
  </property>
  <property name="maxconnections" value="100"></property>
 </bean>
 
 
 <bean id="destinationqueue" class="org.apache.activemq.command.activemqqueue">
  <constructor-arg name="name" value="spring-queue"/>
 </bean>
 
 <!--测试topic-->
 <bean id="destinationtopic" class="org.apache.activemq.command.activemqtopic">
  <constructor-arg name="name" value="spring-topic"/>
 </bean>
 
 <!--配置消息容器-->
 <bean id="topiccontainers" class="org.springframework.jms.config.defaultjmslistenercontainerfactory">
  <property name="pubsubdomain" value="true"/>
  <property name="connectionfactory" ref="jmsfactory"/>
 </bean>
 
 <!--配置队列消息容器-->
 <bean id="queuecontainers" class="org.springframework.jms.config.defaultjmslistenercontainerfactory">
  <property name="connectionfactory" ref="jmsfactory"/>
 </bean>
 
</beans>

javaconfig配置为：

package com.zzf.finals.domain;
 
import org.apache.activemq.command.activemqqueue;
import org.apache.activemq.command.activemqtopic;
import org.springframework.context.annotation.bean;
import org.springframework.context.annotation.configuration;
import org.springframework.jms.config.defaultjmslistenercontainerfactory;
import org.springframework.jms.config.jmslistenercontainerfactory;
 
import javax.jms.connectionfactory;
import javax.jms.queue;
import javax.jms.topic;
 
@configuration
public class jmsconfig {
 public final static string topic = "topic.test";
 public final static string queue = "queue.test";
 @bean
 public queue queue() {
  return new activemqqueue(queue);
 }
 
 @bean
 public topic topic() {
  return new activemqtopic(topic);
 }
 
 // topic模式的listenercontainer
 @bean
 public jmslistenercontainerfactory<?> jmslistenercontainertopic(connectionfactory activemqconnectionfactory) {
  defaultjmslistenercontainerfactory bean = new defaultjmslistenercontainerfactory();
  bean.setpubsubdomain(true);
  bean.setconnectionfactory(activemqconnectionfactory);
  return bean;
 }
 // queue模式的listenercontainer
 @bean
 public jmslistenercontainerfactory<?> jmslistenercontainerqueue(connectionfactory activemqconnectionfactory) {
  defaultjmslistenercontainerfactory bean = new defaultjmslistenercontainerfactory();
  bean.setconnectionfactory(activemqconnectionfactory);
  return bean;
 }
 
}

控制台代码为：

package com.zzf.finals.controller;
 
import com.zzf.finals.service.jmsproducerservice;
import org.apache.activemq.command.activemqqueue;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.web.bind.annotation.requestmapping;
import org.springframework.web.bind.annotation.restcontroller;
 
import javax.jms.destination;
import javax.jms.queue;
import javax.jms.topic;
 
@restcontroller
@requestmapping("/jms")
public class producercontroller {
 
 @autowired
 private jmsproducerservice jmsproducerservice;
 @autowired
 private topic topic;
 @autowired
 private queue queue;
 
 @autowired
 private topic destinationtopic;
 @autowired
 private queue destinationqueue;
 
 
 @requestmapping("/send3")
 public void testjms2() {
  for (int i=0;i<10;i++) {
   jmsproducerservice.sendmessage(destinationqueue,"queue,world!" + i);
   jmsproducerservice.sendmessage(destinationtopic, "topic,world!" + i);
  }
 }
 
 @requestmapping("/send2")
 public void testjms() {
  for (int i=0;i<10;i++) {
   jmsproducerservice.sendmessage(queue,"queue,world!" + i);
   jmsproducerservice.sendmessage(topic, "topic,world!" + i);
  }
 }
 
  @requestmapping("/send")
 public void contextloads() throws interruptedexception {
  destination destination = new activemqqueue("mytest.queue");
  for(int i=0; i<10; i++){
   jmsproducerservice.sendmessage(destination, "我是超人啊");
  }
  system.out.println("发送成功");
 }
}

最后的监听器类：

package com.zzf.finals.domain;
 
import org.springframework.jms.annotation.jmslistener;
import org.springframework.stereotype.component;
 
@component
public class consumer {
 
 @jmslistener(destination = "mytest.queue")
 public void receivequeue(string text) {
  system.out.println("message:"+text);
 }
 
 @jmslistener(destination = jmsconfig.topic,containerfactory = "jmslistenercontainertopic")
 public void ontopicmessage(string msg) {
  system.out.println("topic:"+msg);
 }
 
 @jmslistener(destination = jmsconfig.queue,containerfactory = "jmslistenercontainerqueue")
 public void onqueuemessage(string msg) {
  system.out.println("queue:"+msg);
 }
 
 @jmslistener(destination = "spring-topic",containerfactory = "topiccontainers")
 public void ontopicmessagexml(string msg) {
  system.out.println("topic1:"+msg);
 }
 @jmslistener(destination = "spring-topic",containerfactory = "topiccontainers")
 public void ontopicmessagexml2(string msg) {
  system.out.println("topic2:"+msg);
 }
 
 @jmslistener(destination = "spring-queue",containerfactory = "queuecontainers")
 public void onqueuemessagexml(string msg) {
  system.out.println("queue:"+msg);
 }
}