SpringBoot整合ActiveMq

今天药忘吃喽~ 2022-05-30 10:25 459阅读 0赞

1、先下载activemq安装

从ActiveMq官方上下载ActiveMq服务
下载地址：http://activemq.apache.org/download.html

我当前下载的是版本是5.15.3 官方备注：当前最新的稳定版本。

下载下来解压后进到window相对应的版本的bin目录下执行activemq.bat

2、创建SpringBoot项目

pom.xml 

  1. <parent>
  2.         <groupId>org.springframework.boot</groupId>
  3.         <artifactId>spring-boot-starter-parent</artifactId>
  4.         <version>1.5.8.RELEASE</version>
  5.         <relativePath /> <!-- lookup parent from repository -->
  6.     </parent>
  8.     <properties>
  9.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  10.         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
  11.         <java.version>1.8</java.version>
  12.     </properties>
  14.     <dependencies>
  15.         <dependency>
  16.             <groupId>org.springframework.boot</groupId>
  17.             <artifactId>spring-boot-starter</artifactId>
  18.         </dependency>
  19.         <dependency>
  20.             <groupId>org.springframework.boot</groupId>
  21.             <artifactId>spring-boot-starter-web</artifactId>
  22.         </dependency>
  23.         <dependency>
  24.             <groupId>org.springframework.boot</groupId>
  25.             <artifactId>spring-boot-starter-activemq</artifactId>
  26.         </dependency>
  27.         <dependency>
  28.             <groupId>org.apache.activemq</groupId>
  29.             <artifactId>activemq-pool</artifactId>
  30.         </dependency>
  32.         <dependency>
  33.             <groupId>org.springframework.boot</groupId>
  34.             <artifactId>spring-boot-starter-test</artifactId>
  35.             <scope>test</scope>
  36.         </dependency>
  38.         <!-- https://mvnrepository.com/artifact/com.alibaba/fastjson -->
  39.         <dependency>
  40.             <groupId>com.alibaba</groupId>
  41.             <artifactId>fastjson</artifactId>
  42.             <version>1.2.40</version>
  43.         </dependency>
  45.         <dependency>
  46.             <groupId>org.springframework.boot</groupId>
  47.             <artifactId>spring-boot-devtools</artifactId>
  48.             <optional>true</optional>
  49.             <!-- optional=true,依赖不会传递，该项目依赖devtools；之后依赖myboot项目的项目如果想要使用devtools，需要重新引入 -->
  50.         </dependency>
  52.     </dependencies>
  54.     <build>
  55.         <plugins>
  56.             <plugin>
  57.                 <groupId>org.springframework.boot</groupId>
  58.                 <artifactId>spring-boot-maven-plugin</artifactId>
  59.             </plugin>
  60.         </plugins>
  61.     </build>

生产者：

  1. @Service("userProducer")
  2. public class UserProducer {
  4.     @Resource
  5.     private JmsMessagingTemplate jmsMessagingTemplate;
  7.     @Resource
  8.     private Queue userQueue;
  9.     @Resource
  10.     private Queue queue;
  11.     @Resource
  12.     private Topic topic;
  15.     public void sendData(Serializable obj) {
  16.         this.jmsMessagingTemplate.convertAndSend(userQueue,obj);
  17.     }
  18.     public void sendMessage(String message) {
  19.         this.jmsMessagingTemplate.convertAndSend(queue,message);
  20.     }
  22.     public void sendTopicMessage(String message) {
  23.         this.jmsMessagingTemplate.convertAndSend(topic,message);
  24.     }
  25. }

