activeMQ Topic方式（发布/订阅）

Love The Way You Lie 2021-09-28 06:42 349阅读 0赞

在发布订阅消息方式中，消息是无状态的，不保证每条消息被消费，只有监听该TOPIC地址才能收到消息并消费，否则该消息将会丢失。一对多的发布接受策略，可以同时消费多个消息。

订阅/发布模式，同样可以有着多个发送端与多个接收端，但是接收端与发送端存在时间上的依赖，就是如果发送端发送消息的时候，接收端并没有监听消息，那么ActiveMQ将不会保存消息，将会认为消息已经发送，换一种说法，就是发送端发送消息的时候，接收端不在线，是接收不到消息的，哪怕以后监听消息，同样也是接收不到的。这个模式还有一个特点，那就是发送端发送的消息，将会被所有的接收端给接收到，不类似点对点，一条消息只会被一个接收端给接收到。

**生产者：**

public class JmsTopicProduct {
    
        private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
        private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;
        private static final String TOPIC_NAME = "topic.test";
        
        public static void main(String[] args) {
    		
        	//1. 创建连接工厂
        	ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD,BROKEURL);
        	Connection createConnection = null;
        	Session createSession = null;
        	Destination destination = null;
        	MessageProducer createProducer = null;
        	try {
        		//2. 创建连接
        		createConnection = connectionFactory.createConnection();
        		//3. 启动连接
        		createConnection.start();
        		//4. 创建会话
        		createSession = createConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        		//5. 创建消息目的地
        		destination = createSession.createTopic(TOPIC_NAME);
        		//6. 创建消息生产者
        		createProducer = createSession.createProducer(destination);
        		//7. 发送消息
        		send(createSession,createProducer);
    		} catch (JMSException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    	}
    
    	private static void send(Session createSession, MessageProducer createProducer) throws JMSException {
    		
    		for (int i = 0; i < 20; i++) {
    			TextMessage createTextMessage = createSession.createTextMessage("activeMQ发出一条消息"+i);
    			System.out.println("activeMQ发出一条消息"+i);
    			createProducer.send(createTextMessage);
    		}
    		
    	}
    }

**消费者：**

public class JmsTopicConsumer {
    
        private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
        private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;
        private static final String TOPIC_NAME = "topic.test";
        
        public static void main(String[] args) {
    		//1. 创建连接工厂
        	ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKEURL);
        	Connection connection = null;
        	Session session = null;
        	Destination destination = null;
        	MessageConsumer messageConsumer = null;
        	try {
        		//2. 创建连接
    			connection = connectionFactory.createConnection();
    			//3. 开启连接
    			connection.start();
    			//4. 创建会话
    			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			//5. 创建目的地
    			destination = session.createTopic(TOPIC_NAME);
    			//6. 创建消费者
    			messageConsumer = session.createConsumer(destination);
    			//7. 创建消费者监听
    			messageConsumer.setMessageListener(new MessageListener() {
    				@Override
    				public void onMessage(Message message) {
    					TextMessage textMessage = (TextMessage)message;
    					try {
    						System.out.println("消费者收到一条消息，内容是："+textMessage.getText());
    					} catch (JMSException e) {
    						// TODO Auto-generated catch block
    						e.printStackTrace();
    					}
    				}
    			});
    			
    		} catch (JMSException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    	}
    }