Spring Boot教程のActiveMQを利用して遅延メッセージを実現します。
一、activemQを取り付ける
Linux環境ActiveMQ配置方法:https://www.jb51.net/article/162320.htm
インストールの手順は上記の文章を参照してください。ここでは紹介しません。
Windows下にActiveMQをインストールする:
公式サイト(http://activemq.apache.org/download-archives.html)に最新発表の圧縮パッケージ(私が作ったのは5.15.9です。)をダウンロードして現地に行って解凍してください。解圧後のbinディレクトリに入ります。64のマシンです。またwin 64のディレクトリに入ったら、activemq.batをダブルクリックして起動します。
私たちは上の席に座って、vbox logsという列を作ります。
二、activeMQプロファイルを修正する
brook新規配置情報scheduler Support="true"
1、ActiveMQ工場情報を配置し、信頼カバンは必ず配置しなければエラーが発生する。
2019-11-12 22:18:52.39 INFO 17263--[ main]c.e.demoactivemq.producer.Producer : メッセージの送信:Message Model(titile=10秒の遅延、message=テストメッセージ0)
2019-11-12 22:18:52.953 INFO 17263--[ main]c.e.demoactivemq.producer.Producer : メッセージ送信:Message Model(tile=10秒遅延実行、message=テストメッセージ1)
2019-11-12 22:18:52.958 INFO 17263--[ main]c.e.demoactivemq.producer.Producer : メッセージの送信:Message Model(titile=10秒の遅延、message=テストメッセージ2)
2019-11-12 22:18:52.944 INFO 17263--[ main]c.e.demoactivemq.producer.Producer : メッセージの送信:Message Model(titile=10秒遅延実行、message=テストメッセージ3)
2019-11-12 22:18:52.970 INFO 17263--[ main]c.e.demoactivemq.producer.Producer : メッセージの送信:Message Model(titile=10秒の遅延、message=テストメッセージ4)
2019-11-12 22:19:03.012 INFO 17263-[enerConttainer-1]c.e.demoactivemq.producer.Continmer : メッセージ受信:Message Model(titile=10秒遅延実行、message=テストメッセージ0)
2019-11-12 22:19:03.017 INFO 17263-[enerConttainer-1]c.e.demoactivemq.producer.Continmer : メッセージ受信:Message Model(titile=10秒遅延実行、message=テストメッセージ1)
2019-11-12 22:19:03.019 INFO 17263-[enerConttainer-1]c.e.demoactivemq.producer.Continmer : メッセージ受信:Message Model(tile=10秒遅延実行、message=テストメッセージ2)
2019-11-12 22:19:03.020 INFO 17263-[enerConttainer-1]c.e.demoactivemq.producer.Continmer : 受信メッセージ:Message Model(titile=10秒遅延実行、message=テストメッセージ3)
2019-11-12 22:19:03.021 INFO 17263-[enerConttainer-1]c.e.demoactivemq.producer.Continmer : メッセージ受信:Message Model(titile=10秒遅延実行、message=テストメッセージ4)
あなたより優秀な人はあなたより努力しています。奮闘しない資格がありますか?
締め括りをつける
以上はこの文章の全部の内容です。本文の内容は皆さんの学習や仕事に対して一定の参考学習価値を持ってほしいです。ありがとうございます。
Linux環境ActiveMQ配置方法:https://www.jb51.net/article/162320.htm
インストールの手順は上記の文章を参照してください。ここでは紹介しません。
Windows下にActiveMQをインストールする:
公式サイト(http://activemq.apache.org/download-archives.html)に最新発表の圧縮パッケージ(私が作ったのは5.15.9です。)をダウンロードして現地に行って解凍してください。解圧後のbinディレクトリに入ります。64のマシンです。またwin 64のディレクトリに入ったら、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 | [email protected]@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で、コンソールページに入ります。私たちは上の席に座って、vbox logsという列を作ります。
二、activeMQプロファイルを修正する
brook新規配置情報scheduler Support="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>
三、SprigBootプロジェクトの創建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;
}
生産者
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
@Slf4j
public class Producer {
public static final Destination DEFAULT_QUEUE = new ActiveMQQueue("delay.queue");
@Autowired
private JmsMessagingTemplate template;
/**
*
*
* @param destination destination
* @param message message
*/
public <T extends Serializable> void send(Destination destination, T message) {
template.convertAndSend(destination, message);
}
/**
*
*
* @param destination
* @param data
* @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);
//
producer = session.createProducer(destination);
producer.setDeliveryMode(JmsProperties.DeliveryMode.PERSISTENT.getValue());
ObjectMessage message = session.createObjectMessage(data);
//
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, time);
//
producer.send(message);
log.info(" :{}", 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);
}
}
appication.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.39 INFO 17263--[ main]c.e.demoactivemq.producer.Producer : メッセージの送信:Message Model(titile=10秒の遅延、message=テストメッセージ0)
2019-11-12 22:18:52.953 INFO 17263--[ main]c.e.demoactivemq.producer.Producer : メッセージ送信:Message Model(tile=10秒遅延実行、message=テストメッセージ1)
2019-11-12 22:18:52.958 INFO 17263--[ main]c.e.demoactivemq.producer.Producer : メッセージの送信:Message Model(titile=10秒の遅延、message=テストメッセージ2)
2019-11-12 22:18:52.944 INFO 17263--[ main]c.e.demoactivemq.producer.Producer : メッセージの送信:Message Model(titile=10秒遅延実行、message=テストメッセージ3)
2019-11-12 22:18:52.970 INFO 17263--[ main]c.e.demoactivemq.producer.Producer : メッセージの送信:Message Model(titile=10秒の遅延、message=テストメッセージ4)
2019-11-12 22:19:03.012 INFO 17263-[enerConttainer-1]c.e.demoactivemq.producer.Continmer : メッセージ受信:Message Model(titile=10秒遅延実行、message=テストメッセージ0)
2019-11-12 22:19:03.017 INFO 17263-[enerConttainer-1]c.e.demoactivemq.producer.Continmer : メッセージ受信:Message Model(titile=10秒遅延実行、message=テストメッセージ1)
2019-11-12 22:19:03.019 INFO 17263-[enerConttainer-1]c.e.demoactivemq.producer.Continmer : メッセージ受信:Message Model(tile=10秒遅延実行、message=テストメッセージ2)
2019-11-12 22:19:03.020 INFO 17263-[enerConttainer-1]c.e.demoactivemq.producer.Continmer : 受信メッセージ:Message Model(titile=10秒遅延実行、message=テストメッセージ3)
2019-11-12 22:19:03.021 INFO 17263-[enerConttainer-1]c.e.demoactivemq.producer.Continmer : メッセージ受信:Message Model(titile=10秒遅延実行、message=テストメッセージ4)
あなたより優秀な人はあなたより努力しています。奮闘しない資格がありますか?
締め括りをつける
以上はこの文章の全部の内容です。本文の内容は皆さんの学習や仕事に対して一定の参考学習価値を持ってほしいです。ありがとうございます。