springboot整合activeMQ

一、添加依赖

<!-- 整合消息队列ActiveMQ -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

<!-- 如果配置线程池则加入 -->
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-pool</artifactId>
</dependency>

二、配置文件增加

activemq:
  #默认端口号61616
  broker-url: tcp://127.0.0.1:61616
  #集群配置
  broker-url: failover:(tcp://localhost:61616,tcp://localhost:61617)
  user: admin
  password: admin
  pool:
    enabled: true
    max-connections: 100
#支持发布订阅模型，默认只支持点对点
jms:
  pub-sub-domain: true

三、开启注解

在启动类或者config类增加注解@EnableJms

四、配置生产者

如果需要配置默认的Queue，在配置类下新增bean

//队列
@Bean
public Queue queue() {
    return new ActiveMQQueue("connect.queue");
}
//发布订阅
@Bean
public Topic topic() {
    return new ActiveMQTopic("video.topic");
}

//配置修改工厂配置，即同时支持点对点和发布订阅模型
@Bean
public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ConnectionFactory activeMQConnectionFactory) {
    DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
    bean.setPubSubDomain(true);
    bean.setConnectionFactory(activeMQConnectionFactory);
    return bean;
}

配置Service和实现类。

//配置发送消息的队列
@Override
public void sendMessage(Destination destination, String message) {
    jmsTemplate.convertAndSend(destination, message);
}
//配置默认发送消息的队列
@Override
public void sendMessage(String message) {
    jmsTemplate.convertAndSend(this.queue, message);
}
//配置默认发布订阅
@Override
public void publish(String message) {
    this.jmsTemplate.convertAndSend(this.topic, message);
}

五、配置消费者

@Component
public class JmsConsumer {
    //监听器监听消息队列，如果有消息发送过来，触发该方法
    @JmsListener(destination = "connect.order")
    public void receiveQueue(String message) {
        System.out.println(LocalDateTime.now() + ":" + message);
    }
    //订阅消息，pub-sub-domain: true配置开启的情况下
    @JmsListener(destination = "video.topic")
    public void receiveTopic1(String message) {
        System.out.println(LocalDateTime.now() + " receive1:" + message);
    }
    //订阅消息,pub-sub-domain: true配置开启的情况下
    @JmsListener(destination = "video.topic")
    public void receiveTopic2(String message) {
        System.out.println(LocalDateTime.now() + " receive2:" + message);
    }
    //订阅消息,不开启配置，同时支持点对点和发布订阅模型
    @JmsListener(destination = "video.topic", containerFactory = "jmsListenerContainerTopic")
    public void receiveTopic3(String message) {
        System.out.println(LocalDateTime.now() + " receive3:" + message);
    }
}

六、队列的使用

@GetMapping("order")
public Map<String, Object> order(String message) {
    Map<String, Object> jsonMap = new HashMap<>(16);
    Destination destination = new ActiveMQQueue("connect.order");
    jmsService.sendMessage(destination, message);
    CommonUtil.writeSuccessMap(jsonMap, "消息发送成功");
    return jsonMap;
}

@GetMapping("commonOrder")
public Map<String, Object> common(String message) {
    Map<String, Object> jsonMap = new HashMap<>(16);
    jmsService.sendMessage(message);
    CommonUtil.writeSuccessMap(jsonMap, "消息发送成功");
    return jsonMap;
}

@GetMapping("topic")
public Map<String, Object> topic(String message) {
    Map<String, Object> jsonMap = new HashMap<>(16);
    jmsService.publish(message);
    CommonUtil.writeSuccessMap(jsonMap, "订阅消息发送成功");
    return jsonMap;
}