ActiveMQ入门教程(四) - ActiveMQ Pub/Sub版的HelloWorld

本文主要是介绍ActiveMQ入门教程(四) - ActiveMQ Pub/Sub版的HelloWorld，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

为什么80%的码农都做不了架构师？>>>

在上一篇中，我们说了P2P版的HelloWorld，在这一篇，我们简要说一下，基于发布，订阅模式的HelloWorld。

基础知识就不在介绍了，需要的会一点一点讲。

1. pom.xml

这个和上一篇是一样的：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.ygy</groupId><artifactId>activemq</artifactId><version>0.0.1-SNAPSHOT</version><packaging>jar</packaging><name>activemq</name><url>http://maven.apache.org</url><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><dependencies><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.10</version><scope>test</scope></dependency><!-- activemq,学习中 --><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-core</artifactId><version>5.7.0</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.5.6</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.5.6</version></dependency></dependencies>
</project>

2. Pub/Sub版的HelloWorld

生产者：

package org.ygy.mq.lesson01;import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnectionFactory;
import org.ygy.mq.constants.MQConstants;public class HelloTopicProducer {public void send(String msg) {// 生产者的主要流程Connection connection = null;try {// 1.初始化connection工厂,使用默认的URL// failover://tcp://localhost:61616ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();// 2.创建Connectionconnection = connectionFactory.createConnection();// 3.打开连接connection.start();// 4.创建Session，(是否支持事务)Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 5.创建消息目标Destination destination_send = session.createTopic(MQConstants.DESTINATION_SEND);// 6.创建生产者MessageProducer producer = session.createProducer(destination_send);// 7.配置消息是否持久化/** DeliverMode有2种方式：* * public interface DeliveryMode { static final int NON_PERSISTENT =* 1;//不持久化：服务器重启之后，消息销毁* * static final int PERSISTENT = 2;//持久化：服务器重启之后，该消息仍存在 }*/producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);// 8.初始化要发送的消息TextMessage message = session.createTextMessage(msg);// 9.发送消息producer.send(message);connection.close();} catch (JMSException e) {e.printStackTrace();}}public static void main(String[] args) {new HelloTopicProducer().send("我来试一试发布/订阅...");}}

消费者：

package org.ygy.mq.lesson01;import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
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;
import org.ygy.mq.constants.MQConstants;public class HelloTopicConsumer implements MessageListener {@Overridepublic void onMessage(Message message) {if (message instanceof TextMessage) {TextMessage txtMsg = (TextMessage) message;try {System.out.println("哈，我接收到了消息：" + txtMsg.getText());} catch (JMSException e) {e.printStackTrace();}}}public void receive() {// 消费者的主要流程Connection connection = null;try {// 1.初始化connection工厂ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();// 2.创建Connectionconnection = connectionFactory.createConnection();// 3.打开连接connection.start();// 4.创建sessionSession session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 5.创建消息目标Destination destination = session.createTopic(MQConstants.DESTINATION_SEND);// 6.创建消费者MessageConsumer consumer = session.createConsumer(destination);// 7.配置监听consumer.setMessageListener(new HelloTopicConsumer());} catch (JMSException e) {e.printStackTrace();}}public static void main(String[] args) {new HelloTopicConsumer().receive();}}