amqp - 연결이 끊어진 경우 자동 복구 사용

Question

연결이 끊어진 경우 자동 복구 기능을 사용하려고합니다. 2 가지 문제가 있습니다 :

1) 자동 복구 코드는 networkRecoveryInterval을 완전히 무시합니다. 내 로그 파일에서 연결이 끊어지는 1 분 동안 파일은 1.5GB로 증가합니다. 다음 오류가 계속 반복됩니다.

연결 복구 중에 예외가 발생했습니다! java.net.SocketException의 : 네트워크 java.net.AbstractPlainSocketImpl.connectToAddress에서 java.net.AbstractPlainSocketImpl.doConnect (AbstractPlainSocketImpl.java:339)에서 java.net.PlainSocketImpl.socketConnect (기본 방법) (에 도달 할 수없는 java.net.Socket.connect에서 java.net.AbstractPlainSocketImpl.connect (AbstractPlainSocketImpl.java:182) java.net.SocksSocketImpl.connect (SocksSocketImpl.java:391에서 )에서 AbstractPlainSocketImpl.java:200) (소켓 .java : 579) at com.rabbitmq.client.impl.FrameHandlerFactory.create (FrameHandlerFactory.java:32) at com.rabbitmq.client.impl.recovery.RecoveryAwareAMQConnectionFactory com.rabbitmq.client.impl.recovery.AutorecoveringConnection.beginAutomaticRecovery에서 .newConnection com.rabbitmq.client.impl.recovery.AutorecoveringConnection.recoverConnection (AutorecoveringConnection.java:388)에서 (RecoveryAwareAMQConnectionFactory.java:34) (AutorecoveringConnection 된 .java : 360) 에서 com.rabbitmq.client.impl.recovery.AutorecoveringConnection.access $ 000 com.rabbitmq.client.impl.recovery.AutorecoveringConnection $ 1.shutdownCompleted에서 (AutorecoveringConnection.java:48) (AutorecoveringConnection.java : 345) at com.rabbitmq.client.impl.ShutdownNotifierComponent.notifyListeners (ShutdownNotifierComponent.java:75) atjava.lang.Thread.run에서com.rabbitmq.client.impl.AMQConnection $ MainLoop.run (AMQConnection.java:572) (Thread.java:722)

2) 스위치을 마지막으로 할 때 내 라우터를 켜면 네트워크 복구가 작동하지 않습니다. 내 로그 파일에 이걸 가져 왔어.

는 큐 8923yrbk com.rabbitmq.client.TopologyRecoveryException를 회수하면서 회수 토폴로지 예외를 잡은 잡 예외 : com.rabbitmq.client.impl.recovery.AutorecoveringConnection에서 큐 8923yrbk 회수 동안 예외 잡힌 .recoverQueues (AutorecoveringConnection.java:459) 에서 com.rabbitmq.client.impl.recovery.AutorecoveringConnection.recoverEntities com.rabbitmq.client.impl.recovery.AutorecoveringConnection.beginAutomaticRecovery에서 (AutorecoveringConnection.java:424) (AutorecoveringConnection .java : 365) 에서 91,363,210 com.rabbitmq.client.impl.recovery.AutorecoveringConnection.access $ 000 com.rabbitmq.client.impl.recovery.AutorecoveringConnection $ 1.shutdownCompleted (AutorecoveringConnection.java:345) 에서 (AutorecoveringConnection.java:48) 에서 com.rabbitmq.client.impl.ShutdownNotifierComponent.notifyListeners (ShutdownNotifierComponent.java:75) com.rabbitmq.client.impl.AMQConnection $ MainLoop.run (AMQConnection.java:572) java.lang.Thread의에서 에서 . (스레드.java : 722) 원인 : com.rabbitmq.client.AlreadyClosedException : 연결 오류로 인해 연결이 이미 으로 종료되었습니다. 원인 : com.rabbitmq.client.MissedHeartbeatException : 하트 비트 심장 박동 누락 = com.rabbitmq.client.impl.AMQChannel.ensureIsOpen (AMQChannel.java:190)에서 45초 com.rabbitmq.client.impl.AMQChannel에서 (AMQChannel.java:209) 에서 .rpc (AMQChannel.java:223) com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc (AMQChannel.java:209) com.rabbitmq에서 com.rabbitmq.client.impl.recovery.AutorecoveringChannel.queueDeclare (AutorecoveringChannel.java:230) 에서 com.rabbitmq.client.impl.ChannelN.queueDeclare (ChannelN.java:779) 에서 . client.impl.recovery.RecordedQueue.reco com.rabbitmq.client.impl.recovery.AutorecoveringConnection.recoverQueues (AutorecoveringConnection.java:448) ...의 버전 (RecordedQueue.java:36) 7 개의 다음

내 비교적 간단 소비자 측 인 암호. 생성자를 차단하고 싶지 않기 때문에 모든 코드를 새 스레드에 씁니다.

private ConnectionFactory factory = null; 
private Connection connection = null; 
private Channel channel = null; 

private PaymentInfoFromGlobalServerConsumer() { 
    new Thread(new Runnable() { 
     public void run() { 
      factory = new ConnectionFactory(); 
      try { 
       factory.setUri(amqpServerUrl); 
       factory.setAutomaticRecoveryEnabled(true); 
       factory.setNetworkRecoveryInterval(30000); // In case of broken connection, try again every 30 seconds (hope this is correct understanding) 
       factory.setRequestedHeartbeat(45); //Keep sending the heartbeat every 45 seconds to prevent any routers from considering the connection stale. 
      } catch (KeyManagementException | NoSuchAlgorithmException | URISyntaxException e) { 
       //Will never happen if configured properly 
       logger.error(e); 
       return; 
      } 

      try { 
       connection = factory.newConnection(); 
       channel = connection.createChannel(); 
       //Create a durable queue (if not already present) 
       channel.queueDeclare(merchantId, true, false, false, null); 

       QueueingConsumer consumer = new QueueingConsumer(channel); 
       channel.basicConsume(merchantId, false, consumer); 

       while (true) { 
        QueueingConsumer.Delivery delivery = consumer.nextDelivery(); 
        String billId = new String(delivery.getBody()); 

        //TODO - Redeliveries are possible as per design 
        System.out.println(" [x] Received '" + billId + "'"); 
        System.out.println(" [x] Done"); 

        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); 
       }    
      } catch (IOException | ConsumerCancelledException | InterruptedException e) { 
       e.printStackTrace(); 
       logger.error(e); 
      } catch (ShutdownSignalException e) { 
       System.out.println(e.isInitiatedByApplication() + " " + e.isHardError()); 
      } finally { 
       close(); 
      } 
     }   
    }).start(); 
} 

public void close() { 
    try { 
     if (channel != null) channel.close(); 
    } catch (IOException | AlreadyClosedException e) { 
     //Cannot do anything now 
    } 
    try { 
     if (connection != null) connection.close(); 
    } catch (IOException | AlreadyClosedException e) { 
     //Cannot do anything now 
    } 
} 

저는 amqp의 초보자입니다. 따라서 도움을 주시면 감사하겠습니다. 감사합니다

Answer 1

연결과 함께 토폴로지를 복구하십시오.

factory.setTopologyRecoveryEnabled(true);