当前位置：首页 > news >正文

从 JMS 到 ActiveMQ：API 设计与扩展机制分析（二）

news 来源：原创 2025/8/6 21:04:34

（三）消息结构与 API 操作

JMS 消息结构：JMS 消息主要由消息头（Header）、属性（Properties）和消息体（Body）三部分组成。消息头包含了许多预定义的字段，用于标识消息、设置优先权、失效时间等，例如 JMSMessageID 用于唯一标识一条消息，JMSDeliveryMode 用于指定消息的投递模式（持久或非持久），JMSPriority 用于设置消息的优先级，JMSTimestamp 用于记录消息发送的时间，JMSExpiration 用于设置消息的过期时间等。这些字段由 JMS Provider 在消息发送时自动填充或根据发送者的设置进行配置。属性部分允许开发者自定义一些键值对，用于添加额外的信息，比如消息的业务类型、处理标志等，这些属性可以在消息的发送和接收过程中被使用，方便对消息进行分类和处理。消息体则是消息的实际内容，根据不同的消息类型，消息体的结构和内容也不同，JMS 提供了多种消息类型，如 TextMessage 用于发送文本消息，MapMessage 用于发送键值对形式的消息，BytesMessage 用于发送字节数组消息，StreamMessage 用于发送 Java 原始类型的数据流消息，ObjectMessage 用于发送可序列化的 Java 对象消息。

API 操作示例：通过代码示例展示创建、发送、接收不同类型消息的操作。

创建并发送 TextMessage

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 TextMessageProducer {

public static void main(String[] args) throws Exception {

ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");

Connection connection = factory.createConnection();

connection.start();

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

Queue queue = session.createQueue("textQueue");

MessageProducer producer = session.createProducer(queue);

TextMessage message = session.createTextMessage("This is a text message");

// 设置自定义属性

message.setStringProperty("messageType", "text");

producer.send(message);

System.out.println("Sent text message: " + message.getText());

producer.close();

session.close();

connection.close();

}

接收 TextMessage

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

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 TextMessageConsumer {

public static void main(String[] args) throws Exception {

ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");

Connection connection = factory.createConnection();

connection.start();

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

Queue queue = session.createQueue("textQueue");

MessageConsumer consumer = session.createConsumer(queue);

Message message = consumer.receive();

if (message instanceof TextMessage) {

TextMessage textMessage = (TextMessage) message;

System.out.println("Received text message: " + textMessage.getText());

// 获取自定义属性

System.out.println("Message type: " + textMessage.getStringProperty("messageType"));

}

consumer.close();

session.close();

connection.close();

}

创建并发送 MapMessage

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

import javax.jms.MessageProducer;

import javax.jms.Queue;

import javax.jms.Session;

import javax.jms.MapMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class MapMessageProducer {

public static void main(String[] args) throws Exception {

ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");

Connection connection = factory.createConnection();

connection.start();

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

Queue queue = session.createQueue("mapQueue");

MessageProducer producer = session.createProducer(queue);

MapMessage message = session.createMapMessage();

message.setString("key1", "value1");

message.setInt("key2", 123);

// 设置自定义属性

message.setStringProperty("messageType", "map");

producer.send(message);

System.out.println("Sent map message");

producer.close();

session.close();

connection.close();

}

接收 MapMessage

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

import javax.jms.Message;

import javax.jms.MessageConsumer;

import javax.jms.Queue;

import javax.jms.Session;

import javax.jms.MapMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class MapMessageConsumer {

public static void main(String[] args) throws Exception {

ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");

Connection connection = factory.createConnection();

connection.start();

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

Queue queue = session.createQueue("mapQueue");

MessageConsumer consumer = session.createConsumer(queue);

Message message = consumer.receive();

if (message instanceof MapMessage) {

MapMessage mapMessage = (MapMessage) message;

System.out.println("Received map message");

System.out.println("Value of key1: " + mapMessage.getString("key1"));

System.out.println("Value of key2: " + mapMessage.getInt("key2"));

// 获取自定义属性

System.out.println("Message type: " + mapMessage.getStringProperty("messageType"));

}

consumer.close();

session.close();

connection.close();

}

消息结构在 API 操作中的作用至关重要。消息头中的字段为消息的传输和处理提供了基本的控制信息，例如通过设置 JMSPriority 可以让重要的消息优先被处理，设置 JMSExpiration 可以避免消息长时间占用资源。属性部分则增强了消息的灵活性和可扩展性，开发者可以根据业务需求添加各种自定义信息，方便在消息处理过程中进行判断和操作。消息体则承载了实际的业务数据，不同的消息类型适用于不同的业务场景，例如 TextMessage 适合传输文本内容，MapMessage 适合传输结构化的键值对数据。

（三）消息结构与 API 操作

创建并发送 TextMessage

接收 TextMessage

创建并发送 MapMessage

接收 MapMessage

相关文章：