C++ - 仿 RabbitMQ 实现消息队列--案例

实现生产者客户端

#include "connection.hpp"

int main()
{
    jiuqi::AsyncWorker::ptr worker = std::make_shared<jiuqi::AsyncWorker>();
    jiuqi::Connection::ptr client = std::make_shared<jiuqi::Connection>("127.0.0.1", 8080, worker);
    jiuqi::Channel::ptr channel = client->openChannel();
    google::protobuf::Map<std::string, std::string> map;
    channel->declareExchange("exchange1", jiuqi::ExchangeType::TOPIC, true, false, map);
    channel->declareQueue("queue1", true, false, false, map);
    channel->declareQueue("queue2", true, false, false, map);
    channel->queueBind("exchange1", "queue1", "queue1");
    channel->queueBind("exchange1", "queue2", "news.music.#");

    for (int i = 0; i < 10; i++)
    {
        jiuqi::BasicProperties bp;
        bp.set_id(jiuqi::UUIDHelper::uuid());
        bp.set_deliver_mode(jiuqi::DelivertMode::DURABLE);
        bp.set_routing_key("news.music.pop");
        channel->basicPublish("exchange1", &bp, "hello world");
    }
    client->closeChannel(channel);
    return 0;
}

实现消费者客户端

#include "connection.hpp"

void cb(jiuqi::Channel::ptr &channel, const std::string &ctag, const jiuqi::BasicProperties *bp, const std::string &body)
{
    std::cout << ctag << "消费了: " << body << std::endl;
    channel->basicAck(bp->id());
}

int main(int argc, char *argv[])
{
    if (argc != 2)
    {
        std::cout << "usage: ./consumer_client queuename\n";
        return -1;
    }
    jiuqi::AsyncWorker::ptr worker = std::make_shared<jiuqi::AsyncWorker>();
    jiuqi::Connection::ptr client = std::make_shared<jiuqi::Connection>("127.0.0.1", 8080, worker);
    jiuqi::Channel::ptr channel = client->openChannel();
    google::protobuf::Map<std::string, std::string> map;
    channel->declareExchange("exchange1", jiuqi::ExchangeType::TOPIC, true, false, map);
    channel->declareQueue("queue1", true, false, false, map);
    channel->declareQueue("queue2", true, false, false, map);
    channel->queueBind("exchange1", "queue1", "queue1");
    channel->queueBind("exchange1", "queue2", "news.music.#");

    auto func = std::bind(cb, channel, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
    channel->basicConsumer("consumer1", argv[1], false, func);

    std::this_thread::sleep_for(std::chrono::seconds(100));

    client->closeChannel(channel);

    return 0;
}