activeMQ 点对点方式

不念不忘少年蓝@ 2021-09-28 06:40 388阅读 0赞

在点对点的传输方式中，消息数据被持久化，每条消息都能被消费，没有监听QUEUE地址也能被消费，数据不会丢失，一对一的发布接受策略，保证数据完整。

点对点的模式主要建立在一个队列上面，当连接一个列队的时候，发送端不需要知道接收端是否正在接收，可以直接向ActiveMQ发送消息，发送的消息将会先进入队列中，如果有接收端在监听，则会发向接收端，如果没有接收端接收，则会保存在activemq服务器，直到接收端接收消息，点对点的消息模式可以有多个发送端，多个接收端，但是一条消息，只会被一个接收端给接收到，哪个接收端先连上ActiveMQ，则会先接收到，而后来的接收端则接收不到那条消息。

**生产者：**

public class JmsProduct {
    	
    	private static final String USER_NAME = ActiveMQConnection.DEFAULT_USER;
    	private static final String USER_PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    	private static final String URL = ActiveMQConnection.DEFAULT_BROKER_URL;
    	private static final String QUEUE_NAME="queue.test";
    
    	public static void main(String[] args) {
    		
    		Connection createConnection;
    		Session createSession;
    		Queue createQueue;
    		//消息目的地
    		Destination destination = null;
    		//消息目的地
    		MessageProducer createProducer = null; 
    		
    		//1.  创建连接工厂
    		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(USER_NAME,USER_PASSWORD,URL);
    		
    		//2.  创建连接
    		try {
    			createConnection = connectionFactory.createConnection();
    			
    			//3. 启动连接
    			createConnection.start();
    			
    			//4. 创建会话
    			createSession = createConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			
    			//5. 创建消息目的地  其实是连接哪个队列 如果是点对点的方式  ，那么他实现的的是Queue。如果是订阅模式，那么他实现的是Topic
    			//	这里我们创建一个名为queue.test的队列
    			destination = createSession.createQueue(QUEUE_NAME);
    			
    			//6. 创建消息生产者
    			createProducer = createSession.createProducer(destination);
    			
    			//7. 发送消息
    			sendMessage(createSession,createProducer);
    		} catch (JMSException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    	}
    
    	private static void sendMessage(Session createSession, MessageProducer createProducer) throws JMSException {
    		// TODO Auto-generated method stub
    		for(int i=0; i<10; i++){
    			//8. 创建一条文本消息
    			TextMessage message = createSession.createTextMessage("activeMQ 发送的消息"+i);
    			System.out.println("activeMQ 发送的消息"+i);
    			//9 发送消息
    			createProducer.send(message);
    		}
    	}
    }

**消费者：**

public class JmsConsumerHasListen {
    
    	private static final String USER_NAME = ActiveMQConnection.DEFAULT_USER;
    	private static final String USER_PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    	private static final String URL = ActiveMQConnection.DEFAULT_BROKER_URL;
    	private static final String QUEUE_NAME="queue.test";
    	
    	public static void main(String[] args) {
    		//1. 创建连接工厂
    		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
    		Connection createConnection = null;
    		Session createSession = null;
    		Destination destination = null;
    		MessageConsumer createConsumer;
    		//2. 创建连接
    		try {
    			createConnection = connectionFactory.createConnection();
    			//3. 启动连接
    			createConnection.start();
    			//4.创建会话
    			createSession = createConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			//5. 创建消息目的地
    			destination = createSession.createQueue(QUEUE_NAME);
    			//6. 创建消费者
    			createConsumer = createSession.createConsumer(destination);
    			
    			//7. 创建消息监听者
    			createConsumer.setMessageListener(new MessageListener() {
    				
    				@Override
    				public void onMessage(Message message) {
    					TextMessage textMessage = (TextMessage) message;
                        try {
                            System.out.println("收到的消息：" + textMessage.getText());
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
    				}
    			});
    			
    		} catch (JMSException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		} finally {
                if (null != createSession) {
                    try {
                    	createSession.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
                if (null != createConnection) {
                    try {
                    	createConnection.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
    	}
    }