当前位置：文档之家 › ActiveMQ的持久化(数据库)

ActiveMQ的持久化(数据库)

  • 格式：doc
  • 大小：119.50 KB
  • 文档页数：11

利用消息队列的异步策略，可以从很大程序上缓解程序的压力，但是，如果MQ 所在的机器down机了,又如果队列中的数据不是持久的就会发生数据丢失，后果是可想而知的，所以消息的持久化是不可不讨论的话题。

1) 关于ActiveMQ消息队列的持久化，主要是在ActiveMQ的配置文件中设置(看粗体部分).Java代码1.<!--2. Licensed to the Apache Software Foundation (ASF) under oneor more3. contributor license agreements. See the NOTICE file distributed with4. this work for additional information regarding copyright ownership.5. The ASF licenses this file to You under the Apache License,Version 2.06. (the "License"); you may not use this file except in compliance with7. the License. You may obtain a copy of the License at8.9. /licenses/LICENSE-2.010.11. Unless required by applicable law or agreed to in writing,software12. distributed under the License is distributed on an "AS IS"BASIS,13. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.14. See the License for the specific language governing permissions and15. limitations under the License.16.-->17.<!-- START SNIPPET: example -->18.<beans19. xmlns="/schema/beans"20. xmlns:amq="/schema/core"21. xmlns:xsi="/2001/XMLSchema-instance"22. xsi:schemaLocation="/schema/beans /schema/beans/spring-beans-2.0 .xsd23. /schema/core http://activemq.apache.org/schema/core/activemq-core.xsd24. /camel/schema/spring http://activem/camel/schema/spring/camel-spring.xsd">25.26. <!-- Allows us to use system properties as variables in this configuration file -->27. <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">28. <property name="locations">29. <value>file:///${activemq.base}/conf/credentials.properties</value>30. </property>31. </bean>32.33. <broker xmlns="/schema/core" brokerName="localhost" dataDirectory="${activemq.base}/data">34.35. <!-- Destination specific policies using destination names or wildcards -->36. <destinationPolicy>37. <policyMap>38. <policyEntries>39. <policyEntry queue=">" memoryLimit="5mb"/>40. <policyEntry topic=">" memoryLimit="5mb">41. <!-- you can add other policies too suchas these42. <dispatchPolicy>43. <strictOrderDispatchPolicy/>44. </dispatchPolicy>45. <subscriptionRecoveryPolicy>46. <lastImageSubscriptionRecoveryPolicy/>47. </subscriptionRecoveryPolicy>48. -->49. </policyEntry>50. </policyEntries>51. </policyMap>52. </destinationPolicy>53.54. <!-- Use the following to configure how ActiveMQ is exposed in JMX -->55. <managementContext>56. <managementContext createConnector="false"/>57. </managementContext>58.59. <!-- The store and forward broker networks ActiveMQ will listen to -->60. <networkConnectors>61. <!-- by default just auto discover the other brokers -->62. <networkConnector name="default-nc" uri="multicast://default"/>63. <!-- Example of a static configuration:64. <networkConnector name="host1 and host2" uri="static://(tcp://host1:61616,tcp://host2:61616)"/>65. -->66. </networkConnectors>67.68. <persistenceAdapter>69. <amqPersistenceAdapter syncOnWrite="false" directory="${activemq.base}/data" maxFileLength="20 mb"/>70. </persistenceAdapter>71.72. <!-- Use the following if you wish to configure the journal with JDBC -->73. <!--74. <persistenceAdapter>75. <journaledJDBC dataDirectory="${activemq.base}/data" dataSource="#postgres-ds"/>76. </persistenceAdapter>77. -->78.79. <!-- Or if you want to use pure JDBC without a journal-->80.81. <SPAN style="COLOR: #ff6600"> <SPAN style="COLOR: #000000"><persistenceAdapter>82. <jdbcPersistenceAdapter dataSource="#mysql-ds"/>83. </persistenceAdapter></SPAN></SPAN>84.85.86. <sslContext>87. <sslContext keyStore="file:${activemq.base}/conf/broker.ks" keyStorePassword="password" trustStore="file:${activem q.base}/conf/broker.ts" trustStorePassword="password"/>88. </sslContext>89.90. <!-- The maximum about of space the broker will use before slowing down producers -->91. <systemUsage>92. <systemUsage>93. <memoryUsage>94. <memoryUsage limit="20 mb"/>95. </memoryUsage>96. <storeUsage>97. <storeUsage limit="1 gb" name="foo"/>98. </storeUsage>99. <tempUsage>100. <tempUsage limit="100 mb"/>101. </tempUsage>102. </systemUsage>103. </systemUsage>104.105.106. <!-- The transport connectors ActiveMQ will listen to -->107. <transportConnectors>108. <transportConnector name="openwire" uri="tcp:/ /localhost:61616" discoveryUri="multicast://default"/> 109. <transportConnector name="ssl" uri="ssl://loca lhost:61617"/>110. <transportConnector name="stomp" uri="stomp:// localhost:61613"/>111. <transportConnector name="xmpp" uri="xmpp://lo calhost:61222"/>112. </transportConnectors>113.114. </broker>115.116. <!--117. ** Lets deploy some Enterprise Integration Patterns in side the ActiveMQ Message Broker118. ** For more details see119. **120. ** /enterprise-integration-p atterns.html122. <camelContext id="camel" xmlns="http://activemq.apache .org/camel/schema/spring">123.124. <!-- You can use a <package> element for each root package to search for Java routes -->125. <package>org.foo.bar</package>126.127. <!-- You can use Spring XML syntax to define the r outes here using the <route> element -->128. <route>129. <from uri="activemq:example.A"/>130. <to uri="activemq:example.B"/>131. </route>132. </camelContext>133.134. <!--135. ** Lets configure some Camel endpoints136. **137. ** /camel/components.html 138. -->139.140. <!-- configure the camel activemq component to use the current broker -->141. <bean id="activemq" class="org.apache.activemq.camel.c omponent.ActiveMQComponent" >142. <property name="connectionFactory">143. <bean class="org.apache.activemq.ActiveMQConnect ionFactory">144. <property name="brokerURL" value="vm://localho st?create=false&amp;waitForStart=10000" />145. <property name="userName" value="${ ername}"/>146. <property name="password" value="${activemq.pa ssword}"/>147. </bean>148. </property>149. </bean>150.151.152.153. <!-- Uncomment to create a command agent to respond to message based admin commands on the ActiveMQ.Agent topic -->155. <commandAgent xmlns="/schema /core" brokerUrl="vm://localhost" username="${ername }" password="${activemq.password}"/>156. -->157.158.159. <!-- An embedded servlet engine for serving up the Adm in console -->160. <jetty xmlns="/schemas/jetty/1.0">161. <connectors>162. <nioConnector port="8161"/>163. </connectors>164.165. <handlers>166. <webAppContext contextPath="/admin" resourceBa se="${activemq.base}/webapps/admin" logUrlOnStart="true"/> 167. <webAppContext contextPath="/demo" resourceBas e="${activemq.base}/webapps/demo" logUrlOnStart="true"/> 168. <webAppContext contextPath="/fileserver" resou rceBase="${activemq.base}/webapps/fileserver" logUrlOnStart="tr ue"/>169. </handlers>170. </jetty>171.172. <!-- This xbean configuration file supports all the s tandard spring xml configuration options -->173.174. <!-- Postgres DataSource Sample Setup -->175. <!--176. <bean id="postgres-ds" class="org.postgresql.ds.PGPool ingDataSource">177. <property name="serverName" value="localhost"/> 178. <property name="databaseName" value="activemq"/> 179. <property name="portNumber" value="0"/>180. <property name="user" value="activemq"/>181. <property name="password" value="activemq"/> 182. <property name="dataSourceName" value="postgres"/>183. <property name="initialConnections" value="1"/> 184. <property name="maxConnections" value="10"/> 185. </bean>186. -->187.188. <!-- MySql DataSource Sample Setup -->189.190.<SPAN style="COLOR: #000000"> </SPAN><SPAN style="COLOR: #000000"> <bean id="mysql-ds" class="mons.dbcp.B asicDataSource" destroy-method="close">191. <property name="driverClassName" value="com.mysql.jd bc.Driver"/>192. <property name="url" value="jdbc:mysql://localhost:3 306/activemq?relaxAutoCommit=true"/>193. <property name="username" value="root"/>194. <property name="password" value=""/>195. <property name="maxActive" value="200"/>196. <property<STRONG> </STRONG>name="poolPreparedStateme nts" value="true"/>197. </bean></SPAN>198.199.200. <!-- Oracle DataSource Sample Setup -->201. <!--202. <bean id="oracle-ds" class="mons.dbcp.Ba sicDataSource" destroy-method="close">203. <property name="driverClassName" value="oracle.jdbc.driver.OracleDriver"/>204. <property name="url" value="jdbc:oracle:thin:@localh ost:1521:AMQDB"/>205. <property name="username" value="scott"/>206. <property name="password" value="tiger"/>207. <property name="maxActive" value="200"/>208. <property name="poolPreparedStatements" value="true"/>209. </bean>210. -->211.212. <!-- Embedded Derby DataSource Sample Setup --> 213. <!--214. <bean id="derby-ds" class="org.apache.derby.jdbc.Embed dedDataSource">215. <property name="databaseName" value="derbydb"/> 216. <property name="createDatabase" value="create"/> 217. </bean>218. -->219.220.</beans>221.<!-- END SNIPPET: example -->改动部分主要是设置了mysql的datasource声明, 还有就是采用mysql作为persistenceAdapter,并声明如下。

