ActiveMq数据持久化进Mysql

曾经终败给现在 2023-07-13 12:16 19阅读 0赞

1.activeMq配置文件/usr/local/activemq/conf/activemq.xml 下方注意红色修改

<table> 
 <tbody> 
 <tr> 
 <td> &lt;beans &nbsp; xmlns="http://www.springframework.org/schema/beans" &nbsp; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" &nbsp; xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd &nbsp; http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd"&gt; &nbsp;&nbsp;&nbsp; &lt;!-- Allows us to use system properties as variables in this configuration file --&gt; &nbsp;&nbsp;&nbsp; &lt;bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;property name="locations"&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;value&gt;file:${activemq.conf}/credentials.properties&lt;/value&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/property&gt; &nbsp;&nbsp;&nbsp; &lt;/bean&gt; &nbsp;&nbsp; &lt;!-- Allows accessing the server log&nbsp; --&gt; &nbsp;&nbsp;&nbsp; &lt;bean id="logQuery" class="io.fabric8.insight.log.log4j.Log4jLogQuery" &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; lazy-init="false" scope="singleton" &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; init-method="start" destroy-method="stop"&gt; &nbsp;&nbsp;&nbsp; &lt;/bean&gt; &nbsp;&nbsp;&nbsp; &lt;!-- &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; The &lt;broker&gt; element is used to configure the ActiveMQ broker. &nbsp;&nbsp;&nbsp; --&gt; &nbsp;&nbsp;&nbsp; &lt;broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}"&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;destinationPolicy&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;policyMap&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;policyEntries&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;policyEntry topic="&gt;" &gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;!-- The constantPendingMessageLimitStrategy is used to prevent &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; slow topic consumers to block producers and affect other consumers &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; by limiting the number of messages that are retained &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; For more information, see: &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; http://activemq.apache.org/slow-consumer-handling.html &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; --&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;pendingMessageLimitStrategy&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;constantPendingMessageLimitStrategy limit="1000"/&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/pendingMessageLimitStrategy&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/policyEntry&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/policyEntries&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/policyMap&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/destinationPolicy&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;!-- &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; The managementContext is used to configure how ActiveMQ is exposed in &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; JMX. By default, ActiveMQ uses the MBean server that is started by &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; the JVM. For more information, see: &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; http://activemq.apache.org/jmx.html &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; --&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;managementContext&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;managementContext createConnector="false"/&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/managementContext&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;!-- &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Configure message persistence for the broker. The default persistence &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mechanism is the KahaDB store (identified by the kahaDB tag). &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; For more information, see: &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; http://activemq.apache.org/persistence.html &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&lt;kahaDB directory="${activemq.data}/kahadb"/&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; --&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;persistenceAdapter&gt; &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&lt;jdbcPersistenceAdapter dataDirectory="${activemq.data}" cleanupPeriod="0" dataSource="#mysql-ds" /&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/persistenceAdapter&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;!-- &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; The systemUsage controls the maximum amount of space the broker will &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; use before disabling caching and/or slowing down producers. For more information, see: &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; http://activemq.apache.org/producer-flow-control.html &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; --&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;systemUsage&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;systemUsage&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;memoryUsage&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;memoryUsage percentOfJvmHeap="70" /&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/memoryUsage&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;storeUsage&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;storeUsage limit="100 gb"/&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/storeUsage&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;tempUsage&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;tempUsage limit="50 gb"/&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/tempUsage&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/systemUsage&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/systemUsage&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;!-- &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; The transport connectors expose ActiveMQ over a given protocol to &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; clients and other brokers. For more information, see: &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; http://activemq.apache.org/configuring-transports.html &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; --&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;transportConnectors&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB --&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;transportConnector name="openwire" uri="tcp://douzi8:61616?maximumConnections=1000&amp;amp;wireFormat.maxFrameSize=104857600"/&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;transportConnector name="amqp" uri="amqp://douzi8:5672?maximumConnections=1000&amp;amp;wireFormat.maxFrameSize=104857600"/&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;transportConnector name="stomp" uri="stomp://douzi8:61613?maximumConnections=1000&amp;amp;wireFormat.maxFrameSize=104857600"/&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;transportConnector name="mqtt" uri="mqtt://douzi8:1883?maximumConnections=1000&amp;amp;wireFormat.maxFrameSize=104857600"/&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;transportConnector name="ws" uri="ws://douzi8:61614?maximumConnections=1000&amp;amp;wireFormat.maxFrameSize=104857600"/&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/transportConnectors&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;!-- destroy the spring context on shutdown to stop jetty 关闭时销毁spring上下文以停止jetty --&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;shutdownHooks&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" /&gt; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &lt;/shutdownHooks&gt; &nbsp;&nbsp;&nbsp; &lt;/broker&gt; &nbsp;&nbsp; &nbsp;&lt;bean id="mysql-ds" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close"&gt; &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&lt;property name="driverClassName" value="com.mysql.jdbc.Driver"/&gt; &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&lt;property name="url" value="jdbc:mysql://localhost:3306/persistence_mq?relaxAutoCommit=true"/&gt; &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&lt;property name="username" value="root"/&gt; &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&lt;property name="password" value="123456"/&gt; &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&lt;property name="initialSize" value="5" /&gt; &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&lt;property name="maxTotal" value="100" /&gt; &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&lt;property name="maxIdle" value="30" /&gt; &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&lt;property name="maxWaitMillis" value="10000" /&gt; &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&lt;property name="minIdle" value="1" /&gt; &nbsp;&nbsp; &nbsp;&lt;/bean&gt; &nbsp;&nbsp;&nbsp; &lt;!-- &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Enable web consoles, REST and Ajax APIs and demos &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; The web consoles requires by default login, you can disable this in the jetty.xml file &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Take a look at ${ACTIVEMQ_HOME}/conf/jetty.xml for more details &nbsp;&nbsp;&nbsp; --&gt; &nbsp;&nbsp;&nbsp; &lt;import resource="jetty.xml"/&gt; &lt;/beans&gt; </td> 
 </tr> 
 </tbody> 