activemq配置

  1. @Configuration
  2. public class ActiveMqConfig {
  4.     @Bean
  5.     public Queue userQueue() {
  6.         return new ActiveMQQueue("userMqQueue");
  7.     }
  9.     @Bean
  10.     public ActiveMQQueue queue() {
  11.         return new ActiveMQQueue("weisg.queue");
  12.     }
  14.     @Bean
  15.     public Topic topic() {
  16.         return new ActiveMQTopic("weisg.topic");
  17.     }
  19.     // topic模式的ListenerContainer
  20.     @Bean
  21.     public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ConnectionFactory activeMQConnectionFactory) {
  22.         DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
  23.         bean.setPubSubDomain(true);
  24.         bean.setConnectionFactory(activeMQConnectionFactory);
  25.         return bean;
  26.     }
  28.     // queue模式的ListenerContainer
  29.     @Bean
  30.     public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ConnectionFactory activeMQConnectionFactory) {
  31.         DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
  32.         bean.setConnectionFactory(activeMQConnectionFactory);
  33.         return bean;
  34.     }
  35. }

以上的配置为 支持同时发送和接收queue/topic

消费者

  1. @Service
  2. public class UserConsumer {
  3.     @JmsListener(destination = "userMqQueue",containerFactory = "jmsListenerContainerQueue")
  4.     public void receive(ObjectMessage message) throws JMSException {
  5.         System.out.println("----------UserConsumer-----------"+message);
  6.         System.out.println("----------message-----------"+message.getObject());
  7.         User user = new User();
  9.         BeanUtils.copyProperties(message.getObject(), user);
  11.         System.out.println("user--------------"+user);
  12.         System.out.println("user数据数据完成...");
  13.     }
  15.     @JmsListener(destination = "weisg.topic",containerFactory = "jmsListenerContainerTopic")
  16.     public void receiveTopicMessage(String message) throws JMSException {
  17.         System.out.println("receiveTopicMessage--------------"+message);
  18.     }
  20.     @JmsListener(destination = "weisg.topic",containerFactory = "jmsListenerContainerTopic")
  21.     public void receiveTopicMessage2(String message) throws JMSException {
  22.         System.out.println("receiveTopicMessage2--------------"+message);
  23.     }
  24. }

请求方法

  1. @RequestMapping(value="/api/addUser",method = RequestMethod.POST)
  2.     public Map<String, Object> beetl(@RequestParam Map<String, Object> params){
  3.         Map<String, Object> retMap = new HashMap<String, Object>();
  4.         User user = new User();
  5.         user.setUserId((String)params.get("userId"));
  6.         user.setUserName((String)params.get("userName"));
  7.         user.setMobile((String)params.get("mobile"));
  8.         userProducer.sendData((Serializable)user);
  9.         userProducer.sendMessage(user.getUserId());
  10.         for (int i = 0; i < 10; i++) {
  11.             userProducer.sendTopicMessage((String)params.get("userId")+i);
  12.         }
  13.         retMap.put("code", "200");
  14.         retMap.put("msg", "success!");
  16.         return retMap;
  17.     }

配置文件：

  1. spring.activemq.broker-url=tcp://127.0.0.1:61616
  2. #如果此处设置为true，需要加activemq-pool的依赖包，否则会自动配置失败，报JmsMessagingTemplate注入失败
  3. spring.activemq.pool.enabled=true
  4. spring.activemq.user=admin
  5. # 密码
  6. spring.activemq.password=admin
  8. # 在考虑结束之前等待的时间
  9. spring.activemq.close-timeout=150
  10. # 默认代理URL是否应该在内存中。如果指定了显式代理，则忽略此值。
  11. spring.activemq.in-memory=false 
  12. # 是否在回滚回滚消息之前停止消息传递。这意味着当启用此命令时，消息顺序不会被保留。
  13. spring.activemq.non-blocking-redelivery=false
  15. # 等待消息发送响应的时间。设置为0等待永远。
  16. spring.activemq.send-timeout=0
  18. #发布模式，为true时是topic模式，为false是queue模式
  19. spring.jms.pub-sub-domain=true
  21. # 是否信任所有包
  22. #注意：对象传输需要开启包白名单，否则会报错
  23. spring.activemq.packages.trust-all=true
  24. # 要信任的特定包的逗号分隔列表（当不信任所有包时）
  26. #spring.activemq.packages.trusted=
  27. # 当连接请求和池满时是否阻塞。设置false会抛“JMSException异常”。
  28. #spring.activemq.pool.block-if-full=true
  29. # 如果池仍然满，则在抛出异常前阻塞时间。
  30. #spring.activemq.pool.block-if-full-timeout=-1ms
  31. # 是否在启动时创建连接。可以在启动时用于加热池。
  32. #spring.activemq.pool.create-connection-on-startup=true
  33. # 是否用Pooledconnectionfactory代替普通的ConnectionFactory。
  34. #spring.activemq.pool.enabled=false 
  35. # 连接过期超时。
  36. #spring.activemq.pool.expiry-timeout=0ms
  37. #空闲的连接过期时间，默认为30秒
  38. spring.activemq.pool.idle-timeout=30
  39. # 连接池最大连接数
  40. spring.activemq.pool.max-connections=10
  41. # 每个连接的有效会话的最大数目。
  42. #spring.activemq.pool.maximum-active-session-per-connection=500
  43. # 当有"JMSException"时尝试重新连接
  44. #spring.activemq.pool.reconnect-on-exception=true
  45. # 在空闲连接清除线程之间运行的时间。当为负数时，没有空闲连接驱逐线程运行。
  46. #spring.activemq.pool.time-between-expiration-check=-1ms
  47. # 是否只使用一个MessageProducer
  48. #spring.activemq.pool.use-anonymous-producers=true

