安装mq
https://blog.csdn.net/zl_momomo/article/details/82986368
需要安裝rabbitmq-server
开启rabbitmq服务
systemctl start rabbitmq-server
systemctl enable rabbitmq-server.
客户端 amqp_sendstring.c
include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <amqp.h>
// #include <rabbitmq-c/tcp_socket.h>
// #include "utils.h"
int main(int argc, char const* const* argv)
{
char const* hostname;
int port;
char const* exchange;
char const* routingkey;
char const* messagebody;
amqp_socket_t* socket = NULL;
amqp_connection_state_t conn;
amqp_rpc_reply_t rpc_reply;
hostname = "localhost";
port = 5672;
exchange = "amq.topic";
routingkey = "test";
messagebody = argv[1];
conn = amqp_new_connection();
if (NULL == (socket = amqp_tcp_socket_new(conn)))
{
printf("creating TCP socket\n");
}
if (AMQP_STATUS_OK != amqp_socket_open(socket, hostname, port))
{
printf("opening TCP socket\n");
}
rpc_reply = amqp_login(conn, "/", 0, AMQP_DEFAULT_FRAME_SIZE, AMQP_DEFAULT_HEARTBEAT, AMQP_SASL_METHOD_PLAIN, "guest", "guest");
if (AMQP_RESPONSE_NORMAL != rpc_reply.reply_type)
printf("xxxxxxxx\n");
// die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in");
if (NULL == amqp_channel_open(conn, 1))
printf("xxxxxxxx2\n");
rpc_reply = amqp_get_rpc_reply(conn);
if (AMQP_RESPONSE_NORMAL != rpc_reply.reply_type)
printf("Opening channel\n");
else
{
amqp_basic_properties_t props;
props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
props.content_type = amqp_cstring_bytes("text/plain");
props.delivery_mode = 2; /* persistent delivery mode */
if (0 != amqp_basic_publish(conn, 1, amqp_cstring_bytes(exchange), amqp_cstring_bytes(routingkey), 0, 0, &props, amqp_cstring_bytes(messagebody)))
printf("Publishing\n");
}
rpc_reply = amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS);
if (AMQP_RESPONSE_NORMAL != rpc_reply.reply_type)
printf("Closing channel\n");
rpc_reply = amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
if (AMQP_RESPONSE_NORMAL != rpc_reply.reply_type)
printf("Closing connection");
amqp_destroy_connection(conn);
return 0;
}
服务端 amqp_listen.c
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <amqp.h>
int main(int argc, char const* const* argv)
{
char const* hostname;
int port, status;
char const* exchange;
char const* bindingkey;
amqp_socket_t* socket = NULL;
amqp_connection_state_t conn;
amqp_rpc_reply_t rpc_reply;
amqp_bytes_t queuename;
hostname = "localhost";
port = 5672;
exchange = "amq.topic";
bindingkey = "test";
conn = amqp_new_connection();
socket = amqp_tcp_socket_new(conn);
if (!socket)
{
printf("creating TCP socket\n");
}
if (AMQP_STATUS_OK != amqp_socket_open(socket, hostname, port))
{
printf("opening TCP socket\n");
}
rpc_reply = amqp_login(conn, "/", 0, AMQP_DEFAULT_FRAME_SIZE, AMQP_DEFAULT_HEARTBEAT, AMQP_SASL_METHOD_PLAIN, "guest", "guest");
if (AMQP_RESPONSE_NORMAL != rpc_reply.reply_type)
printf("xxxxxxxx\n");
if (NULL == amqp_channel_open(conn, 1))
printf("xxxxxxxx2\n");
// die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
// "guest", "guest"),
// "Logging in");
// amqp_channel_open(conn, 1);
// die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");
rpc_reply = amqp_get_rpc_reply(conn);
if (AMQP_RESPONSE_NORMAL != rpc_reply.reply_type)
printf("Opening channel\n");
else
{
amqp_queue_declare_ok_t* r = amqp_queue_declare(conn, 1, amqp_empty_bytes, 0, 0, 0, 1, amqp_empty_table);
// die_on_amqp_error(amqp_get_rpc_reply(conn), "Declaring queue");
rpc_reply = amqp_get_rpc_reply(conn);
if (AMQP_RESPONSE_NORMAL != rpc_reply.reply_type)
printf("Declaring queue\n");
queuename = amqp_bytes_malloc_dup(r->queue);
if (queuename.bytes == NULL)
{
fprintf(stderr, "Out of memory while copying queue name");
return 1;
}
}
amqp_queue_bind(conn, 1, queuename, amqp_cstring_bytes(exchange), amqp_cstring_bytes(bindingkey), amqp_empty_table);
// die_on_amqp_error(amqp_get_rpc_reply(conn), "Binding queue");
rpc_reply = amqp_get_rpc_reply(conn);
if (AMQP_RESPONSE_NORMAL != rpc_reply.reply_type)
printf("Binding queue\n");
amqp_basic_consume(conn, 1, queuename, amqp_empty_bytes, 0, 1, 0, amqp_empty_table);
// die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
rpc_reply = amqp_get_rpc_reply(conn);
if (AMQP_RESPONSE_NORMAL != rpc_reply.reply_type)
printf("Consuming\n");
else
{
for (;;)
{
amqp_rpc_reply_t res;
amqp_envelope_t envelope;
amqp_maybe_release_buffers(conn);
res = amqp_consume_message(conn, &envelope, NULL, 0);
if (AMQP_RESPONSE_NORMAL != res.reply_type)
{
break;
}
// printf("Delivery %u, exchange %.*s routingkey %.*s\n",
// (unsigned)envelope.delivery_tag, (int)envelope.exchange.len,
// (char*)envelope.exchange.bytes, (int)envelope.routing_key.len,
// (char*)envelope.routing_key.bytes);
// if (envelope.message.properties._flags & AMQP_BASIC_CONTENT_TYPE_FLAG)
// {
// printf("Content-type: %.*s\n",
// (int)envelope.message.properties.content_type.len,
// (char*)envelope.message.properties.content_type.bytes);
// }
printf("----\n");
printf("===%s\n\n", envelope.message.body.bytes);
// amqp_dump(envelope.message.body.bytes, envelope.message.body.len);
amqp_destroy_envelope(&envelope);
}
}
amqp_bytes_free(queuename);
rpc_reply = amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS);
if (AMQP_RESPONSE_NORMAL != rpc_reply.reply_type)
printf("Closing channel\n");
rpc_reply = amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
if (AMQP_RESPONSE_NORMAL != rpc_reply.reply_type)
printf("Closing connection");
amqp_destroy_connection(conn);
return 0;
}
编译:
gcc amqp_listen.c utils.c -lrabbitmq -o s
gcc amqp_sendstring.c utils.c -lrabbitmq -o c
测试: