ActiveMQ 持久化Topic
持久模式Topic 发送端
public class PersistanceSender { public static void main(String[] args) throws Exception { ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616"); Connection connection = connectionFactory.createConnection(); Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); Destination destination = session.createTopic("Mytopic"); MessageProducer producer = session.createProducer(destination); producer.setDeliveryMode(DeliveryMode.PERSISTENT);//设置传送模式为持久模式,默认为非持久 connection.start(); for(int i=0; i<3; i++) { TextMessage message = session.createTextMessage("message--"+i); Thread.sleep(1000); //通过消息生产者发出消息 producer.send(message); } session.commit(); session.close(); connection.close(); } }1:要用持久化订阅,发送消息者要用 DeliveryMode.PERSISTENT 模式发现,在连接之前设定
2:一定要设置完成后,再start 这个 connection
持久化模式Topic 接收端
public class PersistanceReceiver { public static void main(String[] args) throws Exception { ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616"); Connection connection = cf.createConnection(); connection.setClientID("cc1");//设置ClientID final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); Topic destination = session.createTopic("Mytopic"); //创建持久化用户 TopicSubscriber consumer = session.createDurableSubscriber(destination,"T1"); connection.start(); Message message = consumer.receive(); while(message!=null) { TextMessage txtMsg = (TextMessage)message; System.out.println("收到消 息:" + txtMsg.getText()); message = consumer.receive(1000L); } session.close(); connection.close(); } }
1:需要在连接上设置消费者id,用来识别消费者 2:需要创建TopicSubscriber来订阅 3:要设置好了过后再start 这个 connection 4:一定要先运行一次,等于向消息服务中间件注册这个消费者,然后再运行客户端发送信息,这个时候, 无论消费者是否在线,都会接收到,不在线的话,下次连接的时候,会把没有收过的消息都接收下来。注册消费者成功之后,在ActiveMQ后台可看到
当cc1用户不在线时,发送端发送topic,之后当cc1用户登录时,可接受到之前发送端发送的内容