生产者
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class PersistenceSender {
public static void main(String[] args) throws JMSException, InterruptedException {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
Connection connection = connectionFactory.createConnection();
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createTopic("MyTopic2");
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
connection.start();
for (int i = 0; i < 3; i++) {
TextMessage message = session.createTextMessage("message--" + i);
producer.send(message);
}
session.commit();
session.close();
connection.close();
}
}
消费者
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class PersistenceReceiver {
public static void main(String[] args) throws JMSException {
//1:需要在连接上设置笑着id,用来识别消费者
//2.需要创建TopicSubscriber来订阅
//3.要设置好了过后再start这个connection
//4.一定要先运行一次,等于向消息服务中间件注册这个消费者,然后在运行客户端发送消息,
// 这个时候,无论消费者是否在线,都会接收到,不在线的话,下次连接的时候,会把没有收过的消息都接收下来。
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
Connection connection = connectionFactory.createConnection();
connection.setClientID("cc1");
Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
Topic destination = session.createTopic("MyTopic2");
TopicSubscriber ts = session.createDurableSubscriber(destination, "T1");
connection.start();
Message message = ts.receive();//receive()是阻塞方法
while (message != null) {
TextMessage txtMsg = (TextMessage) message;
System.out.println("收到消息:" + txtMsg.getText());
message = ts.receive(1000L);
}
session.commit();
session.close();
connection.close();
}
}
先运行消费者,然后停止消费者。看ActiveMQ的状态:
运行生产者:
再运行消费者:
说明,消费者消费消息之后,topic的Messages Dequeued还是0,这是因为,有可能有其他的消费者还订阅了。