RabbitMQ Kubernetes集群Java Stream客户端生产消息报错求助

问题核心原因 #

报错是因为客户端尝试直接连接RabbitMQ集群节点的5552端口（Stream协议默认端口），但该端口未正确暴露，或客户端未配置正确的集群节点发现逻辑。创建流操作正常是因为它依赖AMQP协议（默认5672端口），而生产流消息需要直接使用Stream端口。

解决方案步骤 #

1. 修改RabbitMQ集群配置，暴露Stream端口 #

更新你的RabbitmqCluster YAML配置，在service字段下添加additionalPorts，显式暴露Stream的5552端口：

apiVersion: rabbitmq.com/v1beta1
kind: RabbitmqCluster
metadata:
  name: rabbitmq-deployment
  namespace: rabbitmq-namespace
spec:
  replicas: 2
  image: rabbitmq:3.11.13
  persistence:
    storage: 20Gi
  service:
    type: LoadBalancer
    additionalPorts:
      - name: stream
        port: 5552
        targetPort: 5552
  rabbitmq:
    additionalPlugins:
      - rabbitmq_stream
      - rabbitmq_stream_management

应用更新：

kubectl apply -f <你的配置文件路径> -n rabbitmq-namespace

2. 调整Java客户端的集群发现配置 #

RabbitMQ Stream客户端需要正确识别集群节点，在K8s环境下可通过以下两种方式配置：

方式一：同K8s集群内客户端配置

如果Java应用部署在同一个K8s集群中，将RABBITMQ_HOST环境变量设置为RabbitMQ节点的headless服务域名：rabbitmq-deployment-nodes.rabbitmq-namespace.svc.cluster.local，并启用云原生地址解析：

EnvironmentBuilder environmentBuilder = Environment.builder();
environmentBuilder.host(System.getenv("RABBITMQ_HOST"));
environmentBuilder.port(Integer.parseInt(System.getenv("RABBITMQ_STREAM_PORT")));
environmentBuilder.username(System.getenv("RABBITMQ_USERNAME"));
environmentBuilder.password(System.getenv("RABBITMQ_PASSWORD"));
// 启用K8s环境下的节点自动发现
environmentBuilder.addressResolver(AddressResolver.cloudNative());
mainConnection = environmentBuilder.build();

方式二：外部客户端配置

如果Java应用在K8s集群外，确保LoadBalancer的5552端口已对外开放，将RABBITMQ_HOST设为LoadBalancer的外部IP，同时保留AddressResolver.cloudNative()配置，客户端会自动通过DNS发现集群节点。

3. 验证端口监听与网络连通性 #

• 进入RabbitMQ Pod验证Stream端口是否正常监听：

  kubectl exec -it rabbitmq-deployment-server-0 -n rabbitmq-namespace -- rabbitmq-diagnostics listeners
  

  检查输出是否包含5552端口的监听记录。
• 测试网络连通性：
  同集群内应用：

  nc -zv rabbitmq-deployment-server-0.rabbitmq-deployment-nodes.rabbitmq-namespace 5552
  

  外部应用：

  nc -zv <LoadBalancer外部IP> 5552
  

4. 重新测试生产逻辑 #

调整配置后，重新运行生产代码，确认不再抛出连接异常，消息能正常被Broker确认。

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