合集下载

activemq + mysql 分布式队列实现原理

activemq + mysql 分布式队列实现原理

activemq + mysql 分布式队列实现原理一、概述分布式队列是一种在多台计算机之间共享数据的方式，它允许将任务分配到不同的节点进行处理，从而提高系统的处理能力和可扩展性。

ActiveMQ和MySQL是两种常用的分布式队列实现方式，其中ActiveMQ是一个流行的消息队列系统，而MySQL则提供了数据存储功能。

二、ActiveMQ概述ActiveMQ是一个开源的消息队列系统，它提供了发布/订阅模式和消息队列模式两种消息传输方式。

发布/订阅模式允许一个消息发布到多个消费者，而消息队列模式则允许将消息存储在队列中等待消费者处理。

ActiveMQ支持多种消息协议，如AMQP、JMS和MQTT等。

三、MySQL概述MySQL是一个流行的关系型数据库管理系统（RDBMS）。

在这里，我们将MySQL用作分布式队列的数据存储层，为队列中的消息提供持久化存储和检索功能。

MySQL支持丰富的SQL查询语言，可以轻松地进行数据检索、更新和删除操作。

1. 消息生产：消息生产者使用ActiveMQ的API将消息发布到队列中。

这些消息被保存在MySQL数据库中，以便于存储和检索。

