springboot整合Activemq,queue和topic两种消息模式

快来打我* 2023-07-05 14:52 12阅读 0赞

Spring Boot为ActiveMQ配置提供了相关的"Starter",因此整合非常容易。添加ActiveMQ依赖,如下：

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-activemq</artifactId>
    </dependency>
     <dependency>
              <groupId>org.messaginghub</groupId>
              <artifactId>pooled-jms</artifactId>
     </dependency>

然后在applications.properties中进行连接配置，要注意的是发布订阅模式要加上spring.jms.pub-sub-domain=true，点对点与发布订阅不能同时生效，默认是点对点queue。

spring.activemq.broker-url=tcp://localhost:61616
    #配置信任所有包，这个配置是为了支持发送对象消息
    spring.activemq.packages.trust-all=true
    #如果是发布订阅模式要加上这个否则订阅不生效，默认是queue
    #spring.jms.pub-sub-domain=true
    #true 表示使用内置的MQ，false则连接服务器
    spring.activemq.in-memory=false
    #true表示使用连接池；false时，每发送一条数据创建一个连接
    spring.activemq.pool.enabled=true
    #连接池最大连接数
    spring.activemq.pool.max-connections=10
    #空闲的连接过期时间，默认为30秒
    spring.activemq.pool.idle-timeout=30000
    #强制的连接过期时间，与idleTimeout的区别在于：idleTimeout是在连接空闲一段时间失效，而expiryTimeout不管当前连接的情况，只要达到指定时间就失效。默认为0，never

创建消息队列Bean

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;
    
    @Configuration
    public class activemqConfig {
        /**
         * 点对点
         * @return
         */
        @Bean
        Queue queue() {
            return new ActiveMQQueue("ActiveMQQueue");
        }
    
        /**
         * 发布/订阅
         * @return
         */
        @Bean
        Topic topic(){
            return new ActiveMQTopic("ActiveMQTopic");
        }
    }

创建自定义消息体实体类

import java.io.Serializable;
    import java.util.Date;
    
    public class Message implements Serializable {
        private String content;
        private Date date;
    
        public String getContent() {
            return content;
        }
    
        public void setContent(String content) {
            this.content = content;
        }
    
        public Date getDate() {
            return date;
        }
    
        public void setDate(Date date) {
            this.date = date;
        }
    }

创建生产者

import javax.jms.Queue;
    import javax.jms.Topic;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.stereotype.Component;
    
    @Component
    public class Provider {
    
        @Autowired
        Queue queue;
        @Autowired
        Topic topic;
        @Autowired
        JmsMessagingTemplate jmsMessagingTemplate;
    
        public void sendQueue(Message message) {
    
            jmsMessagingTemplate.convertAndSend(this.queue,message);
        }
    
    
        public void sendTopic(Message message) {
            jmsMessagingTemplate.convertAndSend(this.topic,message);
        }
    }

创建消费者,@SendTo注解可以收到消费者反馈的信息,返回给监听的"return-queue"队列，要注意的是订阅模式@SendTo不起作用。

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.annotation.JmsListener;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.messaging.handler.annotation.SendTo;
    import org.springframework.stereotype.Component;
    
    @Component
    public class Consumer {
        @Autowired
        JmsMessagingTemplate jmsMessagingTemplate;
        // 使用JmsListener配置消费者监听的队列
        @JmsListener(destination="ActiveMQQueue")//监听点对点队列
        @SendTo("return-queue")
        public String handleMessageQueue1(Message message) {
            System.out.println("成功接受ActiveMQQueue2" + message.getContent());
            return "成功接受ActiveMQQueue1";
        }
        @JmsListener(destination="ActiveMQQueue")//监听点对点队列 @SendTo返回
        @SendTo("return-queue")
        public String handleMessageQueue2(Message message) {
            System.out.println("成功接受ActiveMQQueue2" + message.getContent());
            return "成功接受ActiveMQQueue2";
        }
     //监听订阅队列，订阅模式 @SendTo无作用，不需要加 @SendTo
        @JmsListener(destination="ActiveMQTopic")
        public void handleMessageTopic1(Message message) {
            System.out.println("成功接受ActiveMQTopic1" + message.getContent());
        }
        @JmsListener(destination="ActiveMQTopic")
        public void handleMessageTopic2(Message message) {
            System.out.println("成功接受ActiveMQTopic2" + message.getContent());
        }
    
        @JmsListener(destination = "return-queue")
        public void Message(String message){
            System.out.println("Product收到:"+message);
        }
    }

编写测试类

import com.alibaba.fastjson.JSONObject;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    import java.util.Date;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class test {
        @Autowired
        Provider JmsProvider;
        @Test
        public void content(){
            Message message=new Message();
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("email", "wangsheng@163.com");
            jsonObject.put("timestamp", System.currentTimeMillis());
            String jsonString = jsonObject.toJSONString();
            message.setContent(jsonString);
            message.setDate(new Date());
           JmsProvider.sendQueue(message);
           JmsProvider.sendTopic(message);
        }
    }

运行测试

![20200216114412811.png][]

[20200216114412811.png]: https://img-blog.csdnimg.cn/20200216114412811.png