SpringBoot整合ActiveMQ

1         <dependency>
2             <groupId>org.springframework.boot</groupId>
3             <artifactId>spring-boot-starter-activemq</artifactId>
4         </dependency>

1 spring.activemq.broker-url=tcp://192.168.205.128:61616
2 spring.activemq.user=admin
3 spring.activemq.password=admin

 1 package cn.coreqi.config;
 2 
 3 import cn.coreqi.entities.User;
 4 import org.apache.activemq.command.ActiveMQQueue;
 5 import org.apache.activemq.command.ActiveMQTopic;
 6 import org.springframework.context.annotation.Bean;
 7 import org.springframework.context.annotation.Configuration;
 8 import org.springframework.jms.support.converter.MappingJackson2MessageConverter;
 9 import org.springframework.jms.support.converter.MessageConverter;
10 import org.springframework.jms.support.converter.MessageType;
11 
12 import java.util.HashMap;
13 import java.util.Map;
14 
15 @Configuration
16 public class ActivemqConfig {
17     //声明一个队列
18     @Bean
19     public ActiveMQQueue queue(){
20         return new ActiveMQQueue("userQueue");
21     }
22     //声明一个主题
23     @Bean
24     public ActiveMQTopic topic(){
25         return new ActiveMQTopic("userTopic");
26     }
27     //使用json进行消息的序列化
28     @Bean
29     public MessageConverter messageConverter(){
30         //return new SimpleMessageConverter();  默认
31         MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
32         converter.setTargetType(MessageType.TEXT);
33         // 定义了typeId到Class的Map
34         Map<String, Class<?>> typeIdMap = new HashMap<>();
35         typeIdMap.put("User", User.class);
36         converter.setTypeIdMappings(typeIdMap);
37         // 设置发送到队列中的typeId的名称
38         converter.setTypeIdPropertyName("User");
39         converter.setEncoding("UTF-8");
40         return converter;
41     }
42 }

 1 package cn.coreqi;
 2 
 3 import org.springframework.boot.SpringApplication;
 4 import org.springframework.boot.autoconfigure.SpringBootApplication;
 5 import org.springframework.jms.annotation.EnableJms;
 6 
 7 @SpringBootApplication
 8 @EnableJms
 9 public class SpringbootactivemqApplication {
10 
11     public static void main(String[] args) {
12         SpringApplication.run(SpringbootactivemqApplication.class, args);
13     }
14 
15 }

 1     @Autowired
 2     private JmsOperations jmsOperations;
 3     public String send(){
 4 //        jmsOperations.send("userQueue", new MessageCreator() {
 5 //            @Override
 6 //            public Message createMessage(Session session) throws JMSException {
 7 //                return session.createObjectMessage(new User(1,"fanqi","fanqisoft",1));
 8 //            }
 9 //        });
10         jmsOperations.convertAndSend("userQueue",new User(2,"fanqi","fanqisoft",1));
11         return "success";
12     }
13     public User receive() throws JMSException {
14         return (User)jmsOperations.receiveAndConvert("userQueue");
15     }

 1 package cn.coreqi.service;
 2 
 3 import cn.coreqi.entities.User;
 4 import org.springframework.jms.annotation.JmsListener;
 5 import org.springframework.stereotype.Service;
 6 
 7 @Service
 8 public class UserService {
 9     @JmsListener(destination = "userQueue")
10     public void receive(User user){
11         System.out.println(user.toString());
12     }
13 }