ActiveMQ 实现点对点的消息模型小Demo

「爱情、让人受尽委屈。」 2022-06-03 01:17 212阅读 0赞

摘要:JMS(JAVA Message Service,java消息服务) API是一个消息服务的标准或者说是规范,允许应用程序组件基于JavaEE平台创建,发送,接受和读取消息.它使分布式通信耦合度更低,消息服务更加可靠以及异步性

**JMS有两种消息模型：点对点和发布订阅模型.本文章主要介绍点对点的消息模型**

**P2P模式图**

**![Center][]**

**涉及到的概念**

1.  消息队列（Queue）
2.  发送者(Sender)
3.  接收者(Receiver)
4.  每个消息都被发送到一个特定的队列，接收者从队列中获取消息。队列保留着消息，直到他们被消费或超时。

**P2P的特点**

1.  每个消息只有一个消费者（Consumer）(即一旦被消费，消息就不再在消息队列中)
2.  发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列
3.  接收者在成功接收消息之后需向队列应答成功
4.  如果你希望发送的每个消息都应该被成功处理的话，那么你需要P2P模式

**消息的消费**   
 在JMS中，消息的产生和消息是异步的。对于消费来说，JMS的消息者可以通过两种方式来消费消息。   
 ○ 同步   
 订阅者或接收者调用receive方法来接收消息，receive方法在能够接收到消息之前（或超时之前）将一直阻塞   
 ○ 异步   
 订阅者或接收者可以注册为一个消息监听器。当消息到达之后，系统自动调用监听器的onMessage方法。

**JMS编程模型**

(1) ConnectionFactory

创建Connection对象的工厂，针对两种不同的jms消息模型，分别有QueueConnectionFactory和TopicConnectionFactory两种。可以通过JNDI来查找ConnectionFactory对象。

(2) Destination

Destination的意思是消息生产者的消息发送目标或者说消息消费者的消息来源。对于消息生产者来说，它的Destination是某个队列（Queue）或某个主题（Topic）;对于消息消费者来说，它的Destination也是某个队列或主题（即消息来源）。

所以，Destination实际上就是两种类型的对象：Queue、Topic可以通过JNDI来查找Destination。

(3) Connection

Connection表示在客户端和JMS系统之间建立的链接（对TCP/IP socket的包装）。Connection可以产生一个或多个Session。跟ConnectionFactory一样，Connection也有两种类型：QueueConnection和TopicConnection。

(4) Session

Session是我们操作消息的接口。可以通过session创建生产者、消费者、消息等。Session提供了事务的功能。当我们需要使用session发送/接收多个消息时，可以将这些发送/接收动作放到一个事务中。同样，也分QueueSession和TopicSession。

(5) 消息的生产者

消息生产者由Session创建，并用于将消息发送到Destination。同样，消息生产者分两种类型：QueueSender和TopicPublisher。可以调用消息生产者的方法（send或publish方法）发送消息。

(6) 消息消费者

消息消费者由Session创建，用于接收被发送到Destination的消息。两种类型：QueueReceiver和TopicSubscriber。可分别通过session的createReceiver(Queue)或createSubscriber(Topic)来创建。当然，也可以session的creatDurableSubscriber方法来创建持久化的订阅者。

(7) MessageListener

消息监听器。如果注册了消息监听器，一旦消息到达，将自动调用监听器的onMessage方法。EJB中的MDB（Message-Driven Bean）就是一种MessageListener。

代码:

package com.east.activemq;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    /**
     * 消息的生产者(发送者)
     * 
     * @author linhaiyun
     * @time 2017.12.12
     */
    public class JMSProducer {
    	// 默认连接用户名
    	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 int SENDNUM = 50;
    
    	public static void main(String[] args) throws Exception {
    		// TODO Auto-generated method stub
    		// 连接工厂
    		ConnectionFactory factory;
    		// 连接
    		Connection connection = null;
    		// 会话 接受或者发送消息的线程
    		Session session;
    		// 消息的目的地
    		Destination destination;
    		// 消息生产者
    		MessageProducer messageProducer;
    		// 实例化连接工厂
    		factory = new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);
    
    		try {
    			// 通过连接工厂获取连接
    			connection = factory.createConnection();
    			// 启动连接
    			connection.start();
    			// 创建session
    			session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
    			// 创建一个名称为HelloHaige的消息队列
    			destination = session.createQueue("HelloHaige");
    			// 创建消息生产者
    			messageProducer = session.createProducer(destination);
    			// 发送消息
    			sendMessage(session, messageProducer);
    			session.commit();
    		} catch (JMSException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		} finally {
    			if (connection != null) {
    				try {
    					connection.close();
    				} catch (JMSException e) {
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    
    	/**
    	 * 发送消息
    	 * 
    	 * @param session
    	 * @param messageProducer
    	 * @time 2017.12.12 12:10:25
    	 */
    	public static void sendMessage(Session session, MessageProducer messageProducer) throws Exception {
    		for (int i = 0; i < JMSProducer.SENDNUM; i++) {
    			// 创建一条文本消息
    			TextMessage message = session.createTextMessage("ActiveMQ 发送消息" + i);
    			System.out.println("发送消息：Activemq 发送消息" + i);
    			// 通过消息生产者发出消息
    			messageProducer.send(message);
    		}
    	}
    
    }

package com.east.activemq;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageConsumer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    /**
     * 消息的消费者(接受者)
     * 
     * @author linhaiyun
     * @time 2017.12.12
     */
    public class JMSConsumer {
    	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;// 默认连接地址
    
    	public static void main(String[] args) {
    		// TODO Auto-generated method stub
    		ConnectionFactory factory; // 连接工厂
    		Connection connection; // 连接
    
    		Session session; // 会话接受或者发送消息的线程
    		Destination destination; // 消息的目的地
    
    		MessageConsumer messageConsumer;// 消息的消费者
    
    		// 实例化连接工厂
    		factory = new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);
    		
    		try {
    			// 通过连接工厂获取连接
    			connection = factory.createConnection();
    			// 启动连接
    			connection.start();
    			// 创建session
    			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			// 创建一个连接HelloWorld的消息队列
    			destination = session.createQueue("HelloHaige");
    			// 创建消息消费者
    			messageConsumer = session.createConsumer(destination);
    			
    			while(true){
    				TextMessage message = (TextMessage) messageConsumer.receive(100000);
    				if(message !=null){
    					System.out.println("收到的消息:" + message.getText());
    				}else{
    					break;
    				}
    			}
    		} catch (JMSException e) {
    			// TODO Auto-generated catch block
    			System.out.println("实例化消费者失败!");
    			e.printStackTrace();
    		}
    		
    	}
    
    }

结果:

打开本地的ActiveMQ 服务

![Center 1][]

运行ActiveMQ生产者,创建50个消息,结果如图

![Center 2][]

运行消费者进行消息的接受消费

![Center 3][]

[Center]: /images/20220603/a911be18bbaa4d54a9717075c6a96498.png
[Center 1]: /images/20220603/3760b988a2b24ab5ad942e0ae0f55b90.png
[Center 2]: /images/20220603/8ac65693e42440908c7b3918541c2663.png
[Center 3]: /images/20220603/f186e43eb68e4137b919a232c93f03e1.png