</table>

2./usr/local/activemq/lib/ 下放mysql-connector-java-5.1.47.jar

3.mysql中建立persistence\_mq数据库

4.执行程序，只发不接即可看到数据库activemq\_msgs中存在数据了。

5.spring boot 项目使用

<table> 
 <tbody> 
 <tr> 
 <td> # activeMq配置 #spring.activemq.broker-url=failover:(tcp://douzi8:61616) spring.activemq.broker-url=tcp://douzi8:61616 spring.activemq.user=admin spring.activemq.password=admin spring.activemq.in-memory=true # true表示使用连接池 spring.activemq.pool.enabled=false # 连接池最大连接数 spring.activemq.pool.max-connections=10 # 空闲的连接过期时间，默认为30秒 spring.activemq.pool.idle-timeout=30000 # 强制的连接过期时间，与idleTimeout的区别在于：idleTimeout是在连接空闲一段时间失效，而expiryTimeout不管当前连接的情况，只要达到指定时间就失效。默认为0，never spring.activemq.pool.expiry-timeout=0 # 当有"JMSException"时尝试重新连接 spring.activemq.pool.reconnect-on-exception=true # 在空闲连接清除线程之间运行的时间。当为负数时，没有空闲连接驱逐线程运行。 spring.activemq.pool.time-between-expiration-check=-1ms # 是否只使用一个MessageProducer #spring.activemq.pool.use-anonymous-producers=true queue.sync=douzi-queue </td> 
 </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
 <tr> 
 <td> import javax.jms.Queue; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.command.ActiveMQQueue; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.jms.annotation.EnableJms; import org.springframework.jms.config.DefaultJmsListenerContainerFactory; import org.springframework.jms.config.JmsListenerContainerFactory; /** &nbsp;* &lt;p&gt;&lt;b&gt;ActiveMQConfig Description:&lt;/b&gt; activemq配置&lt;/p&gt; &nbsp;* @author douzi &nbsp;* &lt;b&gt;DATE&lt;/b&gt; 2020年3月10日 上午10:27:22 &nbsp;*/ @Configuration @EnableJms public class ActiveMQConfig { &nbsp;&nbsp;&nbsp; @Value("${queue.sync}") &nbsp;&nbsp;&nbsp; private String syncQueue; &nbsp;&nbsp;&nbsp; @Value("${spring.activemq.user}") &nbsp;&nbsp;&nbsp; private String usrName; &nbsp;&nbsp;&nbsp; @Value("${spring.activemq.password}") &nbsp;&nbsp;&nbsp; private&nbsp; String password; &nbsp;&nbsp;&nbsp; @Value("${spring.activemq.broker-url}") &nbsp;&nbsp;&nbsp; private&nbsp; String brokerUrl; &nbsp;&nbsp;&nbsp; @Bean &nbsp;&nbsp;&nbsp; public Queue syncQueue(){ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return new ActiveMQQueue(syncQueue); &nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; @Bean &nbsp;&nbsp;&nbsp; public ActiveMQConnectionFactory connectionFactory() { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return new ActiveMQConnectionFactory(usrName, password, brokerUrl); &nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; @Bean &nbsp;&nbsp;&nbsp; public JmsListenerContainerFactory&lt;?&gt; jmsListenerContainerQueue(ActiveMQConnectionFactory connectionFactory){ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; factory.setPubSubDomain(false); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; factory.setConnectionFactory(connectionFactory); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // 开启持久化 //&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; factory.setSubscriptionDurable(true); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return factory; &nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; @Bean &nbsp;&nbsp;&nbsp; public JmsListenerContainerFactory&lt;?&gt; jmsListenerContainerTopic(ActiveMQConnectionFactory connectionFactory){ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //设置为发布订阅方式, 默认情况下使用的生产消费者方式 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; bean.setPubSubDomain(true); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; bean.setConnectionFactory(connectionFactory); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // 开启持久化 //&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; bean.setSubscriptionDurable(true); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return bean; &nbsp;&nbsp;&nbsp; } } </td> 
 </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
 <tr> 
 <td> 生产者: import java.util.Random; import java.util.concurrent.ThreadPoolExecutor; import java.util.stream.IntStream; import javax.jms.Queue; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.CommandLineRunner; import org.springframework.jms.core.JmsMessagingTemplate; import org.springframework.stereotype.Component; import com.wondertek.moms.common.utils.MyThreadPool; import lombok.extern.slf4j.Slf4j; @Slf4j //@Component public class ProducerTest implements CommandLineRunner { &nbsp;&nbsp; &nbsp;@Autowired &nbsp;&nbsp; &nbsp;private JmsMessagingTemplate jmsMessagingTemplate; &nbsp;&nbsp; &nbsp;@Autowired &nbsp;&nbsp; &nbsp;private Queue syncQueue; &nbsp;&nbsp; &nbsp; &nbsp;&nbsp; &nbsp;private final static Random r = new Random(System.currentTimeMillis()); &nbsp;&nbsp; &nbsp;@Override &nbsp;&nbsp; &nbsp;public void run(String... args) throws Exception { &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;log.info("Message was sent to the Queue"); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; for (int j = 0; j &lt; 5; j++) { &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;new Thread(() -&gt;forData(), "t" + j).start(); &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;} &nbsp;&nbsp; &nbsp;} &nbsp;&nbsp; &nbsp; &nbsp;&nbsp; &nbsp;public void forData() { &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;for (int i = 0; i &lt; 10000; i++) { &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;try { &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;Thread.sleep(r.nextInt(100)); &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;} catch (InterruptedException e) { &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;e.printStackTrace(); &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;} &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;send("douzi " + Thread.currentThread().getName() + " " + i); &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;} &nbsp;&nbsp; &nbsp;} &nbsp;&nbsp; &nbsp;public void send(String msg) { &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;try { &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;Thread.sleep(10); &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;} catch (InterruptedException e) { &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;e.printStackTrace(); &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;} &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;this.jmsMessagingTemplate.convertAndSend(this.syncQueue, msg); &nbsp;&nbsp; &nbsp;} } </td> 
 </tr> 
 <tr> 
 <td> 消费者： import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; import lombok.extern.slf4j.Slf4j; //@Component @Slf4j public class ConsumerTest { &nbsp;&nbsp; &nbsp;@JmsListener(destination = "${queue.sync}") &nbsp;&nbsp; &nbsp;public void receiveQueue(String text) { &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;log.info("------- 1:" + text); &nbsp;&nbsp; &nbsp;} &nbsp;&nbsp; &nbsp; &nbsp;&nbsp; &nbsp;@JmsListener(destination = "${queue.sync}") &nbsp;&nbsp; &nbsp;public void receiveQueue1(String text) { &nbsp;&nbsp; &nbsp;&nbsp;&nbsp; &nbsp;log.info("------- 2:" + text); &nbsp;&nbsp; &nbsp;} } </td> 
 </tr> 
 </tbody> 
</table>