Stomp.py向ActiveMQ发消息时Java消费端报越界异常

问题场景 #

• 环境：Apache ActiveMQ Artemis 2.24，Python 3.10.4 + Stomp.py 8.0.1发送消息，Java 1.8.0_342接收消息
• 现象：Java生产者向Java消费者发消息正常，但Python生产者发消息后，Java消费端抛出索引越界异常

异常信息 #

Exception in thread "main" java.lang.IndexOutOfBoundsException: Error reading in simpleString, length=1864388729 is greater than readableBytes=10
    at org.apache.activemq.artemis.api.core.SimpleString.readSimpleString(SimpleString.java:183)
    at org.apache.activemq.artemis.api.core.SimpleString.readSimpleString(SimpleString.java:171)
    at org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper.readStringInternal(ChannelBufferWrapper.java:103)
    at org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper.readString(ChannelBufferWrapper.java:88)
    at org.fogbeam.experimental.bosworth.activemq.ActiveMQConsumerMain.main(ActiveMQConsumerMain.java:54)

相关代码 #

Java消费者代码 #

import org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.client.ActiveMQClient;
import org.apache.activemq.artemis.api.core.client.ClientConsumer;
import org.apache.activemq.artemis.api.core.client.ClientMessage;
import org.apache.activemq.artemis.api.core.client.ClientSession;
import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
import org.apache.activemq.artemis.api.core.client.ServerLocator;

public class ActiveMQConsumerMain 
{
    public static void main(String[] args) throws Exception
    {
        ServerLocator locator = ActiveMQClient.createServerLocator("tcp://172.16.1.141:61616");

        ClientSessionFactory factory =  locator.createSessionFactory();
        ClientSession session = factory.createSession( "username", "password", false, true, true, false, 4096 );

        try
        {
            session.createQueue("example", RoutingType.ANYCAST, "example", true);
        }
        catch( ActiveMQQueueExistsException amqe )
        {
            if( amqe.getMessage().contains("already exists" ))
            {
                System.out.println( "Queue already exists on server!" );
            }
            else
            {
                amqe.printStackTrace();
            }
        }
        
        ClientConsumer consumer =  session.createConsumer("example");

        session.start();
        
        while( true )
        {
            System.out.println( "Listening..." );
            ClientMessage msgReceived = consumer.receive();

            System.out.println("message = " + msgReceived.getBodyBuffer().readString());
            
            msgReceived.acknowledge();
        
            session.commit();
        }
    }
}

Java生产者代码 #

import org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.client.ActiveMQClient;
import org.apache.activemq.artemis.api.core.client.ClientMessage;
import org.apache.activemq.artemis.api.core.client.ClientProducer;
import org.apache.activemq.artemis.api.core.client.ClientSession;
import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
import org.apache.activemq.artemis.api.core.client.ServerLocator;

public class ActiveMQProducerMain 
{
    public static void main(String[] args) throws Exception
    {
        ServerLocator locator = ActiveMQClient.createServerLocator("tcp://172.16.1.141:61616");

        ClientSessionFactory factory =  locator.createSessionFactory();
        ClientSession session = factory.createSession( "username", "password", false, true, true, false, 4096 );
        
        ClientProducer producer = session.createProducer("example");
        ClientMessage message = session.createMessage(true);
        message.getBodyBuffer().writeString("Hello world!!!");

        try
        {
            session.createQueue("example", RoutingType.ANYCAST, "example", true);
        }
        catch( ActiveMQQueueExistsException amqe )
        {
            if( amqe.getMessage().contains("already exists" ))
            {
                System.out.println( "Queue already exists on server!" );
            }
            else
            {
                amqe.printStackTrace();
            }
        }

        producer.send(message); 
    }
}

Python生产者代码 #

import time
import stomp

def main():
    print( "Sending ActiveMQ message using STOMP client!\n" )
    
    conn = stomp.Connection( [('172.16.1.141', 61613)] )
    
    conn.connect( wait=True, headers={'consumerWindowSize': 0})
    
    conn.send(body='Hello Python World', destination='example')
    time.sleep(5)
    conn.disconnect()
    
    exit()
    
