基于libuv与AMQP-CPP的C++连接RabbitMQ报错求助

我想通过libuv和AMQP-CPP在C++里给RabbitMQ发送消息，Windows环境下AMQP-CPP的TCP通信似乎存在问题，因此自己编写了如下代码：

static void alloc_cb(uv_handle_t* handle, size_t size, uv_buf_t* buf) {
  *buf = uv_buf_init((char *)malloc(size), size);
}

void on_close(uv_handle_t* handle)
{
  printf("closed.");
}


void on_write(uv_write_t* req, int status)
{
  if (status) {
    perror("uv_write error ");
    return;
  }
  printf("wrote.\n");
  free(req);
  //uv_close((uv_handle_t*)req->handle, on_close);
}

void on_read(uv_stream_t* tcp, ssize_t nread, const uv_buf_t* buf)
{
  printf("on_read. %p\n", tcp);
  if (nread >= 0) {
    //printf("read: %s\n", tcp->data);
    printf("read: %s\n", buf->base);
  }
  else {
    //we got an EOF
    uv_close((uv_handle_t*)tcp, on_close);
  }

  //cargo-culted
  free(buf->base);
}

void write2(uv_stream_t* stream, char* data, int len2) {
  uv_buf_t buffer[1];
  buffer[0].base = data;
  buffer[0].len = len2;
  uv_write_t* req = (uv_write_t*)malloc(sizeof(uv_write_t));
  uv_write(req, stream, buffer, 1, on_write);
}


void on_connect(uv_connect_t* connection, int status)
{
  if (status < 0) {
    printf("failed to connect\n"); return;
  }
  printf("connected. %p %d\n", connection, status);

  uv_stream_t* stream = connection->handle;
  free(connection);
  write2(stream, (char*)"test!", 10);
  uv_read_start(stream, alloc_cb, on_read);
}

int main() {
  uv_loop_t* loop = uv_default_loop();

  uv_tcp_t* pSock = (uv_tcp_t*)malloc(sizeof(uv_tcp_t));
  uv_tcp_init(loop, pSock);
  uv_tcp_keepalive(pSock, 1, 60);

  struct sockaddr_in dest;
  uv_ip4_addr("10.10.50.242", 5672, &dest);

  uv_connect_t* pConn = (uv_connect_t *)malloc(sizeof(uv_connect_t));
  printf("allocated %p\n", pConn);

  uv_tcp_connect(pConn, pSock, (const struct sockaddr*)&dest, on_connect);

  uv_run(loop, UV_RUN_DEFAULT);
}

运行后，RabbitMQ日志中出现如下错误：

2023-02-28 15:49:10.901000+09:00 [info] <0.17112.1> accepting AMQP connection <0.17112.1> (ipaddr:51509 -> ipaddr:5672)
2023-02-28 15:49:10.901000+09:00 [error] <0.17112.1> closing AMQP connection <0.17112.1> (ipaddr:51509 -> ipaddr:5672):
2023-02-28 15:49:10.901000+09:00 [error] <0.17112.1> {bad_header,<<116,101,115,116,33,0,0,0>>}

错误原因分析 #

• 你直接发送了字符串"test!"，但RabbitMQ遵循AMQP协议，连接建立后第一步必须发送符合AMQP规范的协议头，而非任意文本。
• 日志里的{bad_header,<<116,101,115,116,33,0,0,0>>}是将你发送的内容转成了ASCII码：116对应t、101对应e、115对应s、116对应t、33对应!，后面的0是因为你指定了长度10，超出"test!"（5个字符）的部分是内存中的随机0值，这完全不符合AMQP的协议头要求。
• AMQP协议头有固定格式，比如AMQP 0-9-1的协议头是AMQP\x00\x00\x09\x01，必须先发送这个正确的协议头，RabbitMQ才会认可连接，之后才能进行认证、声明队列、发送消息等操作。
• 手动编写TCP通信对接RabbitMQ需要实现AMQP协议的所有细节，对新手来说难度极高。建议还是使用AMQP-CPP库处理协议层，仅将libuv作为它的事件循环后端，而非从零实现TCP通信。

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