使用 Java 和 AMQP 1.0 消息大小问题的 Windows Server (1.1) 上的服务总线

Question

我们使用下面的 AMQP jar 将消息从 java 发送到 windows server(1.1) 上的服务总线中的队列设置。发送小消息没有问题，但无法发送在 pre prod 中大小超过 45kb 和在 prod 中大小超过 160kb 的消息。该消息根本没有进入队列，java端也不例外。

geronimo-jms_1.1_spec-1.0.1.jar qpid-amqp-1-0-client-0.22.jar qpid-amqp-1-0-client-jms-0.22.jar qpid-amqp-1-0-common-0.22.jar

收到的消息在网关配置中正确配置。

我只是想知道是什么导致了这个消息大小限制？任何帮助将不胜感激。

谢谢你， 斯里。

Answer 1

我遇到了同样的问题。无法发送大于 256kb 的消息。出现错误链接已分离。

问题在于服务总线的配置。默认情况下，SB 不允许大于 256kb 的消息。我们必须更改配置以允许更大的消息。

Answer 2

这是我为发送一条消息（109 kb）而编写的小测试程序。我在 java 端没有收到任何异常，而且消息也没有出现在服务总线队列中。

import org.apache.qpid.amqp_1_0.jms.impl.*;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import java.net.URLEncoder;
import java.util.Hashtable;
import java.io.*;

public class ServiceBusPublisher {

public static void main(String[] args) throws Exception {               
    String userName = "test";
    String password = "test";       
    String fqdn = "test@test.com";

    String connectionString = "amqps://" + encode(userName) + ":" + encode(password) + "@" + fqdn;


        File file = new File("D:\\temp\\test.xml");
    FileReader fileReader = new FileReader(file);
    BufferedReader bufferedReader = new BufferedReader(fileReader);
    StringBuffer stringBuffer = new StringBuffer();
    String line;
    while ((line = bufferedReader.readLine()) != null) {
        stringBuffer.append(line);
        stringBuffer.append("\n");
    }
    bufferedReader.close();
    fileReader.close();

    String messageText = stringBuffer.toString();         

    System.out.print("Initial setup\n");
        Hashtable<String, String> env = new Hashtable<String, String>();
        env.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.amqp_1_0.jms.jndi.PropertiesFileInitialContextFactory");
        env.put(Context.PROVIDER_URL, "blah.txt");
        env.put("connectionfactory.ServiceBusConnectionFactory", connectionString);
        Context context = null;
        ConnectionFactory connectionFactory = null;
        Connection connection = null;
        try
        {
            System.out.print("Creating context\n");
            context = new InitialContext(env);
            System.out.print("Creating connection factory\n");
            connectionFactory = (ConnectionFactory)context.lookup("ServiceBusConnectionFactory");
            System.out.print("Creating connection\n");
            connection = connectionFactory.createConnection();
        }
        catch (Exception e)
        {
            System.out.print("Exception creating context/connection factory/connection\n");
            System.out.print(e.toString());
            return;
        }

        Session session = null;
        MessageProducer producer = null;
        try
        {
            System.out.print("Creating session\n");
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            System.out.print("Creating producer\n");
            producer = session.createProducer(QueueImpl.createQueue("MessageProcessingService-PRE/testqueue"));                                                        
        }
        catch (Exception e)
        {
            System.out.print("Exception creating session/producer\n");
            System.out.print(e.toString());
            return;
        }

        try
        {
            System.out.print("Creating message\n");
            BytesMessage message = session.createBytesMessage();                
            message.writeBytes(messageText.getBytes("UTF-8"));              
            message.setStringProperty("JMSXGroupID", "MessageProcessingService-PRE/testqueue");
            System.out.print("Sending\n");
            producer.send(message);             
        }
        catch (Exception e)
        {
            System.out.print("Exception creating message and sending\n");
            System.out.print(e.toString());
            return;
        }           

        try
        {
            System.out.print("Cleaning up\n");
            producer.close();
            session.close();
            connection.stop();
            connection.close();
        }
        catch (Exception e)
        {
            System.out.print("Exception closing producer and session\n");
            System.out.print(e.toString());
        }
}

public static String encode(String encodee)
    {
        String retval = "";
        try { retval = URLEncoder.encode(encodee, "UTF-8"); } catch (Exception e) { System.out.print("Encoding failed\n"); }
        return retval;
    }

}