這期內(nèi)容當中小編將會給大家?guī)碛嘘P怎么在Spring Boot中利用ActiveMQ實現(xiàn)一個延遲消息，文章內(nèi)容豐富且以專業(yè)的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

創(chuàng)新互聯(lián)是一家集網(wǎng)站建設,陽西企業(yè)網(wǎng)站建設,陽西品牌網(wǎng)站建設,網(wǎng)站定制,陽西網(wǎng)站建設報價,網(wǎng)絡營銷,網(wǎng)絡優(yōu)化,陽西網(wǎng)站推廣為一體的創(chuàng)新建站企業(yè)，幫助傳統(tǒng)企業(yè)提升企業(yè)形象加強企業(yè)競爭力。可充分滿足這一群體相比中小企業(yè)更為豐富、高端、多元的互聯(lián)網(wǎng)需求。同時我們時刻保持專業(yè)、時尚、前沿，時刻以成就客戶成長自我，堅持不斷學習、思考、沉淀、凈化自己，讓我們?yōu)楦嗟钠髽I(yè)打造出實用型網(wǎng)站。

activemq.bat啟動：

wrapper | --> Wrapper Started as Console
wrapper | Launching a JVM...
jvm 1 | Wrapper (Version 3.2.3) http://wrapper.tanukisoftware.org
jvm 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_181
jvm 1 | Heap sizes: current=125952k free=115299k max=932352k
jvm 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=1
jvm 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: ..\..\conf
jvm 1 | ACTIVEMQ_DATA: ..\..\data
jvm 1 | Loading message broker from: xbean:activemq.xml
jvm 1 | INFO | Refreshing org.apache.activemq.xbean.XBeanBrokerFactory$1@f0ef68d: startup date [Fri May 24 15:16:21 CST 2019]; root of context hierarchy
jvm 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] started
jvm 1 | INFO | Apache ActiveMQ 5.15.9 (localhost, ID:wulf00-51057-1558682182909-0:1) is starting
jvm 1 | INFO | Listening for connections at: tcp://wulf00:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600
jvm 1 | INFO | Connector openwire started
jvm 1 | INFO | Listening for connections at: amqp://wulf00:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600
jvm 1 | INFO | Connector amqp started
jvm 1 | INFO | Listening for connections at: stomp://wulf00:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600
jvm 1 | INFO | Connector stomp started
jvm 1 | INFO | Listening for connections at: mqtt://wulf00:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600
jvm 1 | INFO | Connector mqtt started
jvm 1 | INFO | Starting Jetty server
jvm 1 | INFO | Creating Jetty connector
jvm 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=104857600
jvm 1 | INFO | Connector ws started
jvm 1 | INFO | Apache ActiveMQ 5.15.9 (localhost, ID:wulf00-51057-1558682182909-0:1) started
jvm 1 | INFO | For help or more information please see: http://activemq.apache.org
jvm 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 mb
jvm 1 | INFO | No Spring WebApplicationInitializer types detected on classpath
jvm 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 classpath
jvm 1 | INFO | jolokia-agent: Using policy access restrictor classpath:/jolokia-access.xml

默認端口8161，訪問下http://localhost:8161/admin，用戶名密碼都是admin，進入控制臺頁面：

我們用坐上方的Queues來創(chuàng)建一個叫vboxlog的隊列：

 

二、修改activeMQ配置文件

broker新增配置信息 schedulerSupport="true"

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" schedulerSupport="true" >

 <destinationPolicy>
 <policyMap>
 <policyEntries>
 <policyEntry topic=">" >
 <!-- The constantPendingMessageLimitStrategy is used to prevent
 slow topic consumers to block producers and affect other consumers
 by limiting the number of messages that are retained
 For more information, see:

 http://activemq.apache.org/slow-consumer-handling.html

 -->
 <pendingMessageLimitStrategy>
 <constantPendingMessageLimitStrategy limit="1000"/>
 </pendingMessageLimitStrategy>
 </policyEntry>
 </policyEntries>
 </policyMap>
 </destinationPolicy>

三、創(chuàng)建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
 */
@Configuration
public class ActiveMqConfig {

 @Bean
 public ActiveMQConnectionFactory factory(@Value("${spring.activemq.broker-url}") String url){
 ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);
 // 設置信任序列化包集合
 List<String> models = new ArrayList<>();
 models.add("com.example.demoactivemq.domain");
 factory.setTrustedPackages(models);

 return factory;
 }

}

消息實體類

package com.example.demoactivemq.domain;

import lombok.Builder;
import lombok.Data;

import java.io.Serializable;

/**
 * @author shanks on 2019-11-12
 */

@Builder
@Data
public class MessageModel implements Serializable {
 private String titile;
 private String message;
}

生產(chǎn)者

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;


/**
 * 消息生產(chǎn)者
 *
 * @author shanks
 */
@Service
@Slf4j
public class Producer {

 public static final Destination DEFAULT_QUEUE = new ActiveMQQueue("delay.queue");

 @Autowired
 private JmsMessagingTemplate template;

 /**
 * 發(fā)送消息
 *
 * @param destination destination是發(fā)送到的隊列
 * @param message message是待發(fā)送的消息
 */
 public <T extends Serializable> void send(Destination destination, T message) {
 template.convertAndSend(destination, message);
 }

