ActiveMQ（二）Pub/Sub

朴灿烈づ我的快乐病毒、 2022-05-25 04:37 128阅读 0赞

另一种模式就是发布订阅：

![20180508101925396][]

订阅/发布的特点是：一条消息会被所有订阅主题的人所消费。

public class MqProducerPub {
    	public static void main(String[] args) throws JMSException {
    		// 第一步：创建ConnectionFactory连接工厂
    		ConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
    				ActiveMQConnectionFactory.DEFAULT_PASSWORD, ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
    		// 第二步：创建Connection
    		Connection connection = factory.createConnection();
    		// 第三步：创建Session
    		Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    		// 第四步：创建Destination
    		// Destination destination = session.createQueue("demo1");
    		// 第四部：创建Topic
    		Destination destination = session.createTopic("topic1");
    		// 第五步：创建MessageProducer
    		MessageProducer producer = session.createProducer(destination);
    		// 第六步：设置持久化方式
    		producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    		// 第七步：定义消息对象，并发送
    		TextMessage message = session.createTextMessage();
    		message.setText("gogogo！");
    		producer.send(message);
    		// 第八步：关闭连接
    		if (connection != null) {
    			connection.close();
    		}
    	}
    }

public class ConsumerSub {
    	private void consume() throws JMSException {
    		// 1、创建工厂连接对象
    		ConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
    		// 2、使用连接工厂创建一个连接对象
    		Connection connection = factory.createConnection();
    		// 3、开启连接
    		connection.start();
    		// 4、使用连接对象创建会话（session）对象
    		Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    		// 5、使用会话对象创建目标对象，包含queue和topic（一对一和一对多）
    		// Queue queue = session.createQueue("topic1");
    		Topic topic = session.createTopic("topic1");
    		// 6、使用会话对象创建生产者对象
    		MessageConsumer consumer = session.createConsumer(topic);
    		// 7、向MessageConsumer对象中设置一个MessageListener对象，用来接收消息
    		consumer.setMessageListener(new MessageListener() {
    			@Override
    			public void onMessage(Message message) {
    				if (message instanceof TextMessage) {
    					TextMessage textMessage = (TextMessage) message;
    					try {
    						System.out.println(textMessage.getText());
    					} catch (JMSException e) {
    						e.printStackTrace();
    					}
    				}
    			}
    		});
    		try {
    			Thread.sleep(30000);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		// 9、关闭资源
    		consumer.close();
    		session.close();
    		connection.close();
    		
    		
    	}
    	public static void main(String[] args) throws JMSException {
    		new Thread(()->{
    			try {
    				new ConsumerSub().consume();
    			} catch (JMSException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}).start();
    		new Thread(()->{
    			try {
    				new ConsumerSub().consume();
    			} catch (JMSException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}).start();
    	}
    }

INFO | Successfully connected to tcp://localhost:61616
     INFO | Successfully connected to tcp://localhost:61616
    gogogo！
    gogogo！

使用方式上有区别，结果也有区别，可以看到启动两个线程都收到了主题的消息。

注意：此类型要先启动消费者，让他处于监听状态，生产者在监听之前生产的数据是收不到的。

[20180508101925396]: /images/20220525/8b97853cf62e44bcb5e5c76a8f5f30a2.png