ActiveMQ的发布订阅模式

朴灿烈づ我的快乐病毒、 2022-07-15 04:12 238阅读 0赞

1、ActivaMQ的发布订阅模式的简介

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

![20161101210201412][]

2、Publisher的代码

import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MapMessage;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Publisher {
    
    	// 单例模式
    
    	// 1、连接工厂
    	private ConnectionFactory connectionFactory;
    	// 2、连接对象
    	private Connection connection;
    	// 3、Session对象
    	private Session session;
    	// 4、生产者
    	private MessageProducer messageProducer;
    
    	public Publisher() {
    
    		try {
    			this.connectionFactory = new ActiveMQConnectionFactory("zhangsan",
    					"123", "tcp://localhost:61616");
    			this.connection = connectionFactory.createConnection();
    			this.connection.start();
    			// 不使用事务
    			// 设置客户端签收模式
    			this.session = this.connection.createSession(false,
    					Session.AUTO_ACKNOWLEDGE);
    			this.messageProducer = this.session.createProducer(null);
    		} catch (JMSException e) {
    			throw new RuntimeException(e);
    		}
    
    	}
    
    	public Session getSession() {
    		return this.session;
    	}
    
    	public void send1(/* String QueueName, Message message */) {
    		try {
    
    			Destination destination = this.session.createTopic("topic1");
    			MapMessage msg1 = this.session.createMapMessage();
    			msg1.setString("name", "张三");
    			msg1.setInt("age", 22);
    
    			MapMessage msg2 = this.session.createMapMessage();
    			msg2.setString("name", "李四");
    			msg2.setInt("age", 25);
    
    			MapMessage msg3 = this.session.createMapMessage();
    			msg3.setString("name", "张三");
    			msg3.setInt("age", 30);
    
    			// 发送消息到topic1
    			this.messageProducer.send(destination, msg1,
    					DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
    			this.messageProducer.send(destination, msg2,
    					DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
    			this.messageProducer.send(destination, msg3,
    					DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
    
    		} catch (JMSException e) {
    			throw new RuntimeException(e);
    		}
    	}
    
    	public void send2() {
    		try {
    			Destination destination = this.session.createTopic("topic1");
    			TextMessage message = this.session.createTextMessage("我是一个字符串");
    			// 发送消息
    			this.messageProducer.send(destination, message,
    					DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
    		} catch (JMSException e) {
    			throw new RuntimeException(e);
    		}
    
    	}
    
    	public static void main(String[] args) {
    		Publisher producer = new Publisher();
    		producer.send1();
    
    	}
    
    }

3、Subscribe的代码

import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MapMessage;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Subscriber {
    
    	// 单例模式
    
    	// 1、连接工厂
    	private ConnectionFactory connectionFactory;
    	// 2、连接对象
    	private Connection connection;
    	// 3、Session对象
    	private Session session;
    	// 4、生产者
    	private MessageConsumer messageConsumer;
    	// 5、目的地址
    	private Destination destination;
    
    	public Subscriber() {
    
    		try {
    			this.connectionFactory = new ActiveMQConnectionFactory("zhangsan",
    					"123", "tcp://localhost:61616");
    			this.connection = connectionFactory.createConnection();
    			this.connection.start();
    			// 不使用事务
    			// 设置客户端签收模式
    			this.session = this.connection.createSession(false,
    					Session.AUTO_ACKNOWLEDGE);
    			this.destination = this.session.createTopic("topic1");
    			this.messageConsumer = this.session.createConsumer(destination);
    		} catch (JMSException e) {
    			throw new RuntimeException(e);
    		}
    
    	}
    
    	public Session getSession() {
    		return this.session;
    	}
    
    	// 用于监听消息队列的消息
    	class MyLister implements MessageListener {
    
    		@Override
    		public void onMessage(Message message) {
    			try {
    				if (message instanceof TextMessage) {
    
    				}
    				if (message instanceof MapMessage) {
    					MapMessage ret = (MapMessage) message;
    					System.out.println(ret.toString());
    					System.out.println(ret.getString("name"));
    					System.out.println(ret.getInt("age"));
    					// 因为设置的是客户端的签收模式，所以要手动的去确认消息的消费
    					message.acknowledge();
    				}
    			} catch (JMSException e) {
    				throw new RuntimeException(e);
    			}
    		}
    
    	}
    
    	// 用于异步监听消息
    	public void receiver() {
    		try {
    			this.messageConsumer.setMessageListener(new MyLister());
    		} catch (JMSException e) {
    			throw new RuntimeException(e);
    		}
    	}
    
    	public static void main(String[] args) {
    		Subscriber conmuser = new Subscriber();
    		conmuser.receiver();
    
    	}
    
    }

4、ActiveMQ的界面截图

![20161101211938816][]

[20161101210201412]: /images/20220715/9abd18fbaf0a4b5b9aa64876e03e4910.png
[20161101211938816]: /images/20220715/624bf3d7f58a4522846391c1149e2590.png