消费者到 RabbitMQ 流 - 错误标头/命令未实现 0 buff:0

Question

我在 docker-compose 中有这个配置：

services:

  rabbitmq:
    image: rabbitmq:3-management
    volumes:
      - rabbitmq_data:/var/lib/rabbitmq
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      - RABBITMQ_PLUGINS=rabbitmq_stream   // stream plugin enabled?
      - RABBITMQ_DEFAULT_USER=rabbitmq
      - RABBITMQ_DEFAULT_PASS=rabbitmq

我的流消费者看起来像：

package main

import (
    "fmt"
    "log"

    "github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
    "github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
)

func main() {
    // Create an environment
    env, err := stream.NewEnvironment(
        stream.NewEnvironmentOptions().
            SetHost("localhost").
            SetPort(5672).           // 5552
            SetUser("rabbitmq").     // Replace with your username
            SetPassword("rabbitmq"), // Replace with your password
    )

    if err != nil {
        log.Fatalf("Failed to create environment: %s", err)
    }

    // Define a handler for incoming messages
    handleMessage := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
        for _, z := range message.Data {
            fmt.Println("Received message:", string(z))
        }
    }

    // Create a consumer
    _, err = env.NewConsumer(
        "your_stream_name",
        handleMessage,
        stream.NewConsumerOptions().SetConsumerName("my_consumer"),
    )

    if err != nil {
        log.Fatalf("Failed to create consumer: %s", err)
    }

    select {}
}

但我收到此错误：

2023/12/14 11:03:16 [warn] - Command not implemented 0 buff:0 
2023/12/14 11:03:26 [error] - timeout 10000 ns - waiting Code, operation: commandPeerProperties
2023/12/14 11:03:26 [error] - Can't set the peer-properties. Check if the stream server is running/reachable
2023/12/14 11:03:26 Failed to create environment: timeout 10000 ms - waiting Code, operation: commandPeerProperties

在 RabbitMQ 服务器日志中，我看到以下内容：

wss-rabbitmq-1 | 2023-12-14 16:01:07.964018+00:00 [信息] <0.9.0> 时间 启动 RabbitMQ：30736437 us wss-rabbitmq-1 | 2023-12-14 16:02:28.713781+00:00 [信息] <0.752.0> 接受 AMQP 连接 <0.752.0> (192.168.65.1:33513 -> 192.168.240.3:5672) wss-rabbitmq-1 | 2023-12-14 16:02:28.714007+00:00 [错误] <0.752.0> 关闭 AMQP 连接 <0.752.

wss-rabbitmq-1 | 2023-12-14 16:02:28.714007+00:00 [错误] <0.752.0> {bad_header，<<0,0,0,243,0,17,0,1>>} wss-rabbitmq-1 | 2023-12-14 16:03:16.808111+00:00 [信息] <0.781.0> 接受 AMQP 连接 <0.781.0>（192.168.65.1:33844 -> 192.168.240.3:5672）

wss-rabbitmq-1 | 2023-12-14 16:03:16.808803+00:00 [错误] <0.781.0> 关闭 AMQP 连接<0.781.0>（192.168.65.1:33844 -> 192.168.240.3:5672）： wss-rabbitmq-1 | 2023-12-14 16:03:16.808803+00:00 [错误] <0.781.0> {bad_header,<<0,0,0,243,0,17,0,1>>}0> (192.168.65.1:33513 -> 192.168.240.3:5672): 块引用

我确保使用以下方式启用了流插件：

docker exec 1d0e959696d8 rabbitmq-plugins enable rabbitmq_stream

即使 RMQ 日志显示插件已成功启用，我仍然遇到相同的错误..

Answer 1

您使用的客户端需要启用流插件。

rabbitmq-plugins enable rabbitmq_stream rabbitmq_stream_management

然后你需要暴露正确的端口

5552

（流端口）

5672

适用于 AMQP

请参阅此链接： https://www.rabbitmq.com/stream-core-plugin-comparison.html