Springboot整合ActiveMQ

一、安装ActiveMQ

Linux版本的ActiveMQ安装包

tar zxvf apache-activemq-5.15.2-bin.tar.gz
cd apache-activemq-5.15.2
./bin/activemq

直接启动activemq

之后访问 ip 为自己的ip 就可以看到自己的界面

登录进去的话：默认账号密码为: admin/admin

http://192.168.2.127:8161/

二、新建SpringBoot工程

项目结构为：



2.1：pom.xml

<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.4.3.RELEASE</version>
<relativePath /> <!-- lookup parent from repository -->
</parent>

<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
</properties>

<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-aop -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!--spring boot start -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-thymeleaf</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
<!-- <version>5.7.0</version> -->
</dependency>
</dependencies>

<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>

三、修改application.properties文件

##自己的activeMQ访问路径
spring.activemq.broker-url= tcp://192.168.2.127:61616
spring.activemq.in-memory=true
# Login password of the broker.
spring.activemq.password=admin
spring.activemq.user=admin
# Login user of the broker.
spring.activemq.packages.trust-all=false
# Trust all packages.
spring.activemq.pool.enabled=false

四、消息生产者：Producer

package com.htf.activemq;

import javax.jms.Destination;
import javax.jms.Queue;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
@EnableScheduling
public class Producer {

@Autowired
private JmsMessagingTemplate jmsMessagingTemplate;

@Autowired
private Queue queue;

@Scheduled(fixedDelay=3000)//每3s执行1次
public void send() {
this.jmsMessagingTemplate.convertAndSend(this.queue, "hi,activeMQ");
}

public void send(String msg) {
this.jmsMessagingTemplate.convertAndSend(this.queue, msg);
}
// 发送消息，destination是发送到的队列，message是待发送的消息
public void sendMessage(Destination destination, final String message){
jmsMessagingTemplate.convertAndSend(destination, message);
}

@JmsListener(destination="out.queue")
public void consumerMessage(String text){
System.out.println("从out.queue队列收到的回复报文为:"+text);
}
}

五、消息消费者：Consumer

package com.htf.activemq;

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

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

@JmsListener(destination = "mytest.queue")
public void receiveQueue1(String text) {
System.out.println("Consumer收到的报文为:"+text);
}
}

六、启动项目

package com.htf;

import javax.jms.Queue;

import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.annotation.EnableAsync;

@SpringBootApplication
public class Application {
@Bean
public Queue queue() {
return new ActiveMQQueue("sample.queue");
}
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
}

控制台消息：



每隔三秒发送一次是因为消息生产者设置了

@Scheduled(fixedDelay=3000)//每3s执行1次

activeMQ网站可观察到



七、添加Consumer2返回确认收到报文

package com.htf.activemq;

import java.util.concurrent.Executors;

import org.springframework.jms.annotation.JmsListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

@Component
public class Consumer2 {
@JmsListener(destination = "mytest.queue")
@SendTo("out.queue")
public String receiveQueue(String text) {
System.out.println("Consumer2收到的报文为:"+text);
Executors.newFixedThreadPool(10);
return "return message:"+text;
}
}

八添加web测试

TestController

package com.htf.controller;

import javax.jms.Destination;
import javax.servlet.http.HttpServletRequest;

import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import com.htf.activemq.Producer;

@RestController
public class TestController {

@Autowired
private Producer producer;

@GetMapping("/activemq/send")
public String activemq(HttpServletRequest request, String msg) {
msg = StringUtils.isEmpty(msg) ? "This is Empty Msg." : msg;

try {
producer.send(msg);
} catch (Exception e) {
e.printStackTrace();
}
return "Activemq has sent OK.";
}

@GetMapping("/activemq/send1")
public String receiveQueue(String text) {
Destination destination = new ActiveMQQueue("mytest.queue");

for(int i=0; i<100; i++){
producer.sendMessage(destination, "myname is htf!!!");
}
return "success";
}

}

启动项目。浏览器访问后台会看到消息test log

http://127.0.0.1:8080/activemq/send?msg=test log

访问http://127.0.0.1:8080/activemq/send1 后台会看到消息生产者收到确认报文

http://127.0.0.1:8080/activemq/send1

ActiveMQ监控页面可以看到队列消息



topics



项目源码路径

项目源码