This is the third day of my participation in the August More Text Challenge.

Mysql is used for persistence, but everything else is pretty much the same

Step 1: Copy the MySQL database driver into the activeMQ lib directory.

​

Step 2: Configure the persistent adapter in the ${Activemq.base}/conf/ Activemq.xml file

Create table createTablesOnStartup = “false”

UseDatabaseLock =”false” lock (Problem 2)

<persistenceAdapter>

<jdbcPersistenceAdapter dataDirectory="${activemq.base}/data" dataSource="#derby-ds"/>

</persistenceAdapter>
Copy the code

Step 3: Configure the data source in the ${Activemq.base}/conf/ Activemq.xml file

<bean id="derby-ds" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close">

<property name="driverClassName" value="com.mysql.jdbc.Driver"/>

<property name="url" value="jdbc:mysql://localhost/activemq?relaxAutoCommit=true"/>

<property name="username" value="root"/>

<property name="password" value="root"/>

<property name="poolPreparedStatements" value="true"/>

</bean>
Copy the code

Configuration problems:

1. Caused by: org.xml.sax.SAXParseException; lineNumber: 92; columnNumber: 92; Cvc-complex-type.2.4.a: Invalid content starting with element ‘bean’ was found. It should be ‘{“Activemq.apache.org/schema/core…’one of the beginning.

Solution:

Place the bean in the bean label section

​

2. Failed to acquire lock. Sleeping for 10000 milli(s) before trying again…

Solution:

UseDatabaseLock = “false” configuration

<persistenceAdapter>

<jdbcPersistenceAdapter dataDirectory="${activemq.base}/data" dataSource="#derby-ds" useDatabaseLock="false" />

</persistenceAdapter>
Copy the code

3. After adding mysql persistence to ActiveMQ, a message was sent, but there was no record in MSGS table

1. Activemq database will create 3 tables after persistence

​

This is required in producer

producer.setDeliveryMode(DeliveryMode.PERSISTENT); // producer.setTimeToLive(10); / / send a producer. The send (textMessage); // producer.send(textMessage, DeliveryMode.PERSISTENT, 1, 60 * 60 * 24);Copy the code

Add notice to The Consumer consumer that you need to set the client ID before opening the link

// Set the client ID

// Set the client ID connection.setclientid ("client-1"); connection.start(); // final MessageConsumer messageConsumer = session.createConsumer(topic); / / normal subscription MessageConsumer MessageConsumer = session. CreateDurableSubscriber (topic, "bb"); // Persistent subscriptionCopy the code

​

Code:

import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; /** * @author Yang * Producer */ public class TopicProducer {/** * userName */ private static final String userName = ActiveMQConnection.DEFAULT_USER; / * * * * / passWord private static final String passWord = ActiveMQConnection. DEFAULT_PASSWORD; /** * url */ private static final String brokerUrl = ActiveMQConnection.DEFAULT_BROKER_URL; public void send(String message) { try { ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(userName, passWord, brokerUrl); final Connection connection = connectionFactory.createConnection(); connection.start(); Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); Final Topic = session.createTopic(" Topic "); final MessageProducer producer = session.createProducer(topic); TextMessage textMessage = session.createTextMessage(message); producer.setDeliveryMode(DeliveryMode.PERSISTENT); // producer.setTimeToLive(10); / / send a producer. The send (textMessage); // producer.send(textMessage, DeliveryMode.PERSISTENT, 1, 60 * 60 * 24); // session.commit(); producer.close(); session.close(); connection.close(); } catch (JMSException e) { e.printStackTrace(); } } public static void main(String[] args) { TopicProducer producer = new TopicProducer(); producer.send("hello world"); }}Copy the code

import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; /** * @author Yang * Consumer */ public class TopicConsumer {/** * userName */ private static final String userName = ActiveMQConnection.DEFAULT_USER; / * * * * / passWord private static final String passWord = ActiveMQConnection. DEFAULT_PASSWORD; /** * url */ private static final String brokerUrl = ActiveMQConnection.DEFAULT_BROKER_URL; public void receive() { try { ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(userName, passWord, brokerUrl); final Connection connection = connectionFactory.createConnection(); // Set the client ID connection.setclientid ("client-1"); connection.start(); Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); Final Topic = session.createTopic(" Topic "); // final MessageConsumer messageConsumer = session.createConsumer(topic); / / normal subscription MessageConsumer MessageConsumer = session. CreateDurableSubscriber (topic, "bb"); / / durable subscriptions messageConsumer. SetMessageListener (n - > {try {TextMessage MSG = (TextMessage) n; final String text = msg.getText(); If (text.equalsignorecase ("hello world")) {system.out.println (" accept message: "+ msg.gettext ()); } else {system.out.println (" test reprint times "); int i = 1 / 0; } } catch (JMSException e) { // e.printStackTrace(); }}); } catch (JMSException e) { // e.printStackTrace(); } } public static void main(String[] args) { TopicConsumer consumer = new TopicConsumer(); consumer.receive(); }}Copy the code