java - activemq 故障转移连接不工作

Question

我正在尝试使用以下代码连接到 tcp://datafeeds.networkrail.co.uk:61619 以接收消息。该代码在没有故障转移的情况下工作正常(使用以下代码中的 url1)，但是当我尝试添加故障转移功能(使用以下代码中的 url)时，连接一直失败。从调试信息看来，activemq 无法为 URI 查找 INetAddress。 INetAddress 是什么以及可能导致问题的原因是什么？我现在很困惑。任何人都可以阐明如何解决这个问题吗？

最好的问候，

代码如下:

import javax.jms.*;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.log4j.BasicConfigurator;

public class Consumer {
 // URL of the JMS server
 private static String url = "failover://(tcp://datafeeds.networkrail.co.uk:61619)";
 private static String url1 = "tcp://datafeeds.networkrail.co.uk:61619";
 private static String username = "my username";
 private static String password = "my password";
 private static String topic = "*";

 public static void main(String[] args) throws JMSException {
  BasicConfigurator.configure();
  // Getting JMS connection from the server
  ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
  Connection connection = connectionFactory.createConnection(username,password);
   connection.start();
  Session session = connection.createSession(false,
    Session.AUTO_ACKNOWLEDGE);
  Destination destination = session.createTopic(topic);

  MessageConsumer consumer = session.createConsumer(destination);

    while (true)
    {
        //System.out.println("Waiting for message...");
        Message message = consumer.receive();
        if (message != null && message instanceof TextMessage) {
            TextMessage txtMsg = (TextMessage)message;
            System.out.println("Received: " + txtMsg.getText());
        }
    }
    //System.out.println("Closing connection");
    //consumer.close();
    //session.close();
    //connection.close();

}}

这是来自 activemq 的调试信息。

