ActiveMQ实战(三)--ActiveMQ的通信方式之主题发布订阅式(publish-subscribe)

心已赠人 2022-06-12 12:07 349阅读 0赞

**一、简介**

发布/订阅方式用于多接收客户端的方式.作为发布订阅的方式，可能存在多个接收客户端，并且接收端客户端与发送客户端存在时间上的依赖。一个接收端只能接收他创建以后发送客户端发送的信息。作为subscriber ,在接收消息时有两种方法，destination的receive方法，和实现message listener 接口的onMessage 方法。

发布订阅模式有点类似于我们日常生活中订阅报纸。每年到年尾的时候，邮局就会发一本报纸集合让我们来选择订阅哪一个。在这个表里头列了所有出版发行的报纸，那么对于我们每一个订阅者来说，我们可以选择一份或者多份报纸。比如北京日报、潇湘晨报等。那么这些个我们订阅的报纸，就相当于发布订阅模式里的topic。有很多个人订阅报纸，也有人可能和我订阅了相同的报纸。那么，在这里，相当于我们在同一个topic里注册了。对于一份报纸发行方来说，它和所有的订阅者就构成了一个1对多的关系。这种关系如下图所示：

![20170722212708095][]

注意: Subscriber只有在订阅了主题Topic后，才能接收到Publisher发送的消息。

**二、实战主题发布订阅式通信实战**

首先我们必须创建的是publisher,然后才是subscribe:

1）Publisher

创建两个消息订阅者:

JMSConsumer1

JMSConsumer2

package com.fendo.mq.publishsubscribe;
    
    
    
    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 org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    /**
     * 消息消费者-消息订阅者二
     * @author fendo
     *
     */
    public class JMSConsumer2 {
    
    	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) {
    		ConnectionFactory connectionFactory; // 连接工厂
    		Connection connection = null; // 连接
    		Session session; // 会话 接受或者发送消息的线程
    		Destination destination; // 消息的目的地
    		MessageConsumer messageConsumer; // 消息的消费者
    		
    		// 实例化连接工厂
    		connectionFactory=new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD, JMSConsumer2.BROKEURL);
    				
    		try {
    			connection=connectionFactory.createConnection();  // 通过连接工厂获取连接
    			connection.start(); // 启动连接
    			session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session
    			destination=session.createTopic("FirstTopic1");
    			messageConsumer=session.createConsumer(destination); // 创建消息消费者
    			messageConsumer.setMessageListener(new Listener2()); // 注册消息监听
    		} catch (JMSException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		} 
    	}
    }

2）subscribe

JMSProducer

package com.fendo.mq.publishsubscribe;
    
    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 fendo
     *
     */
    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=10; // 发送的消息数量
    	
    	public static void main(String[] args) {
    		
    		ConnectionFactory connectionFactory; // 连接工厂
    		Connection connection = null; // 连接
    		Session session; // 会话 接受或者发送消息的线程
    		Destination destination; // 消息的目的地
    		MessageProducer messageProducer; // 消息生产者
    		
    		// 实例化连接工厂
    		connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);
    		
    		try {
    			connection=connectionFactory.createConnection(); // 通过连接工厂获取连接
    			connection.start(); // 启动连接
    			session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 创建Session
    			
    			destination=session.createTopic("FirstTopic1"); //创建消息发布者
    			messageProducer=session.createProducer(destination); // 创建消息生产者
    			sendMessage(session, messageProducer); // 发送消息
    			session.commit();
    		} catch (Exception e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		} finally{
    			if(connection!=null){
    				try {
    					connection.close();
    				} catch (JMSException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    	
    	/**
    	 * 发送消息
    	 * @param session
    	 * @param messageProducer
    	 * @throws Exception
    	 */
    	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);
    		}
    	}
    }

3）消息监听者

Listener1

package com.fendo.mq.publishsubscribe;
    
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    /**
     * 消息监听-订阅者一
     * @author fendo
     *
     */
    public class Listener1 implements MessageListener{
    
    	@Override
    	public void onMessage(Message message) {
    		try {
    			System.out.println("订阅者一收到的消息："+((TextMessage)message).getText());
    		} catch (JMSException e) {
    			e.printStackTrace();
    		}
    	}
    
    }

Listener2

package com.fendo.mq.publishsubscribe;
    
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    /**
     * 消息监听-订阅者二
     * @author fendo
     *
     */
    public class Listener2 implements MessageListener{
    
    	@Override
    	public void onMessage(Message message) {
    		try {
    			System.out.println("订阅者二收到的消息："+((TextMessage)message).getText());
    		} catch (JMSException e) {
    			e.printStackTrace();
    		}
    	}
    
    }

**三、测试**

首先运行两个主题订阅者,运行之后,在ActiveMQ控制台中可以看到一些信息

![20170722214137124][]

"Number Of Consumers" 消费者的数量为2,然后运行发布者,再次查看页面会变成如下

![20170722214744676][]

完整示例如下: [http://download.csdn.net/detail/u011781521/9907601][http_download.csdn.net_detail_u011781521_9907601]

[20170722212708095]: /images/20220612/b6193a53f5ec4a598d90ad04c76fb068.png
[20170722214137124]: /images/20220612/4bfa34234ca5482eb0751f1f171f02b8.png
[20170722214744676]: /images/20220612/d180b871553044fb807db9c412f56577.png
[http_download.csdn.net_detail_u011781521_9907601]: http://download.csdn.net/detail/u011781521/9907601