2. 消息传输：消费者从ActiveMQ的队列中获取消息。

如果队列为空，消费者将等待直到有新消息可用。

3. 数据持久化：MySQL为分布式队列提供了数据持久化的功能。

消息被写入MySQL数据库，以便在系统故障时能够恢复数据。

4. 负载均衡：ActiveMQ和MySQL可以协同工作，实现负载均衡，将任务分配到不同的节点进行处理，从而提高系统的处理能力和可扩展性。

5. 故障恢复：如果某台节点的ActiveMQ出现故障，系统可以自动将消息重新路由到其他健康的节点，从而保证系统的可用性和可靠性。

6. 消息确认：消费者在处理完消息后，通过MySQL数据库进行消息确认，以确保消息已被成功处理。

7. 配置管理：通过配置管理，可以轻松地扩展系统规模，添加或删除节点，而无需人工干预。

ActiveMQ面试题及答案

ActiveMQ面试题及答案

ActiveMQ面试题及答案1、什么是ActiveMQ？activeMQ是一种开源的，实现了JMS1.1规范的，面向消息(MOM)的中间件，为应用程序提供高效的、可扩展的、稳定的和安全的企业级消息通信。

2、Activemq的瓶颈值根据网上一般评测文档上来看，每秒的消息吞吐在2000以上, acticemq 也可以集群化部署，也是使用zookeeper来搭建。

3、ActiveMQ服务器宕机怎么办？这得从ActiveMQ的储存机制说起。

在通常的情况下，非持久化消息是存储在内存中的，持久化消息是存储在文件中的，它们的最大限制在配置文件的<systemUsage>节点中配置。

但是，在非持久化消息堆积到一定程度，内存告急的时候，ActiveMQ会将内存中的非持久化消息写入临时文件中，以腾出内存。

虽然都保存到了文件里，但它和持久化消息的区别是，重启后持久化消息会从文件中恢复，非持久化的临时文件会直接删除。

那如果文件增大到达了配置中的最大限制的时候会发生什么？我做了以下实验：设置2G左右的持久化文件限制，大量生产持久化消息直到文件达到最大限制，此时生产者阻塞，但消费者可正常连接并消费消息，等消息消费掉一部分，文件删除又腾出空间之后，生产者又可继续发送消息，服务自动恢复正常。

设置2G左右的临时文件限制，大量生产非持久化消息并写入临时文件，在达到最大限制时，生产者阻塞，消费者可正常连接但不能消费消息，或者原本慢速消费的消费者，消费突然停止。

整个系统可连接，但是无法提供服务，就这样挂了。

具体原因不详，解决方案：尽量不要用非持久化消息，非要用的话，将临时文件限制尽可能的调大。

