正文
SpringBoot集成ActiveMQ
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
前面提到了原生API访问ActiveMQ和Spring集成ActiveMQ。今天讲一下SpringBoot集成ActiveMQ.
SpringBoot就是为了解决我们的Maven配置烦恼而生，因此使用SpringBoot无疑就会让我们减少很多繁琐的xml配置。具体操作如下：
1.pom.xml
2.application.properties
3.启动类ActiveBootApplication
4.生产者Producer
5.消费者Consumer
6.测试代码
1.pom.xml
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-activemq</artifactId>
</dependency> <dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
2.application.properties
# 用户名和密码都是空
spring.activemq.broker-url=tcp://localhost:
spring.activemq.user=
spring.activemq.password=
spring.activemq.in-memory=true
# 不使用连接池
spring.activemq.pool.enabled=false
3.启动类ActiveBootApplication
@SpringBootApplication
@EnableJms //必须要启动
public class ActiveBootApplication { public static void main(String[] args) {
SpringApplication.run(ActiveBootApplication.class, args);
}
}
4.生产者Producer
@Service
public class Producer {
@Autowired // 也可以注入JmsTemplate，JmsMessagingTemplate对JmsTemplate进行了封装 ，JmsTemplate可以进行更为细微的操作
private JmsMessagingTemplate jmsTemplate; public void sendMessage(Destination destination, final String message){
jmsTemplate.convertAndSend(destination, message);
}
}
5.消费者Consumer
@Component
public class Consumer {
// 使用JmsListener配置消费者监听的队列，其中text是接收到的消息
@JmsListener(destination = "mytest.queue")
public void receiveQueue(String text) {
System.out.println(this.getClass().getName() + " 收到的报文为:" + text);
}
}
6.测试代码
@RunWith(SpringRunner.class)
@SpringBootTest //springboot测试注解
public class ActiveBootApplicationTests { @Autowired
private Producer producer; @Test
public void contextLoads() {
//定义队列模式，而非Topic模式
Destination destination = new ActiveMQQueue("mytest.queue"); for (int i = 0; i < 5; i++) {
String msg = "Hello World!" + i;
producer.sendMessage(destination, msg);
System.out.println("send:" + msg);
}
}
}
注意：
1.必须要启动@EnableJms
2.次示例没有池化
3.测试代码里面定义点对点模式还是订阅模式
下面再对池化的代码进行说明：
1.application.properties修改如下：
spring.activemq.broker-url=tcp://localhost:
spring.activemq.user=
spring.activemq.password=
spring.activemq.in-memory=true # 启动池化
spring.activemq.pool.enabled=true
# 最大连接数
spring.activemq.pool.max-connections=
#表示空闲多少时间将被清理，而expiry-timeout则表示创建之后多少时间后就会被清理（0表示永不过期）
spring.activemq.pool.idle-timeout=
spring.activemq.pool.expiry-timeout=
2.增加配置类ActiveMqConfiguration
@Configuration
public class ActiveMqConfiguration { @Value("${spring.activemq.user}")
private String usrName; @Value("${spring.activemq.password}")
private String password; @Value("${spring.activemq.broker-url}")
private String brokerUrl; //池化必须配置，否者会出现找不到org.springframework.jms.core.JmsMessagingTemplate的异常
@Bean
public ConnectionFactory connectionFactory(){
ActiveMQConnectionFactory connectionFactory
= new ActiveMQConnectionFactory();
connectionFactory.setBrokerURL(brokerUrl);
connectionFactory.setUserName(usrName);
connectionFactory.setPassword(password);
return connectionFactory;
} }
其它代码保持一致即可。
下面对Topic的监听进行说明：
1.需要指定消费者的监听类型为Topic，即订阅模式，在ActiveMqConfiguration增加代码：
@Bean("jmsTopicListenerContainerFactory")
public JmsListenerContainerFactory<?> jmsTopicListenerContainerFactory(
ConnectionFactory connectionFactory
){
DefaultJmsListenerContainerFactory factory
= new DefaultJmsListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setPubSubDomain(true);
return factory;
}
2.消费者注解增加containerFactory
@Component
public class Consumer {
// 使用JmsListener配置消费者监听的队列，其中text是接收到的消息
@JmsListener(destination = "mytest.queue",containerFactory="jmsTopicListenerContainerFactory")
public void receiveQueue(String text) {
System.out.println(this.getClass().getName() + " 收到的报文为:" + text);
}
}
注意containerFactory的值和前面定义Bean的名称保持一直。不写containerFactory就表示队列模式。
