ActiveMQ示例代码第38行连接失败：Connection refused错误求助

运行ActiveMQ的AMQP示例代码时，执行到第38行Connection connection = factory.createConnection(user, password);总是失败，核心错误为Connection refused: localhost/127.0.0.1:5672。

示例代码 #

public static void main(String[] args) throws JMSException {

    final String TOPIC_PREFIX = "topic://";

    String user = env("ACTIVEMQ_USER", "admin");
    String password = env("ACTIVEMQ_PASSWORD", "admin");
    String host = env("ACTIVEMQ_HOST", "localhost");
    int port = Integer.parseInt(env("ACTIVEMQ_PORT", "5672"));

    String connectionURI = "amqp://" + host + ":" + port;
    String destinationName = arg(args, 0, "stage");

    JmsConnectionFactory factory = new JmsConnectionFactory(connectionURI);

    Connection connection = factory.createConnection(user, password);
    connection.start();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

    Destination destination = null;
    if (destinationName.startsWith(TOPIC_PREFIX)) {
        destination = session.createTopic(destinationName.substring(TOPIC_PREFIX.length()));
    } else {
        destination = session.createQueue(destinationName);
    }

    MessageConsumer consumer = session.createConsumer(destination);
    long start = System.currentTimeMillis();
    long count = 1;
    System.out.println("Waiting for messages...");
    while (true) {
        Message msg = consumer.receive();
        if (msg instanceof TextMessage) {
            String body = ((TextMessage) msg).getText();
            if ("SHUTDOWN".equals(body)) {
                long diff = System.currentTimeMillis() - start;
                System.out.println(String.format("Received %d in %.2f seconds", count, (1.0 * diff / 1000.0)));
                connection.close();
                try {
                    Thread.sleep(10);
                } catch (Exception e) {}
                System.exit(0);
            } else {
                        System.out.println(String.format("Received %d messages.", msg.toString()));
            }

        } else {
            System.out.println("Unexpected message type: " + msg.getClass());
        }
    }
}

错误信息 #

/home/thabangligavha/.jdks/openjdk-18.0.2.1/bin/java -javaagent:/snap/intellij-idea-community/405/lib/idea_rt.jar=39811:/snap/intellij-idea-community/405/bin -Dfile.encoding=UTF-8 -Dsun.stdout.encoding=UTF-8 -Dsun.stderr.encoding=UTF-8 -classpath /home/thabangligavha/apache-activemq-5.17.3/examples/amqp/java/target/classes:/home/thabangligavha/.m2/repository/org/apache/geronimo/specs/geronimo-jms_1.1_spec/1.1/geronimo-jms_1.1_spec-1.1.jar:/home/thabangligavha/.m2/repository/org/apache/qpid/qpid-jms-client/1.6.0/qpid-jms-client-1.6.0.jar:/home/thabangligavha/.m2/repository/org/slf4j/slf4j-api/1.7.36/slf4j-api-1.7.36.jar:/home/thabangligavha/.m2/repository/jakarta/jms/jakarta.jms-api/2.0.3/jakarta.jms-api-2.0.3.jar:/home/thabangligavha/.m2/repository/org/apache/qpid/proton-j/0.33.10/proton-j-0.33.10.jar:/home/thabangligavha/.m2/repository/io/netty/netty-buffer/4.1.75.Final/netty-buffer-4.1.75.Final.jar:/home/thabangligavha/.m2/repository/io/netty/netty-common/4.1.75.Final/netty-common-4.1.75.Final.jar:/home/thabangligavha/.m2/repository/io/netty/netty-handler/4.1.75.Final/netty-handler-4.1.75.Final.jar:/home/thabangligavha/.m2/repository/io/netty/netty-resolver/4.1.75.Final/netty-resolver-4.1.75.Final.jar:/home/thabangligavha/.m2/repository/io/netty/netty-codec/4.1.75.Final/netty-codec-4.1.75.Final.jar:/home/thabangligavha/.m2/repository/io/netty/netty-transport/4.1.75.Final/netty-transport-4.1.75.Final.jar:/home/thabangligavha/.m2/repository/io/netty/netty-transport-native-epoll/4.1.75.Final/netty-transport-native-epoll-4.1.75.Final-linux-x86_64.jar:/home/thabangligavha/.m2/repository/io/netty/netty-transport-native-unix-common/4.1.75.Final/netty-transport-native-unix-common-4.1.75.Final.jar:/home/thabangligavha/.m2/repository/io/netty/netty-transport-classes-epoll/4.1.75.Final/netty-transport-classes-epoll-4.1.75.Final.jar:/home/thabangligavha/.m2/repository/io/netty/netty-transport-native-kqueue/4.1.75.Final/netty-transport-native-kqueue-4.1.75.Final-osx-x86_64.jar:/home/thabangligavha/.m2/repository/io/netty/netty-transport-classes-kqueue/4.1.75.Final/netty-transport-classes-kqueue-4.1.75.Final.jar:/home/thabangligavha/.m2/repository/io/netty/netty-codec-http/4.1.75.Final/netty-codec-http-4.1.75.Final.jar example.Listener
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Exception in thread "main" javax.jms.JMSException: finishConnect(..) failed: Connection refused: localhost/127.0.0.1:5672
    at org.apache.qpid.jms.provider.ProviderException.toJMSException(ProviderException.java:34)
    at org.apache.qpid.jms.exceptions.JmsExceptionSupport.create(JmsExceptionSupport.java:80)
    at org.apache.qpid.jms.exceptions.JmsExceptionSupport.create(JmsExceptionSupport.java:112)
    at org.apache.qpid.jms.JmsConnection.connect(JmsConnection.java:176)
    at org.apache.qpid.jms.JmsConnectionFactory.createConnection(JmsConnectionFactory.java:213)
    at example.Listener.main(Listener.java:38)
Caused by: org.apache.qpid.jms.provider.exceptions.ProviderIOException: finishConnect(..) failed: Connection refused: localhost/127.0.0.1:5672
    at org.apache.qpid.jms.provider.exceptions.ProviderExceptionSupport.createOrPassthroughFatal(ProviderExceptionSupport.java:46)
    at org.apache.qpid.jms.provider.amqp.AmqpProvider.connect(AmqpProvider.java:309)
    at org.apache.qpid.jms.JmsConnection.connect(JmsConnection.java:162)
    ... 2 more

排查步骤 #

• 确认ActiveMQ服务状态：检查ActiveMQ broker是否已启动，未启动则执行启动脚本（如Linux下./activemq start），启动后查看控制台确认服务正常。
• 检查AMQP连接器配置：打开ActiveMQ安装目录下的conf/activemq.xml，确认存在AMQP端口的连接器配置，示例如下：

  <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
  

  若缺失该配置，添加后重启ActiveMQ。
• 验证端口占用情况：使用端口检测命令确认5672端口是否被其他进程占用：
  • Linux：netstat -tulpn | grep 5672
  • Windows：netstat -ano | findstr :5672
    若端口被占用，终止占用进程或修改ActiveMQ的AMQP端口。
• 检查防火墙设置：确认本地防火墙未拦截5672端口的TCP连接，可临时关闭防火墙测试连通性。
• 核对连接参数：确认代码中的用户名、密码与ActiveMQ配置一致，检查conf/users.properties和conf/groups.properties中是否存在admin用户且权限正确。

内容的提问来源于stack exchange，提问作者Thabang Ligavha