springboot整合activemq ( mq )

pom

  <parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.1.3.RELEASE</version>
	</parent>
  <dependencies>
  	<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		
		<dependency>
		    <groupId>org.springframework.boot</groupId>
		    <artifactId>spring-boot-starter-activemq</artifactId>
		</dependency>
  </dependencies>

application.yml配置文件

server:
  port: 8081
spring: 
  activemq:
    #ActiveMQ通讯地址
    broker-url: tcp://192.168.166.139:61616
    #用户名
    user: admin
    #密码
    password: admin
    #是否启用内存模式（就是不安装MQ，项目启动时同时启动一个MQ实例）
    in-memory: false
    packages:
      #信任所有的包
      trust-all: true
    pool:
      #是否替换默认的连接池，使用ActiveMQ的连接池需引入的依赖
      enabled: false

activemq配置类

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

import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerContainerFactory;

@Configuration
@EnableJms
public class ActiveMQConfig {

    //springboot默认只配置queue类型消息，如果要使用topic类型的消息，则需要配置该bean
    @Bean
    public JmsListenerContainerFactory<?> jmsTopicListenerContainerFactory(ConnectionFactory connectionFactory){
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        //这里必须设置为true，false则表示是queue类型
        factory.setPubSubDomain(true);
        return factory;
    }
    
    @Bean("Queue")
    public Queue queue() {
        return new ActiveMQQueue("springboot.queue") ;
    }
    
    @Bean("Queue2")
    public Queue queue2() {
        return new ActiveMQQueue("springboot.queue2") ;
    }

    @Bean("Topic")
    public Topic topic() {
        return new ActiveMQTopic("springboot.topic") ;
    }
}

生产者

import java.util.ArrayList;
import java.util.List;

import javax.annotation.Resource;
import javax.jms.Queue;
import javax.jms.Topic;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class Producer {
    @Autowired
    private JmsMessagingTemplate jmsTemplate;

    @Resource(name = "Queue")
    private Queue queue;
    
    @Resource(name = "Queue2")
    private Queue queue2;
    
    @Resource(name = "Topic")
    private Topic topic;

    //发送queue类型消息
    @GetMapping("/queue")
    public void sendQueueMsg(String msg){
        jmsTemplate.convertAndSend(queue, msg);
    }
    
    //发送queue2类型消息(对象)
    @GetMapping("/queue2")
    public void sendQueueMsg2(String msg){
    	List<String> list=new ArrayList<String>();
    	list.add("123");
    	list.add(msg);
        jmsTemplate.convertAndSend(queue2, list);
    }
    

    //发送topic类型消息
    @GetMapping("/topic")
    public void sendTopicMsg(String msg){
        jmsTemplate.convertAndSend(topic, msg);
    }
}

消费者：

import java.util.List;

import javax.jms.JMSException;
import javax.jms.TextMessage;

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

@Component
public class Consumer {

    //接收queue类型消息
    //destination对应配置类中ActiveMQQueue("springboot.queue")设置的名字
    @JmsListener(destination="springboot.queue")
    public void ListenQueue(final TextMessage text) throws JMSException{
        System.out.println("接收到queue消息：" + text.getText());
    }
    
    //接收queue类型消息
    //destination对应配置类中ActiveMQQueue("springboot.queue2")设置的名字
	@JmsListener(destination="springboot.queue2")
    public void ListenQueue2(final List<String> list) throws JMSException{
        System.out.println("接收到queue2消息：" + list);
    }
	
    //接收topic类型消息
    //destination对应配置类中ActiveMQTopic("springboot.topic")设置的名字
    //containerFactory对应配置类中注册JmsListenerContainerFactory的bean名称
    @JmsListener(destination="springboot.topic", containerFactory = "jmsTopicListenerContainerFactory")
    public void ListenTopic(final TextMessage text) throws JMSException{
        System.out.println("接收到topic消息：" + text.getText());
    }
}