java - 如何使用故障转移传输处理 Activemq 的最大帧大小异常

Question

我正在开发一个使用 activemq 交换消息的应用程序，有些消息太大以至于我想取消。

我们使用两个 ActiveMQ 实例(主/从)的 activemq 故障转移传输。代理本身对消息有 100mb 的帧大小限制。

问题是:如果我尝试发送大于 100mb 的消息，ActiveMQ 服务器将关闭连接。此时，故障转移传输将尝试重新连接并再次发送消息，从而形成无限循环。

客户端记录如下:

2017-01-05 09:19:11.910  WARN 14680 --- [0.1:61616@57025] o.a.a.t.failover.FailoverTransport       : Transport (tcp://localhost:61616) failed , attempting to automatically reconnect: {}

java.io.EOFException: null
    at java.io.DataInputStream.readInt(DataInputStream.java:392) ~[na:1.8.0_91]
    at org.apache.activemq.openwire.OpenWireFormat.unmarshal(OpenWireFormat.java:267) ~[activemq-client-5.13.4.jar:5.13.4]
    at org.apache.activemq.transport.tcp.TcpTransport.readCommand(TcpTransport.java:240) ~[activemq-client-5.13.4.jar:5.13.4]
    at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:232) ~[activemq-client-5.13.4.jar:5.13.4]
    at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:215) ~[activemq-client-5.13.4.jar:5.13.4]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_91]

2017-01-05 09:19:11.921  INFO 14680 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport       : Successfully reconnected to tcp://localhost:61616
2017-01-05 09:19:11.923  WARN 14680 --- [0.1:61616@57026] o.a.a.t.failover.FailoverTransport       : Transport (tcp://localhost:61616) failed , attempting to automatically reconnect: {}

java.io.EOFException: null
    at java.io.DataInputStream.readInt(DataInputStream.java:392) ~[na:1.8.0_91]
    at org.apache.activemq.openwire.OpenWireFormat.unmarshal(OpenWireFormat.java:267) ~[activemq-client-5.13.4.jar:5.13.4]
    at org.apache.activemq.transport.tcp.TcpTransport.readCommand(TcpTransport.java:240) ~[activemq-client-5.13.4.jar:5.13.4]
    at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:232) ~[activemq-client-5.13.4.jar:5.13.4]
    at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:215) ~[activemq-client-5.13.4.jar:5.13.4]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_91]

当 activeMQ 实例记录时:

2017-01-05 09:19:11,909 | WARN  | Transport Connection to: tcp://127.0.0.1:57025 failed: java.io.IOException: Frame size of 363 MB larger than max allowed 100 MB | org.apache.activemq.broker.TransportConnection.Transport | ActiveMQ Transport: tcp:///127.0.0.1:57025@61616
2017-01-05 09:19:11,922 | WARN  | Transport Connection to: tcp://127.0.0.1:57026 failed: java.io.IOException: Frame size of 363 MB larger than max allowed 100 MB | org.apache.activemq.broker.TransportConnection.Transport | ActiveMQ Transport: tcp:///127.0.0.1:57026@61616

我试图设置一个 TransportListener 来验证我是否可以捕获这种情况，但我只收到一个 transportInterupted 事件，没有任何分类器。

