java - Qpid Java客户端异常: java. lang.IllegalArgumentException:未知代码:105

Question

我正在使用简单的 qpid java 客户端，它使用来自代理的消息并通过 SOAP 服务将其发送到。在生产者方面，我们将所有数据放入映射，然后将该映射发送到 qpid。这是一个片段:

QueueSender conBusQueueSender = (QueueSender) Component.getInstance("conBusQueueSender");
QueueSession queueSession = org.jboss.seam.jms.QueueSession.instance();

Map<String,Object> map = new HashMap<String,Object>();
map.put("applicationId", applicationId);
map.put("soapAction", "urn:changeApplicationStatus");
map.put("soapXML", changeApplicationStatusString);

MapMessage message = queueSession.createMapMessage();
message.setObject("map",map);

conBusQueueSender.send(message);

在客户端，我们接收消息并尝试通过 SOAP 将其发送到 Web 服务

    while (true) {
    MapMessage m = (MapMessage) consumer.receive();

    @SuppressWarnings("unchecked")
    Map<String, Object> map = (Map<String, Object>) m.getObject("map");

    SOAPMessage message = createSoapMessage(map);

    while (true) {
        try {
            SOAPMessage result = makeSoapCall(message, (String)map.get("applicationId"), (String)map.get("soapAction"));
            if(result.getSOAPBody().hasFault()){
                System.out.println("SOAPFault: "+result.getSOAPBody().getFault().getFaultString());
                System.out.println("SOAPMessage tried to send: " + (String)map.get("soapXML"));
            }

            m.acknowledge();

            // if we came here successfully -> break and process next message
            break;
        } catch (SOAPException e) {
            // if we have exception -> sleep and retry
            e.printStackTrace();
            Thread.sleep(1000L);
            continue;
        }
    }
}

一切正常，但当消费者尝试发送更大的消息 ~66KB 时，它只会打印此错误:

73 [main] INFO org.apache.qpid.client.security.DynamicSaslRegistrar - Additional SASL providers successfully registered.
87 [main] INFO org.apache.qpid.client.AMQConnection - Connection:amqp://guest:********@localhost/?brokerlist='tcp://localhost:5672'
314 [main] INFO org.apache.qpid.client.protocol.AMQProtocolSession - Using ProtocolVersion for Session:0-10
330 [main] INFO org.apache.qpid.client.handler.ClientMethodDispatcherImpl - New Method Dispatcher:AMQProtocolSession[null]
341 [main] INFO org.apache.qpid.client.AMQConnection - Connecting with ProtocolHandler Version:0-10
414 [IoReceiver - localhost/127.0.0.1:5672] INFO org.apache.qpid.transport.ClientDelegate - The broker does not support the configured connection idle timeout of 120 sec,  using the brokers max supported value of 0 sec instead.
420 [main] INFO org.apache.qpid.client.AMQConnection - Connected with ProtocolHandler Version:0-10
Connection established to amqp://guest:guest@localhost/?brokerlist='tcp://localhost:5672'
443 [main] INFO org.apache.qpid.client.AMQSession - Created session:org.apache.qpid.client.AMQSession_0_10@112c3327
Session created...
476 [main] INFO org.apache.qpid.client.AMQSession - Prefetching delayed existing messages will not flow until requested via receive*() or setML().
Consumer initialized...
481 [main] INFO org.apache.qpid.client.AMQSession.Dispatcher - Dispatcher-Channel-0 created
481 [Dispatcher-Channel-0] INFO org.apache.qpid.client.AMQSession.Dispatcher - Dispatcher-Channel-0 started
492 [Dispatcher-Channel-0] ERROR org.apache.qpid.client.BasicMessageConsumer - Caught exception (dump follows) - ignoring...
java.lang.IllegalArgumentException: unknown code: 105
    at org.apache.qpid.transport.codec.AbstractDecoder.getType(AbstractDecoder.java:354)
    at org.apache.qpid.transport.codec.AbstractDecoder.readMap(AbstractDecoder.java:287)
    at org.apache.qpid.transport.codec.BBDecoder.readMap(BBDecoder.java:34)
    at org.apache.qpid.transport.codec.AbstractDecoder.read(AbstractDecoder.java:455)
    at org.apache.qpid.transport.codec.AbstractDecoder.readMap(AbstractDecoder.java:288)
    at org.apache.qpid.transport.codec.BBDecoder.readMap(BBDecoder.java:34)
    at org.apache.qpid.client.message.AMQPEncodedMapMessage.populateMapFromData(AMQPEncodedMapMessage.java:96)
    at org.apache.qpid.client.message.JMSMapMessage.<init>(JMSMapMessage.java:71)
    at org.apache.qpid.client.message.AMQPEncodedMapMessage.<init>(AMQPEncodedMapMessage.java:52)
    at org.apache.qpid.client.message.AMQPEncodedMapMessageFactory.createMessage(AMQPEncodedMapMessageFactory.java:36)
    at org.apache.qpid.client.message.AbstractJMSMessageFactory.create010MessageWithBody(AbstractJMSMessageFactory.java:135)
    at org.apache.qpid.client.message.AbstractJMSMessageFactory.createMessage(AbstractJMSMessageFactory.java:166)
    at org.apache.qpid.client.message.MessageFactoryRegistry.createMessage(MessageFactoryRegistry.java:150)
    at org.apache.qpid.client.BasicMessageConsumer_0_10.createJMSMessageFromUnprocessedMessage(BasicMessageConsumer_0_10.java:221)
    at org.apache.qpid.client.BasicMessageConsumer_0_10.createJMSMessageFromUnprocessedMessage(BasicMessageConsumer_0_10.java:47)
    at org.apache.qpid.client.BasicMessageConsumer.notifyMessage(BasicMessageConsumer.java:693)
    at org.apache.qpid.client.BasicMessageConsumer_0_10.notifyMessage(BasicMessageConsumer_0_10.java:205)
    at org.apache.qpid.client.BasicMessageConsumer_0_10.notifyMessage(BasicMessageConsumer_0_10.java:47)
    at org.apache.qpid.client.AMQSession$Dispatcher.notifyConsumer(AMQSession.java:3392)
    at org.apache.qpid.client.AMQSession$Dispatcher.dispatchMessage(AMQSession.java:3336)
    at org.apache.qpid.client.AMQSession$Dispatcher.access$900(AMQSession.java:3117)
    at org.apache.qpid.client.AMQSession.dispatch(AMQSession.java:3110)
    at org.apache.qpid.client.message.UnprocessedMessage.dispatch(UnprocessedMessage.java:55)
    at org.apache.qpid.client.AMQSession$Dispatcher.run(AMQSession.java:3264)
    at java.lang.Thread.run(Thread.java:680)

并且不存在任何错误，例如“队列“bus-status-queue”超出队列容量阈值”。怎么了？

Answer 1

这是由于 AMPQ protocol 中的限制造成的与字符串大小有关。 str16 是 64K 字符串大小。 code part当使用映射和列表等数据结构来包含字符串时，此限制就会生效。对于简单的 TextMessage，它被视为原始数据。

您的问题的解决方案是使用byte[]代替，它被定义为vbin32，可以包含4GB数据。