结果：如下

  1. 2018-03-10 11:51:25.582 [http-nio-8889-exec-4] INFO  org.springframework.web.servlet.DispatcherServlet - FrameworkServlet 'dispatcherServlet': initialization completed in 20 ms
  2. ----------UserConsumer-----------ActiveMQObjectMessage {commandId = 5, responseRequired = true, messageId = ID:admin-PC-51016-1520653879572-1:4:1:1:1, originalDestination = null, originalTransactionId = null, producerId = ID:admin-PC-51016-1520653879572-1:4:1:1, destination = queue://userMqQueue, transactionId = null, expiration = 0, timestamp = 1520653885674, arrival = 0, brokerInTime = 1520653885675, brokerOutTime = 1520653885677, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@6df8e57e, marshalledProperties = org.apache.activemq.util.ByteSequence@4306219a, dataStructure = null, redeliveryCounter = 0, size = 0, properties = {timestamp=1520653885637}, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false}
  3. ----------message-----------User [userId=zhangsan001, userName=张三, userNo=null, mobile=15819295938]
  4. user--------------User [userId=zhangsan001, userName=张三, userNo=null, mobile=15819295938]
  5. user数据数据完成...
  6. receiveTopicMessage--------------zhangsan0010
  7. receiveTopicMessage2--------------zhangsan0010
  8. receiveTopicMessage2--------------zhangsan0011
  9. receiveTopicMessage--------------zhangsan0011
  10. receiveTopicMessage--------------zhangsan0012
  11. receiveTopicMessage2--------------zhangsan0012
  12. receiveTopicMessage--------------zhangsan0013
  13. receiveTopicMessage2--------------zhangsan0013
  14. receiveTopicMessage--------------zhangsan0014
  15. receiveTopicMessage--------------zhangsan0015
  16. receiveTopicMessage2--------------zhangsan0014
  17. receiveTopicMessage2--------------zhangsan0015
  18. receiveTopicMessage--------------zhangsan0016
  19. receiveTopicMessage--------------zhangsan0017
  20. receiveTopicMessage2--------------zhangsan0016
  21. receiveTopicMessage2--------------zhangsan0017
  22. receiveTopicMessage--------------zhangsan0018
  23. receiveTopicMessage2--------------zhangsan0018
  24. receiveTopicMessage--------------zhangsan0019
  25. receiveTopicMessage2--------------zhangsan0019

发表评论

表情：
评论列表 （有 0 条评论，459人围观）

还没有评论，来说两句吧...

相关阅读

    相关 SpringBoot整合ActiveMQ

    前言   ActiveMQ 一个成熟的消息中间件，作用于系统之间的通信，降低模块与模块之间的耦合度。   消息的传递有两种类型： 1. Queue 队列模式：一个生

    红太狼红太狼/ 0 赞/ 431 阅读