if __name__ == "__main__":
    main()

后续尝试 #

• 编辑1：Stomp.py消费端接收Python发送的消息正常，确认跨语言交互存在格式兼容问题
• 编辑2：Python消费端接收Java发送的消息内容为空，进一步验证格式不兼容
• 编辑3：改用readNullableSimpleString()方法，仍触发同类异常：

  java.lang.IndexOutOfBoundsException: Error reading in simpleString, length=1701604463 is greater than readableBytes=13
      at org.apache.activemq.artemis.api.core.SimpleString.readSimpleString(SimpleString.java:183)
      ...
  

• 编辑4：通过读取字节数组的方式成功获取消息内容：

  int bodySize = msgReceived.getBodySize();
  byte[] bytes = new byte[bodySize];
  msgReceived.getBodyBuffer().readBytes( bytes );
  System.out.println("message = " + new String( bytes ) );
  

问题根源 #

核心是协议差异导致的消息格式不匹配：

1. Java端使用Artemis原生核心协议，writeString()会按照Artemis自定义的SimpleString格式写入——先写4字节大端长度前缀（表示字符串UTF-8字节数），再写字符串字节
2. Python端用STOMP协议发送消息，Stomp.py默认将消息体作为原始字节传递，Broker转换为核心消息时不会添加SimpleString的长度前缀
3. Java消费者用readString()读取时，会把STOMP消息开头的字符串字节错误解析为4字节长度值（比如'H'的ASCII码72组合成的int为1864388729），该值远大于实际消息长度，触发IndexOutOfBoundsException
4. Python消费端读Java消息为空，是因为无法识别SimpleString的长度前缀，解析逻辑异常导致内容无法正常读取

优化方案 #

方案一：统一消息格式（推荐） #

修改Python代码，手动添加SimpleString格式的长度前缀：

import time
import stomp
import struct

def main():
    print("Sending ActiveMQ message using STOMP client!\n")
    
    conn = stomp.Connection([('172.16.1.141', 61613)])
    conn.connect(wait=True, headers={'consumerWindowSize': 0})
    
    body = 'Hello Python World'
    body_bytes = body.encode('utf-8')
    # 生成4字节大端长度前缀
    length_prefix = struct.pack('>I', len(body_bytes))
    # 拼接前缀与消息体
    message_body = length_prefix + body_bytes
    
    conn.send(body=message_body, destination='example')
    time.sleep(5)
    conn.disconnect()

if __name__ == "__main__":
    main()

修改后Java消费者可直接用readString()正常读取。

方案二：Java端兼容处理 #

保持Python代码不变，Java消费者添加异常捕获，兼容两种消息格式：

while( true )
{
    System.out.println( "Listening..." );
    ClientMessage msgReceived = consumer.receive();
    
    String messageContent;
    try {
        // 优先按SimpleString格式读取（兼容Java生产者消息）
        messageContent = msgReceived.getBodyBuffer().readString();
    } catch (IndexOutOfBoundsException e) {
        // 读取失败则按字节数组读取（兼容STOMP消息）
        msgReceived.getBodyBuffer().resetReaderIndex();
        int bodySize = msgReceived.getBodySize();
        byte[] bytes = new byte[bodySize];
        msgReceived.getBodyBuffer().readBytes(bytes);
        messageContent = new String(bytes, "UTF-8");
    }
    
    System.out.println("message = " + messageContent);
    
    msgReceived.acknowledge();
    session.commit();
}

方案三：统一使用STOMP协议 #

Java端改用STOMP客户端，消除协议差异：

import org.apache.activemq.transport.stomp.StompConnection;
import org.apache.activemq.transport.stomp.StompFrame;

public class StompConsumerMain {
    public static void main(String[] args) throws Exception {
        StompConnection connection = new StompConnection();
        connection.open("172.16.1.141", 61613);
        connection.connect("username", "password");
        
        connection.subscribe("/queue/example", "client-individual");
        
        while (true) {
            StompFrame frame = connection.receive();
            System.out.println("message = " + frame.getBody());
            connection.ack(frame.getHeaders().get("message-id"));
        }
    }
}

内容的提问来源于stack exchange，提问作者mindcrime