SpringBoot中怎么利用ActiveMQ实现延迟消息
这篇文章将为大家详细讲解有关SpringBoot中怎么利用ActiveMQ实现延迟消息,文章内容质量较高,因此小编分享给大家做个参考,希望大家阅读完这篇文章后对相关知识有一定的了解。
让客户满意是我们工作的目标,不断超越客户的期望值来自于我们对这个行业的热爱。我们立志把好的技术通过有效、简单的方式提供给客户,将通过不懈努力成为客户在信息化领域值得信任、有价值的长期合作伙伴,公司提供的服务项目有:域名注册、网络空间、营销软件、网站建设、前郭网站维护、网站推广。
一、安装activeMQ
Windows下安装ActiveMQ:
到官网(http://activemq.apache.org/download-archives.html)下载最新发布的压缩包(我下的是5.15.9)到本地后解压(我解压到D盘Dev目录下)即可。进入解压后的bin目录,我是64位机器,再进入win64目录后,双击activemq.bat启动:
wrapper | --> Wrapper Started as Consolewrapper | Launching a JVM...jvm 1 | Wrapper (Version 3.2.3) http://wrapper.tanukisoftware.orgjvm 1 | Copyright 1999-2006 Tanuki Software, Inc. All Rights Reserved.jvm 1 |jvm 1 | Java Runtime: Oracle Corporation 1.8.0_181 C:\Program Files\Java\jre1.8.0_181jvm 1 | Heap sizes: current=125952k free=115299k max=932352kjvm 1 | JVM args: -Dactivemq.home=../.. -Dactivemq.base=../.. -Djavax.net.ssl.keyStorePassword=password -Djavax.net.ssl.trustStorePassword=password -Djavax.net.ssl.keyStore=../../conf/broker.ks -Djavax.net.ssl.trustStore=../../conf/broker.ts -Dcom.sun.management.jmxremote -Dorg.apache.activemq.UseDedicatedTaskRunner=true -Djava.util.logging.config.file=logging.properties -Dactivemq.conf=../../conf -Dactivemq.data=../../data -Djava.security.auth.login.config=../../conf/login.config -Xmx1024m -Djava.library.path=../../bin/win64 -Dwrapper.key=mChNCWMZ2FoXhZ9g -Dwrapper.port=32000 -Dwrapper.jvm.port.min=31000 -Dwrapper.jvm.port.max=31999 -Dwrapper.pid=3500 -Dwrapper.version=3.2.3 -Dwrapper.native_library=wrapper -Dwrapper.cpu.timeout=10 -Dwrapper.jvmid=1jvm 1 | Extensions classpath:jvm 1 | [..\..\lib,..\..\lib\camel,..\..\lib\optional,..\..\lib\web,..\..\lib\extra]jvm 1 | ACTIVEMQ_HOME: ..\..jvm 1 | ACTIVEMQ_BASE: ..\..jvm 1 | ACTIVEMQ_CONF: ..\..\confjvm 1 | ACTIVEMQ_DATA: ..\..\datajvm 1 | Loading message broker from: xbean:activemq.xmljvm 1 | INFO | Refreshing org.apache.activemq.xbean.XBeanBrokerFactory$1@f0ef68d: startup date [Fri May 24 15:16:21 CST 2019]; root of context hierarchyjvm 1 | INFO | Using Persistence Adapter: KahaDBPersistenceAdapter[D:\Dev\apache-activemq-5.15.9\bin\win64\..\..\data\kahadb]jvm 1 | INFO | PListStore:[D:\Dev\apache-activemq-5.15.9\bin\win64\..\..\data\localhost\tmp_storage] startedjvm 1 | INFO | Apache ActiveMQ 5.15.9 (localhost, ID:wulf00-51057-1558682182909-0:1) is startingjvm 1 | INFO | Listening for connections at: tcp://wulf00:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600jvm 1 | INFO | Connector openwire startedjvm 1 | INFO | Listening for connections at: amqp://wulf00:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600jvm 1 | INFO | Connector amqp startedjvm 1 | INFO | Listening for connections at: stomp://wulf00:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600jvm 1 | INFO | Connector stomp startedjvm 1 | INFO | Listening for connections at: mqtt://wulf00:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600jvm 1 | INFO | Connector mqtt startedjvm 1 | INFO | Starting Jetty serverjvm 1 | INFO | Creating Jetty connectorjvm 1 | WARN | ServletContext@o.e.j.s.ServletContextHandler@17bc7c8a{/,null,STARTING} has uncovered http methods for path: /jvm 1 | INFO | Listening for connections at ws://wulf00:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600jvm 1 | INFO | Connector ws startedjvm 1 | INFO | Apache ActiveMQ 5.15.9 (localhost, ID:wulf00-51057-1558682182909-0:1) startedjvm 1 | INFO | For help or more information please see: http://activemq.apache.orgjvm 1 | WARN | Store limit is 102400 mb (current store usage is 0 mb). The data directory: D:\Dev\apache-activemq-5.15.9\bin\win64\..\..\data\kahadb only has 92649 mb of usable space. - resetting to maximum available disk space: 92649 mbjvm 1 | INFO | No Spring WebApplicationInitializer types detected on classpathjvm 1 | INFO | ActiveMQ WebConsole available at http://0.0.0.0:8161/jvm 1 | INFO | ActiveMQ Jolokia REST API available at http://0.0.0.0:8161/api/jolokia/jvm 1 | INFO | Initializing Spring FrameworkServlet 'dispatcher'jvm 1 | INFO | No Spring WebApplicationInitializer types detected on classpathjvm 1 | INFO | jolokia-agent: Using policy access restrictor classpath:/jolokia-access.xml
默认端口8161,访问下http://localhost:8161/admin,用户名密码都是admin,进入控制台页面:
我们用坐上方的Queues来创建一个叫vboxlog的队列:
二、修改activeMQ配置文件
broker新增配置信息 schedulerSupport="true"
三、创建SpringBoot工程
1、配置ActiveMQ工厂信息,信任包必须配置否则会报错
package com.example.demoactivemq.config;import org.apache.activemq.ActiveMQConnectionFactory;import org.apache.activemq.RedeliveryPolicy;import org.springframework.beans.factory.annotation.Value;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import java.util.ArrayList;import java.util.List;/** * @author shanks on 2019-11-12 */@Configurationpublic class ActiveMqConfig { @Bean public ActiveMQConnectionFactory factory(@Value("${spring.activemq.broker-url}") String url){ ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url); // 设置信任序列化包集合 List
消息实体类
package com.example.demoactivemq.domain;import lombok.Builder;import lombok.Data;import java.io.Serializable;/** * @author shanks on 2019-11-12 */@Builder@Datapublic class MessageModel implements Serializable { private String titile; private String message;}
生产者
package com.example.demoactivemq.producer;import lombok.extern.slf4j.Slf4j;import org.apache.activemq.ScheduledMessage;import org.apache.activemq.command.ActiveMQQueue;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.autoconfigure.jms.JmsProperties;import org.springframework.jms.core.JmsMessagingTemplate;import org.springframework.stereotype.Service;import javax.jms.*;import java.io.Serializable;/** * 消息生产者 * * @author shanks */@Service@Slf4jpublic class Producer { public static final Destination DEFAULT_QUEUE = new ActiveMQQueue("delay.queue"); @Autowired private JmsMessagingTemplate template; /** * 发送消息 * * @param destination destination是发送到的队列 * @param message message是待发送的消息 */ public
消费者
package com.example.demoactivemq.producer;import com.example.demoactivemq.domain.MessageModel;import lombok.extern.slf4j.Slf4j;import org.springframework.jms.annotation.JmsListener;import org.springframework.stereotype.Component;/** * 消费者 */@Component@Slf4jpublic class Consumer { @JmsListener(destination = "delay.queue") public void receiveQueue(MessageModel message) { log.info("收到消息:{}", message); }}
application.yml
spring: activemq: broker-url: tcp://localhost:61616
测试类
package com.example.demoactivemq;import com.example.demoactivemq.domain.MessageModel;import com.example.demoactivemq.producer.Producer;import org.junit.jupiter.api.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;@SpringBootTest(classes = DemoActivemqApplication.class)@RunWith(SpringRunner.class)class DemoActivemqApplicationTests { /** * 消息生产者 */ @Autowired private Producer producer; /** * 及时消息队列测试 */ @Test public void test() { MessageModel messageModel = MessageModel.builder() .message("测试消息") .titile("消息000") .build(); // 发送消息 producer.send(Producer.DEFAULT_QUEUE, messageModel); } /** * 延时消息队列测试 */ @Test public void test2() { for (int i = 0; i < 5; i++) { MessageModel messageModel = MessageModel.builder() .titile("延迟10秒执行") .message("测试消息" + i) .build(); // 发送延迟消息 producer.delaySend(Producer.DEFAULT_QUEUE, messageModel, 10000L); } try { // 休眠100秒,等等消息执行 Thread.currentThread().sleep(100000L); } catch (InterruptedException e) { e.printStackTrace(); } }}
执行结果
2019-11-12 22:18:52.939 INFO 17263 --- [ main] c.e.demoactivemq.producer.Producer : 发送消息:MessageModel(titile=延迟10秒执行, message=测试消息0)2019-11-12 22:18:52.953 INFO 17263 --- [ main] c.e.demoactivemq.producer.Producer : 发送消息:MessageModel(titile=延迟10秒执行, message=测试消息1)2019-11-12 22:18:52.958 INFO 17263 --- [ main] c.e.demoactivemq.producer.Producer : 发送消息:MessageModel(titile=延迟10秒执行, message=测试消息2)2019-11-12 22:18:52.964 INFO 17263 --- [ main] c.e.demoactivemq.producer.Producer : 发送消息:MessageModel(titile=延迟10秒执行, message=测试消息3)2019-11-12 22:18:52.970 INFO 17263 --- [ main] c.e.demoactivemq.producer.Producer : 发送消息:MessageModel(titile=延迟10秒执行, message=测试消息4)2019-11-12 22:19:03.012 INFO 17263 --- [enerContainer-1] c.e.demoactivemq.producer.Consumer : 收到消息:MessageModel(titile=延迟10秒执行, message=测试消息0)2019-11-12 22:19:03.017 INFO 17263 --- [enerContainer-1] c.e.demoactivemq.producer.Consumer : 收到消息:MessageModel(titile=延迟10秒执行, message=测试消息1)2019-11-12 22:19:03.019 INFO 17263 --- [enerContainer-1] c.e.demoactivemq.producer.Consumer : 收到消息:MessageModel(titile=延迟10秒执行, message=测试消息2)2019-11-12 22:19:03.020 INFO 17263 --- [enerContainer-1] c.e.demoactivemq.producer.Consumer : 收到消息:MessageModel(titile=延迟10秒执行, message=测试消息3)2019-11-12 22:19:03.021 INFO 17263 --- [enerContainer-1] c.e.demoactivemq.producer.Consumer : 收到消息:MessageModel(titile=延迟10秒执行, message=测试消息4)
关于SpringBoot中怎么利用ActiveMQ实现延迟消息就分享到这里了,希望以上内容可以对大家有一定的帮助,可以学到更多知识。如果觉得文章不错,可以把它分享出去让更多的人看到。
网页名称:SpringBoot中怎么利用ActiveMQ实现延迟消息
文章来源:http://pwwzsj.com/article/jopjde.html