目录
一、IDEA建Maven工程
二、pom.xml
三、JMS编码总体架构
四、粗说目的地Destination——队列和主题
五、点对点消息传递域中，目的地被称为队列（queue）
六、在发布订阅消息传递域中，目的地被称为主题（topic）
七、topic和queue对比
一、IDEA建Maven工程
二、pom.xml
<dependencies>
<!-- activemq 所需要的jar 包-->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.15.9</version>
</dependency>
<!-- activemq 和 spring 整合的基础包 -->
<dependency>
<groupId>org.apache.xbean</groupId>
<artifactId>xbean-spring</artifactId>
<version>3.16</version>
</dependency>
<!--基础通用配置-->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.30</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.2.11</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.16.16</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>三、JMS（Java平台上的专业技术规范）编码总体架构
（1）回顾JDBC
（2）JMS（Java平台上的专业技术规范）开发基本步骤
四、粗说目的地Destination——队列和主题
Destination是目的地。下面拿jvm和mq，做个对比。目的地，我们可以理解为是数据存储的地方。
Destination分为两种：队列和主题。
（1） 队列（queue——1对1）
（2）主题（topic——1对N）
五、点对点消息传递域中，目的地被称为队列（queue）
（1）消息生产者
public static final String ACTIVEMQ_URL = "tcp://虚拟机ip:61616";
public static final String QUEUE_NAME = "queue01";
public static void main(String[] args) throws JMSException {
//1.创建链接工厂,按照给定的url地址，采用默认的用户和密码
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
//2.通过链接工厂，获得Connection并启动
Connection connection = activeMQConnectionFactory.createConnection();
connection.start();
//3.创建会话session
//3.1 两个参数 ①事务 ②签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4.创建目的地（queue or topic）
Queue queue = session.createQueue(QUEUE_NAME);
//5.创建生产者
MessageProducer producer = session.createProducer(queue);
//6.通过MessageProducer生成3条消息到MQ队列中
for(int i = 1 ; i <= 3; i ++ ) {
TextMessage textMessage = session.createTextMessage("msg--" + i);
producer.send(textMessage);
}
producer.close();
session.close();
connection.close();
System.out.println("消息发布到MQ完成……");
}（2）控制台说明
回到ActiveMQ前端控制台可以看到如下界面，有三条消息
① Number Of Pending Messages：等待消费的消息，这个是未出队列的数量，公式=总接收数-总出队列数。
② Number Of Consumers：消费者数量，消费者端的消费者数量。
③ Messages Enqueued：进队消息数，进队列的总消息量，包括出队列的。这个数只增不减。
④ Messages Dequeued：出队消息数，可以理解为是消费者消费掉的数量。
总结：
当有一个消息进入这个队列时，等待消费的消息是1，进入队列的消息是1。
当消息消费后，等待消费的消息是0，进入队列的消息是1，出队列的消息是1。
当再来一条消息时，等待消费的消息是1，进入队列的消息就是2。
（3）消息消费者
① 方式一
public static final String ACTIVEMQ_URL = "tcp://虚拟机ip:61616";
public static final String QUEUE_NAME = "queue01";
public static void main(String[] args) throws JMSException {
//1.创建链接工厂,按照给定的url地址，采用默认的用户和密码
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
//2.通过链接工厂，获得Connection并启动
Connection connection = activeMQConnectionFactory.createConnection();
connection.start();
//3.创建会话session
//3.1 两个参数 ①事务 ②签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4.创建目的地（queue or topic）
Queue queue = session.createQueue(QUEUE_NAME);
//5.创建消费者
MessageConsumer consumer = session.createConsumer(queue);
while(true) {
//2000ms timeout时间
//ActiveMQ有五种消息类型，这里发送时是TextMessage，接收时强转一下
/*同步阻赛方式(receive())
订阅者或接收者调Messaoeconsumerreceive()方法来接收消息，receive 方法在能够接收到消息之前(或超时之前)将一直阻塞*/
TextMessage textMessage = (TextMessage) consumer.receive(2000);
if(null != textMessage) {
System.out.println("接收到消息：" + textMessage.getText());
} else {
break;
}
}
consumer.close();
session.close();
connection.close();
}
注意：MessageConsumer的receive方法如果选择的是无参类型的，即不设置超时时间（Timeout），那么上图的Number Of Consumers数量为1，因为消息消费完后receive方法会一直等待，程序一直运行，Consumer也就一直在线
② 方式二
public static void main(String[] args) throws JMSException, IOException {
//1.创建链接工厂,按照给定的url地址，采用默认的用户和密码
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
//2.通过链接工厂，获得Connection并启动
Connection connection = activeMQConnectionFactory.createConnection();
connection.start();
//3.创建会话session
//3.1 两个参数 ①事务 ②签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4.创建目的地（queue or topic）
Queue queue = session.createQueue(QUEUE_NAME);
//5.创建消费者
MessageConsumer consumer = session.createConsumer(queue);
//通过监听方式读取消息
/*
异步非阻塞方式(监听器onMessage())
订阅者或接收者通过MessageConsumer的setMessageListener(MessageListener
listener)注册一个消息监听器当消息到达之后，系统自动调用监听器MessageListener的
onMessage(Message message)方法。
*/
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
if(null != message && message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("监听到消息："+textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
});
//连接AvtiveMQ需要等待时间
System.in.read();
consumer.close();
session.close();
connection.close();
} 
③ 消费者情况分析
- 一个消费者
- 两个消费者
- 先生成消息，先启动1号，在启动2号 结果：1号消费完所有消息，2号等待
- 先启动1，2号，再生成消息 结果：1，2号均分了消息，颇有些负载均衡的意味
④ 小总结
一、连接的基本步骤
二、两种消费方式
① 同步阻塞方式(receive)
订阅者或接收者抵用MessageConsumer的receive()方法来接收消息，receive方法在能接收到消息之前（或超时之前）将一直阻塞。
② 异步非阻塞方式（监听器onMessage()）
订阅者或接收者通过MessageConsumer的setMessageListener(MessageListener listener)注册一个消息监听器，当消息到达之后，系统会自动调用监听器MessageListener的onMessage(Message message)方法。
三、队列特点
六、在发布订阅消息传递域中，目的地被称为主题（topic）
（1）topic介绍
在发布订阅消息传递域中，目的地被称为主题（topic）
发布/订阅消息传递域的特点如下：
① 生产者将消息发布到topic中，每个消息可以有多个消费者，属于1：N的关系；
② 生产者和消费者之间有时间上的相关性。订阅某一个主题的消费者只能消费自它订阅之后发布的消息。
③ 生产者生产时，topic不保存消息它是无状态的不落地，假如无人订阅就去生产，那就是一条废消息，所以，一般先启动消费者再启动生产者。
默认情况下如上所述，但是JMS规范允许客户创建持久订阅，这在一定程度上放松了时间上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时发送的消息。一句话，好比我们的微信公众号订阅。
（2）发布主题生产者
public static final String ACTIVEMQ_URL = "tcp://虚拟机ip:61616";
public static final String TOPIC_NAME = "topic01";
public static void main(String[] args) throws JMSException {
//1.创建链接工厂,按照给定的url地址，采用默认的用户和密码
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
//2.通过链接工厂，获得Connection并启动
Connection connection = activeMQConnectionFactory.createConnection();
connection.start();
//3.创建会话session
//3.1 两个参数 ①事务 ②签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4.创建目的地（queue or topic）
Topic topic = session.createTopic(TOPIC_NAME);
//5.创建生产者
MessageProducer producer = session.createProducer(topic);
//6.通过MessageProducer生成3条消息到MQ队列中
for(int i = 1 ; i <= 3; i ++ ) {
TextMessage textMessage = session.createTextMessage("topicMessage--" + i);
producer.send(textMessage);
}
producer.close();
session.close();
connection.close();
System.out.println("topicMessage消息发布到MQ完成……");
}（3）订阅主题消费者
public static final String ACTIVEMQ_URL = "tcp://虚拟机ip:61616";
public static final String TOPIC_NAME = "topic01";
public static void main(String[] args) throws JMSException, IOException {
//1.创建链接工厂,按照给定的url地址，采用默认的用户和密码
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
//2.通过链接工厂，获得Connection并启动
Connection connection = activeMQConnectionFactory.createConnection();
connection.start();
//3.创建会话session
//3.1 两个参数 ①事务 ②签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4.创建目的地（queue or topic）
Topic topic = session.createTopic(TOPIC_NAME);
//5.创建消费者
MessageConsumer consumer = session.createConsumer(topic);
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
if(null != message && message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("监听到消息："+textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
});
//连接AvtiveMQ需要等待时间
System.in.read();
consumer.close();
session.close();
connection.close();
}（4）先启动消费者再启动生产者
① 启动消费者
② 启动生产者
③ 消费消息
④ 控制台
七、topic和queue对比
