主页 / user-1196726

Undertaker's questions

Martin Hope
Undertaker
Asked: 2024-05-07 23:31:57 +0800 CST
  • 5

我正在将 ActiveMQ Artemis 与 Vertex Broker 客户端结合使用。我想测试下面的代码：

public class ActiveMqMessageSender {

  private String queue = "myQueue";

  public void sendMessage(String message) throws ExecutionException, InterruptedException {
    AmqpMessage msg =
        AmqpMessage.create()
            .withBody(message)
            .build();
    client = createAmqpClient();
    var amqpConnectionFuture = client.connect();
    // Wait here to ensure that the connection is successful.
    var amqpConnection = amqpConnectionFuture.toCompletionStage().toCompletableFuture().get();
    log.info("Vert.x AMQP connection created successfully.");

    var amqpSenderFuture = amqpConnection.createSender(queue);
    // Wait here to ensure that the connection is successful.
    var amqpSender = amqpSenderFuture.toCompletionStage().toCompletableFuture().get();
    log.info("Vert.x AMQP sender created successfully.");

    amqpSender.sendWithAck(msg, ackResult -> {
      if (ackResult.succeeded()) {
        log.info("Message sent successfully.");
      } else {
        log.error("Failed to send message: " + ackResult.cause().getMessage());
      }
    });
  }

  private AmqpClient createAmqpClient() {
    var options =
        new AmqpClientOptions()
            .setHost("localhost")
            .setPort("61616")
            .setUsername("test")
            .setPassword("test");
    return AmqpClient.create(options);
  }

在我的 JUnit 测试中：

@SpringBootTest
class ActiveMqMessageSenderTest {

  private String queue = "myQueue";

  @Autowired
  private ActiveMqMessageSender messageSender;
  private static EmbeddedActiveMQExtension embeddedActiveMQ;

  @BeforeAll
  public static void setUp() throws Exception {
    Configuration configuration =
        new ConfigurationImpl().setName("embedded-server")
            .setPersistenceEnabled(false)
            .setSecurityEnabled(false)
            .addAcceptorConfiguration("default", "tcp://localhost:61616")
            .addAddressSetting("#",
                new AddressSettings().setDeadLetterAddress(SimpleString.toSimpleString("dla"))
                    .setExpiryAddress(SimpleString.toSimpleString("expiry")));

    // Start the embedded ActiveMQ Artemis server
    embeddedActiveMQ = new EmbeddedActiveMQExtension(configuration);
    embeddedActiveMQ.start();
  }

  @AfterAll
  public static void tearDown() throws Exception {
    // Stop the embedded ActiveMQ Artemis server
    if (embeddedActiveMQ != null) {
      embeddedActiveMQ.stop();
    }
  }

  @Test
  void testSendMessage() throws Exception {

    String message = "Test Message";
    messageSender.sendMessage(message);

  }
}

我的 application.yml 看起来像

  artemis:
    broker:
      host: localhost
      port: 61616
      username: artemis
      password: artemis

当我运行本地 ActiveMQ 服务器时，代码可以工作。但是当我尝试通过 JUnit 运行嵌入式服务器时，我在日志中看到以下内容：

[Test worker] INFO org.apache.activemq.artemis.core.server -- AMQ221020: Started NIO Acceptor at localhost:61616 for protocols [CORE]
[Test worker] INFO org.apache.activemq.artemis.core.server -- AMQ221007: Server is now live
[Test worker] INFO org.apache.activemq.artemis.core.server -- AMQ221001: Apache ActiveMQ Artemis Message Broker version 2.31.2 [embedded-server, nodeID=9df62ceb-0c83-11ef-a6a7-00155ddd1843]

在这个时刻它等待：

INFO   [Thread-1 (activemq-netty-threads)] org.apache.activemq.audit.resource - AMQ601767: CORE connection a6e64464 for user [email protected]:54291 created

最后我得到：

WARN   [Thread-1 (ActiveMQ-server-org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl$6@72f3f14c)] o.a.activemq.artemis.core.client - AMQ212037: Connection failure to /127.0.0.1:54291 has been detected: AMQ229014: Did not receive data from /127.0.0.1:54291 within the 60000ms connection TTL. The connection will now be closed. [code=CONNECTION_TIMEDOUT]

io.vertx.core.VertxException: Disconnected
java.util.concurrent.ExecutionException: io.vertx.core.VertxException: Disconnected

似乎与服务器的连接正常，但是当我尝试通过调用连接客户端时，client.connect()它不起作用。

junit