我阅读了有关故障转移传输 (http://activemq.apache.org/failover-transport-reference.html) 的文档，也许我可以使用 maxReconnectAttempts，但我知道在更常见的情况下我会有几个缺点(比如服务器暂时不可用)。

如何检测这种情况并避免客户端和服务器之间的无限连接循环？

Answer 1

如你所说

maxReconnectAttempts -1 | 0 From ActiveMQ 5.6: default is -1, retry forever. 0 means disables re-connection, e.g: just try to connect once. Before ActiveMQ 5.6: default is 0, retry forever. All ActiveMQ versions: a value >0 denotes the maximum number of reconnect attempts before an error is sent back to the client.

因此，如果您希望您的传输监听器在由于消息大小而导致重试失败后收到传输失败通知，您需要将 maxReconnectAttempts 设置为 > 0 之后，当最大重试次数达到传输监听器的 onException 方法时将使用 IOException 作为参数调用，但正如您所说，验证它是由于最大大小还是其他问题并不容易。

如果您想在发送前按照建议检查消息大小，您可以在运行时通过 jmx 访问代理端的 uri 中配置 maxFrameSize，并获取 BrokerViewMBean 实例并调用 getTransportConnectorByType 方法 http://activemq.apache.org/maven/apidocs/src-html/org/apache/activemq/broker/jmx/BrokerViewMBean.html#line.304这将返回在 activemq.xml 中配置的 uri，您可以对其进行解析以检索 maxFrameSize。

JMXServiceURL url = new     JMXServiceURL("service:jmx:rmi:///jndi/rmi://hist:1099/jmxrmi");
JMXConnector jmxc = JMXConnectorFactory.connect(url);
MBeanServerConnection conn = jmxc.getMBeanServerConnection(); 

ObjectName activeMq = new ObjectName("org.apache.activemq:Type=Broker,BrokerName=localhost");

BrokerViewMBean mbean = (BrokerViewMBean)MBeanServerInvocationHandler.newProxyInstance(conn, activeMq, BrokerViewMBean.class, true);
String uri = mbean.getTransportConnectorByType("tcp");// or ("ssl") 
String[] pairs = uri.split("&");
for (String pair : pairs) {
    if (pair.startsWith("wireFormat.maxFrameSize")) {
        int idx = pair.indexOf("=");
        System.out.println(URLDecoder.decode(pair.substring(idx + 1), "UTF-8"));
    }
}

http://activemq.apache.org/maven/apidocs/org/apache/activemq/broker/jmx/BrokerViewMBean.html#getTransportConnectors-- 将返回传输名称作为键和 uri 作为值的映射

要获得更好的消息大小，您可以这样做:

        OpenWireFormat opf = new OpenWireFormat();
        opf.setTightEncodingEnabled(true);
        ByteSequence tab = opf.marshal(message);
        System.out.println(tab.length);

您的业务必须是这样的:

import java.io.IOException;
import java.net.URLDecoder;
import java.util.HashMap;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.management.MBeanServerConnection;
import javax.management.MBeanServerInvocationHandler;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.broker.jmx.BrokerViewMBean;
import org.apache.activemq.command.ActiveMQTextMessage;
import org.apache.activemq.openwire.OpenWireFormat;
import org.apache.activemq.transport.TransportFilter;
import org.apache.activemq.transport.TransportListener;
import org.apache.activemq.transport.failover.FailoverTransport;
import org.apache.activemq.util.ByteSequence;

public class SimpleSenderMaxSizeManager {

    private static Connection conn = null;
    private static boolean transportChanged;
    private static Long MAX_FRAME_SIZE;

    public static void main(String[] args) throws JMSException {
        try {
            SimpleSenderMaxSizeManager.updateMaxSize("host1");
            ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(
                    "failover:(tcp://host1:5670,tcp://host2:5671)?randomize=false");
            cf.setTransportListener(new TransportListener() {

                @Override
                public void transportResumed() {
                    if (transportChanged) {
                        transportChanged = false;
                        try {
                            SimpleSenderMaxSizeManager.updateMaxSize(null);
                        } catch (Exception e) {
                        }
                    }
                }

                @Override
                public void transportInterupted() {
                    transportChanged = true;
                }

                @Override
                public void onException(IOException error) {
                }

                @Override
                public void onCommand(Object command) {
                }
            });
            conn = cf.createConnection();
            ActiveMQSession session = (ActiveMQSession) conn.createSession(false,
                    ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE);
            MessageProducer producer = session.createProducer(session.createQueue("TEST"));
            conn.start();
            ActiveMQTextMessage msg = (ActiveMQTextMessage) session.createTextMessage("test");
            OpenWireFormat opf = new OpenWireFormat();
            opf.setTightEncodingEnabled(true);
            ByteSequence tab = opf.marshal(msg);
            System.out.println(tab.length);
            if (tab.length >= MAX_FRAME_SIZE) {
                throw new RuntimeException(tab.length + ">=" + MAX_FRAME_SIZE);
            }
            producer.send(msg);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (Exception e) {
                }
            }
        }
    }

    protected static void updateMaxSize(String host) throws Exception {
        JMXConnector jmxc = null;
        try {
            String jmxHost = host;
            String scheme = null;
            if (conn == null) {
                scheme = "tcp";
            } else {
                org.apache.activemq.transport.TransportFilter responseCorrelator = (TransportFilter) ((ActiveMQConnection) conn)
                        .getTransport();
                TransportFilter mutexTransport = (TransportFilter) responseCorrelator.getNext();
                FailoverTransport failoverTransport = (FailoverTransport) mutexTransport.getNext();
                while (failoverTransport.getConnectedTransportURI() == null) {
                    try {
                        Thread.sleep(100);
                    } catch (Exception e) {
                    }
                }
                scheme = failoverTransport.getConnectedTransportURI().getScheme();
                jmxHost = failoverTransport.getConnectedTransportURI().getHost();
            }
            JMXServiceURL url = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://" + jmxHost + ":1099/jmxrmi");
            Map<String, String[]> env = new HashMap<>();
            String[] creds = { "admin", "admin" };
            env.put(JMXConnector.CREDENTIALS, creds);
            jmxc = JMXConnectorFactory.connect(url, env);
            MBeanServerConnection conn = jmxc.getMBeanServerConnection();
            ObjectName activeMq = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost");
            BrokerViewMBean mbean = MBeanServerInvocationHandler.newProxyInstance(conn, activeMq, BrokerViewMBean.class,
                    true);
            String value = mbean.getTransportConnectorByType(scheme);
            String[] pairs = value.split("&");
            for (String pair : pairs) {
                if (pair.contains("wireFormat.maxFrameSize")) {
                    int idx = pair.indexOf("=");
                    System.out.println(URLDecoder.decode(pair.substring(idx + 1), "UTF-8"));
                    MAX_FRAME_SIZE = Long.valueOf(URLDecoder.decode(pair.substring(idx + 1), "UTF-8"));
                    MAX_FRAME_SIZE -= 1000;// security for JMS headers added by
                                            // session on sending
                }
            }
        } finally {
            if (jmxc != null) {
                try {
                    jmxc.close();
                } catch (Exception e) {
                }
            }
        }
    }
}