ActiveMQ采用61616端口提供JMS服务
ActiveMQ采用8161端口提供管理控制台服务
二、JMS总体架构Java Message Service
粗说目的地Destination队列(Queue)和主题(Topic) 三、JMS开发基本步骤JMS开发的基本步骤
1：创建一个connection factory
2：通过connection factory来创建JMS connection
3：启动JMS connection
4：通过JMS connection创建JMS session
5：创建JMS destination（目的地 队列/主题）
6：创建JMS producer或者创建JMS consume并设置destination
7：创建JMS consumer或者注册一个JMS message listener
8：发送(send)或者接收(receive)JMS message
9：关闭所有JMS资源
四、Queue队列模式 消息生产者代码public class JmsProduce {
public static final String ACTIVEMQ_URL = "tcp://192.168.16.106:61616";
public static final String QUEUE_NAME = "queue01";
public static void main(String[] args) throws JMSException {
//1.创建连接工厂，按照给定的url地址，采用默认用户名和密码
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
//2.获得连接connection并启动
Connection connection = factory.createConnection();
connection.start();
//3.创建会话session
//两个参数：①事务 ②签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4.创建目的地（具体是队列还是主题topic）
Queue queue = session.createQueue(QUEUE_NAME);
//5.创建消息的生产者
MessageProducer messageProducer = session.createProducer(queue);
//6.通过使用messageProducer生产三个消息发送到MQ的队列
for(int i=0;i<3;i++){
//7.创建消息
//text类型
TextMessage textMessage = session.createTextMessage("mag---" + i);
//8.通过messageProducer发送给mq
messageProducer.send(textMessage);
}
//9.关闭资源
messageProducer.close();
session.close();
connection.close();
System.out.println("---消息发布到mq---");
}
消费者代码一（阻塞式消费者）
public class JmsConsumer {
public static final String ACTIVEMQ_URL = "tcp://192.168.16.106:61616";
public static final String QUEUE_NAME = "queue01";
public static void main(String[] args) throws JMSException, IOException {
//1.创建连接工厂，按照给定的url地址，采用默认用户名和密码
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
//2.获得连接connection并启动
Connection connection = factory.createConnection();
connection.start();
//3.创建会话session
//两个参数：①事务 ②签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4.创建目的地（具体是队列还是主题topic）
Queue queue = session.createQueue(QUEUE_NAME);
//5.创建消费者
MessageConsumer messageConsumer = session.createConsumer(queue);
while (true){
TextMessage testMessage =(TextMessage) messageConsumer.receive();
if(testMessage!=null){
System.out.println("**消费者接受到信息："+testMessage.getText());
}else {
break;
}
}
}
messageConsumer.close();
session.close();
connection.close();
}
消费者代码二（异步监听式消费者）
public class JmsConsumer {
public static final String ACTIVEMQ_URL = "tcp://192.168.16.106:61616";
public static final String QUEUE_NAME = "queue01";
public static void main(String[] args) throws JMSException, IOException {
//1.创建连接工厂，按照给定的url地址，采用默认用户名和密码
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
//2.获得连接connection并启动
Connection connection = factory.createConnection();
connection.start();
//3.创建会话session
//两个参数：①事务 ②签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4.创建目的地（具体是队列还是主题topic）
Queue queue = session.createQueue(QUEUE_NAME);
//5.创建消费者
MessageConsumer messageConsumer = session.createConsumer(queue);
messageConsumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
if(null != message && message instanceof TextMessage){
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("接收到text消息："+(textMessage.getText()));
} catch (JMSException e) {
e.printStackTrace();
}
}
}
});
//保证控制台不灭 不加的话消息没来得及处理程序就结束了
System.in.read();
messageConsumer.close();
session.close();
connection.close();
}
}
控制台说明
Number Of Pending Messages=等待消费的消息，这个是未出队列的数量，公式=总接收数-总出队列数。 Number Of Consumers=消费者数量，消费者端的消费者数量。 Messages Enqueued=进队消息数，进队列的总消息量，包括出队列的。这个数只增不减。 Messages Dequeued=出队消息数，可以理解为是消费者消费掉的数量。 总结： 当有一个消息进入这个队列时，等待消费的消息是1，进入队列的消息是1。 当消息消费后，等待消费的消息是0，进入队列的消息是1，出队列的消息是1。 当再来一条消息时，等待消费的消息是1，进入队列的消息就是2。两种消费方式比较
同步阻塞方式
订阅者或接收者抵用MessageConsumer的receive()方法来接收消息，receive方法在能接收到消息之前（或超时之前）将一直阻塞。
异步非阻塞方式（监听器onMessage()）
订阅者或接收者通过MessageConsumer的setMessageListener(MessageListener listener)注册一个消息监听器，
当消息到达之后，系统会自动调用监听器MessageListener的onMessage(Message message)方法。
五、Topic队列模式发布/订阅消息传递域的特点如下：
（1）生产者将消息发布到topic中，每个消息可以有多个消费者，属于1：N的关系；
（2）生产者和消费者之间有时间上的相关性。订阅某一个主题的消费者只能消费自它订阅之后发布的消息。
（3）生产者生产时，topic不保存消息它是无状态的不落地，假如无人订阅就去生产，那就是一条废消息，所以，一般先启动消费者再启动生产者。
JMS规范允许客户创建持久订阅，这在一定程度上放松了时间上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时发送的消息。一句话，好比我们的微信公众号订阅
生产者代码public class JmsProduce {
public static final String ACTIVEMQ_URL = "tcp://192.168.16.106:61616";
public static final String TOPIC_NAME = "topic-atguigu";
public static void main(String[] args) throws JMSException {
//1.创建连接工厂，按照给定的url地址，采用默认用户名和密码
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
//2.获得连接connection并启动
Connection connection = factory.createConnection();
connection.start();
//3.创建会话session
//两个参数：①事务 ②签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4.创建目的地（具体是队列还是主题topic）
Topic topic = session.createTopic(TOPIC_NAME);
//5.创建消息的生产者
MessageProducer messageProducer = session.createProducer(topic);
//6.通过使用messageProducer生产三个消息发送到MQ的队列
for(int i=0;i<3;i++){
//7.创建消息
TextMessage textMessage = session.createTextMessage("TOPIC_NAME---" + i);
//8.通过messageProducer发送给mq
messageProducer.send(textMessage);
}
//9.关闭资源
messageProducer.close();
session.close();
connection.close();
System.out.println("---TOPIC_NAME消息发布到mq---");
}
}
消费者代码
public class JmsConsumer {
public static final String ACTIVEMQ_URL = "tcp://192.168.16.106:61616";
public static final String TOPIC_NAME = "topic-atguigu";
public static void main(String[] args) throws JMSException, IOException {
System.out.println("----------消费者1---------------");
//1.创建连接工厂，按照给定的url地址，采用默认用户名和密码
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
//2.获得连接connection并启动
Connection connection = factory.createConnection();
connection.start();
//3.创建会话session
//两个参数：①事务 ②签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4.创建目的地（具体是队列还是主题topic）
Topic topic = session.createTopic(TOPIC_NAME);
//5.创建消费者
MessageConsumer messageConsumer = session.createConsumer(topic);
messageConsumer.setMessageListener(message -> {
if(null != message && message instanceof TextMessage){
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("接收到消息："+((TextMessage) message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
System.in.read();//保证控制台不灭 不加的话消息没来得及处理程序就结束了
messageConsumer.close();
session.close();
connection.close();
}
}
六、两种模式比较
| 比较项目 | 工作模式 | 有无状态 | 传递完整性 | 处理效率 |
| Queue队列 | "负载均衡"模式，如果当前没有消费者，消息也不会丢弃；如果有多个消费者，那么一条消息也只会发送给其中一个消费者，并且要求消费者ack信息 | Queue数据默认会在mq服务器上已文件形式保存，比如Active MQ一般保存在$AMQ_HOMEdatakr-storedata下面，也可以配置成DB存储 | 消息不会被丢弃 | 由于一条消息只发送给一个消费者，所以就算消费者再多，性能也不会有明显降低。当然不同消息协议的具体性能也是有差异的 |
| 模式Topic队列模式 | "订阅-发布"模式，如果当前没有订阅者，消息将会被丢弃，如果有多个订阅者，那么这些订阅者都会收到消息 | 无状态 | 如果没有订阅者，消息会被丢弃 | 由于消息要按照订阅者的数量进行复制，所以处理性能会随着订阅者的增加而明显降低，并且还要结合不同消息协议自身的性能差异 |