4、AcitveMQ的作用、原理？（生产者、消费者。

p2p、订阅实现流程）Activemq的作用就是系统之间进行通信。

当然可以使用其他方式进行系统间通信，如果使用Activemq的话可以对系统之间的调用进行解耦，实现系统间的异步通信。

activemq用法

activemq用法

activemq用法ActiveMQ是一个开源的分布式消息中间件，采用Java编写，它是Apache软件基金会的一个顶级项目。

ActiveMQ实现了Java Message Service（JMS）规范，可以在不同系统之间进行消息传递和通信。

下面将介绍ActiveMQ的用法。

1.安装与配置：2.创建队列：在ActiveMQ管理界面上，点击"Queues"选项卡，然后点击"Add a new Queue"按钮。

给队列指定一个名称，并配置其他参数，如持久性、最大消息数等。

点击"Add"按钮后，队列就会被创建。

3.发送和接收消息：ActiveMQ提供了多种编程语言的客户端库，可以方便地发送和接收消息。

以Java为例，使用JMS API可以完成这些操作。

首先，需要创建一个ConnectionFactory对象，指定ActiveMQ的连接URL。

然后，通过ConnectionFactory创建一个Connection对象，并启动连接。

接下来，创建一个Session对象，用于发送和接收消息。

最后，创建一个MessageProducer对象，通过调用它的send方法发送消息。

同样地，可以创建一个MessageConsumer对象，接收消息。

接收消息的时候，可以通过调用MessageConsumer的receive方法，阻塞等待接收消息；也可以注册一个MessageListener，通过回调方式接收消息。

具体代码如下：```javaimport javax.jms.*;public class ActiveMQExamplepublic static void main(String[] args) throws JMSException ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerURL);Connection connection = connectionFactory.createConnection(;connection.start(;Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);Destination destination = session.createQueue("myQueue");MessageProducer producer =session.createProducer(destination);//发送消息TextMessage message = session.createTextMessage(;message.setText("Hello, ActiveMQ!");producer.send(message);//接收消息MessageConsumer consumer =session.createConsumer(destination);Message receivedMessage = consumer.receive(;if (receivedMessage instanceof TextMessage)String text = ((TextMessage) receivedMessage).getText(;System.out.println("Received message: " + text);}session.close(;connection.close(;}}```4.持久化与事务：ActiveMQ支持消息的持久化，保证消息在传送过程中不会丢失。

activemq用法

activemq用法

activemq用法ActiveMQ是一个消息中间件，它实现了Java Message Service (JMS)规范，通过提供高性能、可靠的消息传递机制，帮助开发人员构建分布式系统和异步应用程序。

以下是关于ActiveMQ的一些常见用法：1. 消息队列：ActiveMQ允许应用程序通过消息队列进行异步通信。

发送方将消息放入队列中，接收方可以从队列中获取消息进行处理。

这种方式可以实现解耦和异步处理，提高应用程序的可伸缩性和性能。

2. 发布/订阅模型：ActiveMQ还提供了发布/订阅模型，允许发布者将消息发布到主题（Topic），然后订阅者可以从主题中订阅消息进行处理。

这种模型适用于广播和通知场景，可以实现消息的多播。

3. 消息传递可靠性：ActiveMQ提供了多种机制来确保消息传递的可靠性。

包括持久化存储、消息重发机制、事务机制等。

持久化存储可以确保即使在应用程序关闭或崩溃后，消息也不会丢失。

消息重发机制可以确保消息在网络故障或其他错误情况下的可靠传递。

4. 消息过滤和选择：ActiveMQ允许消费者使用消息选择器来选择订阅的消息。

通过指定一些条件，可以筛选出需要处理的消息，减少消费者的工作量和网络传输的开销。

5. 点对点和异步处理：ActiveMQ支持点对点的消息传递方式，也支持异步消息处理。

可以将消息发送到指定的队列，然后异步地处理消息，提高应用程序的整体性能和吞吐量。

6. 集成Spring框架：ActiveMQ与Spring框架可以很好地集成，Spring提供了ActiveMQ的连接工厂（ConnectionFactory）和JmsTemplate等组件，简化了ActiveMQ的使用和配置。

7. 监控和管理：ActiveMQ提供了Web控制台和JMX接口，用于监控和管理ActiveMQ服务器。

可以查看连接数、队列状态、消息的发送和接收情况等，并可以进行动态配置和调整。

总之，ActiveMQ是一个功能强大的消息中间件，可以帮助开发人员实现可靠、高性能的消息传递。