Java一分钟之-JMS：Java消息服务

Jimaks

发布于 2024-06-13 10:39:07

870

发布于 2024-06-13 10:39:07

文章被收录于专栏：大数据大数据

在现代企业应用中，组件间的解耦与异步通信至关重要，而Java消息服务（Java Message Service，简称JMS）正是为此而生。JMS提供了一套标准的API，允许应用程序创建、发送、接收和读取消息。本文将带你快速了解JMS的基本概念、常见问题、易错点及如何避免，并通过一个简单的代码示例加深理解。

JMS基础

JMS定义了两种消息模型：点对点（Point-to-Point, P2P）和发布/订阅（Publish/Subscribe, Pub/Sub）。在P2P模型中，消息从一个生产者发送到一个特定的队列，然后由一个或多个消费者接收。而在Pub/Sub模型中，消息被发布到一个主题，所有订阅了该主题的消费者都能收到消息。

常见问题与易错点

1. 混淆消息模型

开发者常混淆P2P与Pub/Sub模型，导致消息传递逻辑错误。例如，在需要广播信息的场景下误用了P2P模型，导致消息只能被单个消费者接收。

避免方法：明确业务需求，选择合适的消息模型。若需一对多通信，应采用Pub/Sub模型；若需一对一且确保消息被消费，则选择P2P模型。

2. 忽略事务管理

未正确处理事务可能导致消息丢失或重复消费。例如，生产者发送消息后系统崩溃，但消息已被发送，导致消息状态不一致。

避免方法：利用JMS的事务特性或ACK机制保证消息的可靠传输。确保在业务逻辑成功执行后才提交事务或确认消息。

3. 资源泄露

忘记关闭连接、会话或消息生产者/消费者，会导致资源无法释放，长期运行可能耗尽系统资源。

避免方法：使用try-with-resources语句或在finally块中确保资源被正确关闭。

示例代码

下面是一个使用ActiveMQ作为消息代理，实现P2P模型的简单示例。

生产者代码

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

public class JMSProducer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        try (Connection connection = factory.createConnection();
             Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             MessageProducer producer = session.createProducer(session.createQueue("MyQueue"))) {
            connection.start();
            TextMessage message = session.createTextMessage("Hello, JMS!");
            producer.send(message);
            System.out.println("Message sent.");
        }
    }
}

消费者代码

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

public class JMSConsumer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        try (Connection connection = factory.createConnection();
             Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             MessageConsumer consumer = session.createConsumer(session.createQueue("MyQueue"))) {
            connection.start();
            while (true) {
                Message message = consumer.receive();
                if (message instanceof TextMessage) {
                    TextMessage textMessage = (TextMessage) message;
                    System.out.println("Received: " + textMessage.getText());
                } else {
                    break;
                }
            }
        } catch (JMSException e) {
            System.err.println("Error receiving message: " + e.getMessage());
        }
    }
}