ActiveMQ整合SpringBoot

pom.xml

	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-activemq</artifactId>
		<version>1.5.0.RELEASE</version>
	</dependency>
	<!--消息队列连接池-->
	<dependency>
		<groupId>org.apache.activemq</groupId>
		<artifactId>activemq-pool</artifactId>
		<version>5.15.0</version>
	</dependency>

application.yml配置文件

spring:
activemq:
broker-url: tcp://127.0.0.1:61616
user: admin
password: admin
close-timeout: 15s # 在考虑结束之前等待的时间
in-memory: true # 默认代理URL是否应该在内存中。如果指定了显式代理，则忽略此值。
non-blocking-redelivery: false # 是否在回滚回滚消息之前停止消息传递。这意味着当启用此命令时，消息顺序不会被保留。
send-timeout: 0 # 等待消息发送响应的时间。设置为0等待永远。
queue-name: active.queue
topic-name: active.topic.name.model
pool:
enabled: true
max-connections: 10 #连接池最大连接数
idle-timeout: 30000 #空闲的连接过期时间，默认为30秒

ActiveMQ结构

ActiveConfig

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.config.JmsListenerContainerFactory;
import org.springframework.jms.config.SimpleJmsListenerContainerFactory;
import org.springframework.jms.core.JmsMessagingTemplate;

import javax.jms.ConnectionFactory;
import javax.jms.Queue;
import javax.jms.Topic;

@Configuration
public class ActiveConfig {

@Value("${spring.activemq.broker-url}")
private String brokerUrl;

@Value("${spring.activemq.user}")
private String username;

@Value("${spring.activemq.topic-name}")
private String password;

@Value("${spring.activemq.queue-name}")
private String queueName;

@Value("${spring.activemq.topic-name}")
private String topicName;

@Bean(name = "queue")
public Queue queue() {
    return new ActiveMQQueue(queueName);
}

@Bean(name = "topic")
public Topic topic() {
    return new ActiveMQTopic(topicName);
}

@Bean
public ConnectionFactory connectionFactory(){
    return new ActiveMQConnectionFactory(username, password, brokerUrl);
}

@Bean
public JmsMessagingTemplate jmsMessageTemplate(){
    return new JmsMessagingTemplate(connectionFactory());
}

// 在Queue模式中，对消息的监听需要对containerFactory进行配置
@Bean("queueListener")
public JmsListenerContainerFactory<?> queueJmsListenerContainerFactory(ConnectionFactory connectionFactory){
    SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setPubSubDomain(false);
    return factory;
}

//在Topic模式中，对消息的监听需要对containerFactory进行配置
@Bean("topicListener")
public JmsListenerContainerFactory<?> topicJmsListenerContainerFactory(ConnectionFactory connectionFactory){
    SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setPubSubDomain(true);
    return factory;
}

}

ProducerController

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
import javax.jms.Destination;
import javax.jms.Queue;
import javax.jms.Topic;

/**

• 生产者
  */
  @RestController
  public class ProducerController
  {
  @Autowired
  private JmsMessagingTemplate jmsMessagingTemplate;

  @Autowired
  private Queue queue;

  @Autowired
  private Topic topic;

  @PostMapping("/queue/test")
  public String sendQueue(@RequestBody String str) {
  this.sendMessage(this.queue, str);
  return “success”;
  }

  @PostMapping("/topic/test")
  public String sendTopic(@RequestBody String str) {
  this.sendMessage(this.topic, str);
  return “success”;
  }

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

queue模式的消费者 QueueConsumerListener

import com.alibaba.fastjson.JSONObject;
import com.jeesite.modules.alarm.service.BizAlarmService;
import com.jeesite.modules.influx.InfluxDBService;
import com.jeesite.modules.mqtt.config.MqttData;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class QueueConsumerListener {

@Resource
private InfluxDBService influxDBService;

@Resource
private BizAlarmService bizAlarmService;

//queue模式的消费者
@JmsListener(destination="${spring.activemq.queue-name}", containerFactory="queueListener", concurrency = "4-8")
public void readActiveQueue(String message) {
    MqttData mqttData= JSONObject.parseObject(message, MqttData.class);
    influxDBService.save(mqttData);
}

}

topic模式的消费者 TopicConsumerListener

import com.alibaba.fastjson.JSONObject;
import com.jeesite.modules.alarm.service.BizAlarmService;
import com.jeesite.modules.mqtt.config.MqttData;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class TopicConsumerListener {

@Resource
private BizAlarmService bizAlarmService;

//topic模式的消费者
@JmsListener(destination="${spring.activemq.topic-name}", containerFactory="topicListener", concurrency = "4-8")
public void readActiveQueue(String message) {
    MqttData mqttData= JSONObject.parseObject(message, MqttData.class);
    String choFaultCode = mqttData.getChoFaultCode();//故障码
    if (!"0".equals(choFaultCode)){
        bizAlarmService.addAlarm(choFaultCode,mqttData.getSn());
    }
}

}