0 [main] DEBUG org.apache.activemq.thread.TaskRunnerFactory  - Initialized TaskRunnerFactory[ActiveMQ Task] using ExecutorService: java.util.concurrent.ThreadPoolExecutor@471ed915[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
2 [main] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Reconnect was triggered but transport is not started yet. Wait for start to connect the transport.
135 [main] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Started unconnected
135 [main] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Waking up reconnect task
137 [ActiveMQ Task-1] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - urlList connectionList:[tcp://datafeeds.networkrail.co.uk:61619], from: [tcp://datafeeds.networkrail.co.uk:61619]
214 [ActiveMQ Task-1] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Attempting  0th  connect to: tcp://datafeeds.networkrail.co.uk:61619
284 [ActiveMQ Task-1] DEBUG org.apache.activemq.transport.WireFormatNegotiator  - Sending: WireFormatInfo { version=10, properties={MaxFrameSize=9223372036854775807, CacheSize=1024, CacheEnabled=true, SizePrefixDisabled=false, MaxInactivityDurationInitalDelay=10000, TcpNoDelayEnabled=true, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]}
291 [ActiveMQ Task-1] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Connection established
292 [ActiveMQ Task-1] INFO org.apache.activemq.transport.failover.FailoverTransport  - Successfully connected to tcp://datafeeds.networkrail.co.uk:61619
314 [ActiveMQ Transport: tcp://datafeeds.networkrail.co.uk/176.34.248.27:61619@50427] DEBUG org.apache.activemq.transport.InactivityMonitor  - Using min of local: WireFormatInfo { version=10, properties={MaxFrameSize=9223372036854775807, CacheSize=1024, CacheEnabled=true, SizePrefixDisabled=false, MaxInactivityDurationInitalDelay=10000, TcpNoDelayEnabled=true, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]} and remote: WireFormatInfo { version=9, properties={CacheSize=1024, MaxFrameSize=9223372036854775807, CacheEnabled=true, SizePrefixDisabled=false, TcpNoDelayEnabled=true, MaxInactivityDurationInitalDelay=10000, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]}
317 [ActiveMQ Transport: tcp://datafeeds.networkrail.co.uk/176.34.248.27:61619@50427] DEBUG org.apache.activemq.transport.WireFormatNegotiator  - Received WireFormat: WireFormatInfo { version=9, properties={CacheSize=1024, MaxFrameSize=9223372036854775807, CacheEnabled=true, SizePrefixDisabled=false, TcpNoDelayEnabled=true, MaxInactivityDurationInitalDelay=10000, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]}
318 [ActiveMQ Transport: tcp://datafeeds.networkrail.co.uk/176.34.248.27:61619@50427] DEBUG org.apache.activemq.transport.WireFormatNegotiator  - tcp://datafeeds.networkrail.co.uk/176.34.248.27:61619@50427 before negotiation: OpenWireFormat{version=10, cacheEnabled=false, stackTraceEnabled=false, tightEncodingEnabled=false, sizePrefixDisabled=false, maxFrameSize=9223372036854775807}
373 [ActiveMQ Transport: tcp://datafeeds.networkrail.co.uk/176.34.248.27:61619@50427] DEBUG org.apache.activemq.transport.WireFormatNegotiator  - tcp://datafeeds.networkrail.co.uk/176.34.248.27:61619@50427 after negotiation: OpenWireFormat{version=9, cacheEnabled=true, stackTraceEnabled=true, tightEncodingEnabled=true, sizePrefixDisabled=false, maxFrameSize=9223372036854775807}
445 [ActiveMQ Transport: tcp://datafeeds.networkrail.co.uk/176.34.248.27:61619@50427] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Waking up reconnect task
747 [ActiveMQ Task-2] ERROR org.apache.activemq.transport.failover.FailoverTransport  - Failed to Lookup INetAddress for URI[ nio://ec2netrailprodmaster1:61619 ] : java.net.UnknownHostException: ec2netrailprodmaster1
748 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Doing rebalance from: tcp://datafeeds.networkrail.co.uk:61619 to [nio://ec2netrailprodmaster1:61619]
774 [ActiveMQ Task-2] DEBUG org.apache.activemq.util.ThreadPoolUtils  - Shutdown of ExecutorService: java.util.concurrent.ThreadPoolExecutor@1d60498d[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0] is shutdown: true and terminated: true took: 0.001 seconds.
776 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.tcp.TcpTransport  - Stopping transport tcp://datafeeds.networkrail.co.uk/176.34.248.27:61619@50427
777 [ActiveMQ Task-2] DEBUG org.apache.activemq.thread.TaskRunnerFactory  - Initialized TaskRunnerFactory[ActiveMQ Task] using ExecutorService: java.util.concurrent.ThreadPoolExecutor@4d1793dc[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
779 [ActiveMQ Task-1] DEBUG org.apache.activemq.transport.tcp.TcpTransport  - Closed socket Socket[addr=datafeeds.networkrail.co.uk/176.34.248.27,port=61619,localport=50427]
779 [ActiveMQ Task-2] DEBUG org.apache.activemq.util.ThreadPoolUtils  - Forcing shutdown of ExecutorService: java.util.concurrent.ThreadPoolExecutor@4d1793dc[Running, pool size = 1, active threads = 0, queued tasks = 0, completed tasks = 1]
780 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Waiting 10 ms before attempting connection. 
803 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Attempting  0th  connect to: nio://ec2netrailprodmaster1:61619
804 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Connect fail to: nio://ec2netrailprodmaster1:61619, reason: java.net.UnknownHostException
804 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.tcp.TcpTransport  - Stopping transport nio://ec2netrailprodmaster1:61619
804 [ActiveMQ Task-2] DEBUG org.apache.activemq.thread.TaskRunnerFactory  - Initialized TaskRunnerFactory[ActiveMQ Task] using ExecutorService: java.util.concurrent.ThreadPoolExecutor@7e7ee722[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
805 [ActiveMQ Task-1] DEBUG org.apache.activemq.transport.tcp.TcpTransport  - Closed socket Socket[unconnected]
805 [ActiveMQ Task-2] DEBUG org.apache.activemq.util.ThreadPoolUtils  - Forcing shutdown of ExecutorService: java.util.concurrent.ThreadPoolExecutor@7e7ee722[Running, pool size = 1, active threads = 0, queued tasks = 0, completed tasks = 1]
805 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Waiting 10 ms before attempting connection
815 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Waiting 20 ms before attempting connection. 
835 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Attempting  1th  connect to: nio://ec2netrailprodmaster1:61619
835 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Connect fail to: nio://ec2netrailprodmaster1:61619, reason: java.net.UnknownHostException
835 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.tcp.TcpTransport  - Stopping transport nio://ec2netrailprodmaster1:61619
835 [ActiveMQ Task-2] DEBUG org.apache.activemq.thread.TaskRunnerFactory  - Initialized TaskRunnerFactory[ActiveMQ Task] using ExecutorService: java.util.concurrent.ThreadPoolExecutor@e1ddc89[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
836 [ActiveMQ Task-1] DEBUG org.apache.activemq.transport.tcp.TcpTransport  - Closed socket Socket[unconnected]
836 [ActiveMQ Task-2] DEBUG org.apache.activemq.util.ThreadPoolUtils  - Forcing shutdown of ExecutorService: java.util.concurrent.ThreadPoolExecutor@e1ddc89[Running, pool size = 1, active threads = 0, queued tasks = 0, completed tasks = 1]
836 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Waiting 20 ms before attempting connection
856 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Waiting 40 ms before attempting connection. 
896 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.failover.FailoverTransport  - Attempting  2th  connect to: nio://ec2netrailprodmaster1:61619`enter code here`

Answer 1

我知道是个老问题，但我们刚刚与它发生冲突，所以我想分享答案....

当 ActiveMQ 作为集群安装时，您可以配置客户端连接如下:

private static String url = "failover://(tcp://datafeeds.networkrail.co.uk:61619)";

ActiveMQ 会定期向连接的客户端发送响应以断开连接并重新建立连接。您可以在以下日志中看到这一点:

748 [ActiveMQ Task-2] DEBUG org.apache.activemq.transport.failover.FailoverTransport - Doing rebalance from: tcp://datafeeds.networkrail.co.uk:61619 to [nio://ec2netrailprodmaster1:61619]

当 ActiveMQ 告诉客户端重新平衡并将其发回无法访问的 nio://ec2netrailprodmaster1:61619 uri 时，就会出现不愉快。

因此，正确的解决方法是编辑 ActiveMQ 配置。对于上面的问题会有这样的事情:

<transportConnectors>
    <transportConnector name="openwire" uri="nio://ec2netrailprodmaster1:61619?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"  rebalanceClusterClients="true"/>
</transportConnectors>  

应该改成这样:

<transportConnectors>
    <transportConnector name="openwire" uri="tcp://datafeeds.networkrail.co.uk:61619?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"  rebalanceClusterClients="true"/>
</transportConnectors> 

或者更具体地说，transportConnector uri 应该设置为 ActiveMQ 实例的主机。