 /**
 * 延時發(fā)送
 *
 * @param destination 發(fā)送的隊列
 * @param data 發(fā)送的消息
 * @param time 延遲時間
 */
 public <T extends Serializable> void delaySend(Destination destination, T data, Long time) {
 Connection connection = null;
 Session session = null;
 MessageProducer producer = null;
 // 獲取連接工廠
 ConnectionFactory connectionFactory = template.getConnectionFactory();
 try {
 // 獲取連接
 connection = connectionFactory.createConnection();
 connection.start();
 // 獲取session，true開啟事務，false關閉事務
 session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
 // 創(chuàng)建一個消息隊列
 producer = session.createProducer(destination);
 producer.setDeliveryMode(JmsProperties.DeliveryMode.PERSISTENT.getValue());
 ObjectMessage message = session.createObjectMessage(data);
 //設置延遲時間
 message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, time);
 // 發(fā)送消息
 producer.send(message);
 log.info("發(fā)送消息：{}", data);
 session.commit();
 } catch (Exception e) {
 e.printStackTrace();
 } finally {
 try {
 if (producer != null) {
 producer.close();
 }
 if (session != null) {
 session.close();
 }
 if (connection != null) {
 connection.close();
 }
 } catch (Exception e) {
 e.printStackTrace();
 }
 }
 }
}

消費者

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
@Slf4j
public 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 {

 /**
 * 消息生產(chǎn)者
 */
 @Autowired
 private Producer producer;

 /**
 * 及時消息隊列測試
 */
 @Test
 public void test() {
 MessageModel messageModel = MessageModel.builder()
 .message("測試消息")
 .titile("消息000")
 .build();
 // 發(fā)送消息
 producer.send(Producer.DEFAULT_QUEUE, messageModel);
 }

 /**
 * 延時消息隊列測試
 */
 @Test
 public void test2() {
 for (int i = 0; i < 5; i++) {
 MessageModel messageModel = MessageModel.builder()
 .titile("延遲10秒執(zhí)行")
 .message("測試消息" + i)
 .build();
 // 發(fā)送延遲消息
 producer.delaySend(Producer.DEFAULT_QUEUE, messageModel, 10000L);
 }
 try {
 // 休眠100秒，等等消息執(zhí)行
 Thread.currentThread().sleep(100000L);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }
}

執(zhí)行結果

2019-11-12 22:18:52.939  INFO 17263 --- [           main] c.e.demoactivemq.producer.Producer       : 發(fā)送消息：MessageModel(titile=延遲10秒執(zhí)行, message=測試消息0)
2019-11-12 22:18:52.953  INFO 17263 --- [           main] c.e.demoactivemq.producer.Producer       : 發(fā)送消息：MessageModel(titile=延遲10秒執(zhí)行, message=測試消息1)
2019-11-12 22:18:52.958  INFO 17263 --- [           main] c.e.demoactivemq.producer.Producer       : 發(fā)送消息：MessageModel(titile=延遲10秒執(zhí)行, message=測試消息2)
2019-11-12 22:18:52.964  INFO 17263 --- [           main] c.e.demoactivemq.producer.Producer       : 發(fā)送消息：MessageModel(titile=延遲10秒執(zhí)行, message=測試消息3)
2019-11-12 22:18:52.970  INFO 17263 --- [           main] c.e.demoactivemq.producer.Producer       : 發(fā)送消息：MessageModel(titile=延遲10秒執(zhí)行, message=測試消息4)
2019-11-12 22:19:03.012  INFO 17263 --- [enerContainer-1] c.e.demoactivemq.producer.Consumer       : 收到消息:MessageModel(titile=延遲10秒執(zhí)行, message=測試消息0)
2019-11-12 22:19:03.017  INFO 17263 --- [enerContainer-1] c.e.demoactivemq.producer.Consumer       : 收到消息:MessageModel(titile=延遲10秒執(zhí)行, message=測試消息1)
2019-11-12 22:19:03.019  INFO 17263 --- [enerContainer-1] c.e.demoactivemq.producer.Consumer       : 收到消息:MessageModel(titile=延遲10秒執(zhí)行, message=測試消息2)
2019-11-12 22:19:03.020  INFO 17263 --- [enerContainer-1] c.e.demoactivemq.producer.Consumer       : 收到消息:MessageModel(titile=延遲10秒執(zhí)行, message=測試消息3)
2019-11-12 22:19:03.021  INFO 17263 --- [enerContainer-1] c.e.demoactivemq.producer.Consumer       : 收到消息:MessageModel(titile=延遲10秒執(zhí)行, message=測試消息4)

上述就是小編為大家分享的怎么在Spring Boot中利用ActiveMQ實現(xiàn)一個延遲消息了，如果剛好有類似的疑惑，不妨參照上述分析進行理解。如果想知道更多相關知識，歡迎關注創(chuàng)新互聯(lián)行業(yè)資訊頻道。

網(wǎng)站欄目：怎么在SpringBoot中利用ActiveMQ實現(xiàn)一個延遲消息
文章源于：http://jinyejixie.com/article28/jjhdcp.html

成都網(wǎng)站建設公司_創(chuàng)新互聯(lián)，為您提供軟件開發(fā)、網(wǎng)站排名、云服務器、網(wǎng)站制作、品牌網(wǎng)站制作、

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉載內(nèi)容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉載，或轉載時需注明來源： 創(chuàng)新互聯(lián)