2

我正在使用简单的 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)

并且没有任何错误,例如“队列“总线状态队列”上超出队列容量阈值”。怎么了?

4

1 回答 1

4

这是由于AMPQ 协议中与字符串大小相关的限制。str16是 64K 字符串大小。当 Maps 和 List 等数据结构用于包含字符串时,放置此限制的代码部分会生效。使用简单的 TextMessage,它被视为原始数据。

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

于 2013-07-02T00:07:15.650 回答