1.持久化以后 activemq数据库 会创建3张表
<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>
在 producer 生产者 中需要设置
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
// producer.setTimeToLive(10);
// 发送
producer.send(textMessage);
// producer.send(textMessage, DeliveryMode.PERSISTENT, 1, 60 * 60 * 24);
在consumer 消费者中 添加 注意 设置客户端id 需要在开启链接 前面
//设置客户端id
//设置客户端id
connection.setClientID("client-1");
connection.start();
// final MessageConsumer messageConsumer = session.createConsumer(topic);//普通订阅
MessageConsumer messageConsumer = session.createDurableSubscriber(topic,"bb"); //持久订阅
代码：
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @author Yang
* @create 2018-04-19 17:11
* 描述： 生产者
*/
public class TopicProducer {
/**
* 用户名
*/
private static final String userName = ActiveMQConnection.DEFAULT_USER;
/**
* 密码
*/
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 topic = session.createTopic("topic");
final MessageProducer producer = session.createProducer(topic);
TextMessage textMessage = session.createTextMessage(message);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
// producer.setTimeToLive(10);
// 发送
producer.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");
}
}
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @author Yang
* @create 2018-04-19 17:10
* 描述： 消费者
*/
public class TopicConsumer {
/**
* 用户名
*/
private static final String userName = ActiveMQConnection.DEFAULT_USER;
/**
* 密码
*/
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();
//设置客户端id
connection.setClientID("client-1");
connection.start();
Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
//创建队列
final Topic topic = session.createTopic("topic");
// final MessageConsumer messageConsumer = session.createConsumer(topic);//普通订阅
MessageConsumer messageConsumer = session.createDurableSubscriber(topic,"bb"); //持久订阅
messageConsumer.setMessageListener(n -> {
try {
TextMessage msg = (TextMessage) n;
final String text = msg.getText();
if (text.equalsIgnoreCase("hello world")) {
System.out.println(" 接受信息： " + msg.getText());
} else {
System.out.println(" 测试重发次数 ");
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();
}
}
|
|
收藏
淘帖
踩
