Spring Boot集成ActiveMQ

in-memory ActiveMQ

添加依赖

1
2
3
4
5
6
<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-activemq -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-activemq</artifactId>
<version>2.3.2.RELEASE</version>
</dependency>

修改ActiveMQ配置

application.properties配置:

1
2
spring.activemq.in-memory=true
spring.activemq.pool.enabled=false

备注:实际上这两个是默认值,不配置情况下即是如此

添加Producer类

1
2
3
4
5
6
7
8
9
10
11
12
13
@RestController
public class Producer {
@Autowired
private JmsMessagingTemplate jmsMessagingTemplate;

@Autowired
private Queue queue;

@RequestMapping("/sendMsg")
public void send(String msg) {
this.jmsMessagingTemplate.convertAndSend(this.queue, msg);
}
}

使用rest服务方式来触发发送消息的方法

添加Consumer类

1
2
3
4
5
6
7
8
@Component
public class Consumer {

@JmsListener(destination = "sample.queue")
public void receiveQueue(String text) {
System.out.println(text);
}
}

JmsListenerspring-jms提供的一个注解,会实例化一个Jms的消息监听实例,也就是一个异步的消费者

添加JMS的注解扫描

1
2
3
4
5
6
7
8
9
10
11
12
@SpringBootApplication
@EnableJms
public class ActivemqDemoApplication {
@Bean
public Queue queue() {
return new ActiveMQQueue("sample.queue");
}

public static void main(String[] args) {
SpringApplication.run(ActivemqDemoApplication.class, args);
}
}

@EnableJms会启动jms的注解扫描,相当于<jms:annotation-d riven/>

启动及测试

在浏览器输入:http://localhost:8080/sendMsg?msg=HelloActiveMQ

连接外部的ActiveMQ

修改配置

官方的实例用的是in-memory的ActiveMQ。然而我们实际使用的时候都是连接外部共用的ActiveMQ服务。所以,我们修改一下配置,来使用外部的ActiveMQ服务。

1
2
3
4
5
6
7
spring.activemq.broker-url=tcp://localhost:61616
#spring.activemq.broker-url=failover:(tcp://localhost:61616,tcp://localhost:61617)
spring.activemq.close-timeout=5000
spring.activemq.in-memory=false
spring.activemq.pool.enabled=true
spring.activemq.pool.max-connections=100
spring.activemq.send-timeout=3000

这里被注释掉的spring.activemq.broker-url是对应ActiveMQ集群时候的broker-url配置。

如果未找到activemq-pool依赖包,则自行添加:

1
2
3
4
5
6
<!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-pool -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
<version>5.16.0</version>
</dependency>

启动服务

当没有启动ActiveMQ时,Spring Boot启动失败:

1
2
3
4
5
6
7
8
9
10
11
12
13
***************************
APPLICATION FAILED TO START
***************************

Description:

Field jmsMessagingTemplate in com.example.demo.activemq.Producer required a bean of type 'org.springframework.jms.core.JmsMessagingTemplate' that could not be found.
- Bean method 'jmsMessagingTemplate' not loaded because Ancestor org.springframework.boot.autoconfigure.jms.JmsAutoConfiguration did not match


Action:

Consider revisiting the conditions above or defining a bean of type 'org.springframework.jms.core.JmsMessagingTemplate' in your configuration.

但是失败的消息提示很诡异,告诉的是找不到JmsMessagingTemplate,而不是提示连接不上ActiveMQ

按照提示信息里面说的设置debug=true,能够打印出auto-configure中相关类的匹配信息,但是仍然不能直接定位到是ActiveMQ服务连接不上。

如果遇到上述问题,可以尝试检查ActiveMQ服务是否能够连接。

一个jmsTemplate支持queue和topic

一个jmsTemplate支持queuetopic的消息发送,只需要指定不同的destination即可。

但是消息的接收,也就是@JmsListener如果不指定独立的containerFactory的话是只能消费queue消息的,如果要能够同时接收topic消息,需要给topic对应的@JmsListener增加containerFactory配置:

1
2
3
4
5
6
7
8
9
10
11
12
13
//需要给topic定义独立的JmsListenerContainer
@Bean
public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ConnectionFactory activeMQConnectionFactory) {
DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
bean.setPubSubDomain(true);
bean.setConnectionFactory(activeMQConnectionFactory);
return bean;
}

@JmsListener(destination = "sample.topic", containerFactory="jmsListenerContainerTopic")
public void receiveTopic(String text) {
System.out.println(text);
}