C++实现rabbitmq生产者消费者

RabbitMQ是一个开源的消息队列系统，它实现了高级消息队列协议（AMQP），

特点

• 可靠性：通过持久化、镜像队列等机制保证消息不丢失，确保消息可靠传递。
• 灵活的路由：提供多种路由方式，如直连、主题、扇形等，可根据不同的业务需求进行灵活配置。
• 多语言支持：支持多种编程语言，如Java、Python、C++、Ruby等，方便不同技术栈的开发者使用。
• 高可用性：可以通过集群方式实现高可用性，避免单点故障，保证系统的稳定运行。
• 易于管理：提供了可视化的管理界面，方便管理员对队列、交换机、消息等进行监控和管理。

核心概念

• 消息：应用程序之间传递的数据单元。
• 队列：用于存储消息的缓冲区，消息会在队列中等待消费者来获取。
• 交换机：接收生产者发送的消息，并根据路由键将消息路由到相应的队列。
• 绑定：将交换机和队列通过路由键进行关联，建立路由规则。
• 生产者：负责发送消息到RabbitMQ服务器的应用程序。
• 消费者：从RabbitMQ服务器获取消息并进行处理的应用程序。

工作原理

1. 生产者将消息发送到交换机，消息中包含路由键等信息。
2. 交换机根据路由键和绑定规则，将消息路由到对应的队列。
3. 消费者从队列中获取消息并进行处理。

应用场景

• 异步处理：将一些耗时的操作如发送邮件、生成报表等放在消息队列中，由消费者异步处理，提高系统的响应速度。
• 系统解耦：不同模块之间通过消息队列进行通信，降低模块之间的耦合度，使得各个模块可以独立扩展和维护。
• 流量削峰：在高并发场景下，将大量的请求放入消息队列中，由消费者按照一定的速度进行处理，避免系统因瞬间高流量而崩溃。

C++实现rabbitmq生产者消费者

// MFCRABBITMQDlg.h: 头文件
//#pragma once
#include <iostream>
#include <string>
#include <amqp.h>
#include <amqp_tcp_socket.h>
#pragma comment(lib , "rabbitmq.lib")// CMFCRABBITMQDlg 对话框
class CMFCRABBITMQDlg : public CDialogEx
{
// 构造
public:CMFCRABBITMQDlg(CWnd* pParent = nullptr);	// 标准构造函数// 对话框数据
#ifdef AFX_DESIGN_TIMEenum { IDD = IDD_MFCRABBITMQ_DIALOG };
#endifprotected:virtual void DoDataExchange(CDataExchange* pDX);	// DDX/DDV 支持// 实现
protected:HICON m_hIcon;// 生成的消息映射函数virtual BOOL OnInitDialog();afx_msg void OnSysCommand(UINT nID, LPARAM lParam);afx_msg void OnPaint();afx_msg HCURSOR OnQueryDragIcon();DECLARE_MESSAGE_MAP()
public:afx_msg void OnBnClickedButtonProduct();void CMFCRABBITMQDlg::sendMessageToRabbitMQ(const std::string& message);void CMFCRABBITMQDlg::receiveMessagesFromRabbitMQ();afx_msg void OnBnClickedButtonConsumer();void CMFCRABBITMQDlg::showLog(CString p_str_log);CString m_CstrLog;CEdit m_ctrlLog;
};

// MFCRABBITMQDlg.cpp: 实现文件
//#include "pch.h"
#include "framework.h"
#include "MFCRABBITMQ.h"
#include "MFCRABBITMQDlg.h"
#include "afxdialogex.h"#ifdef _DEBUG
#define new DEBUG_NEW
#endif// 用于应用程序“关于”菜单项的 CAboutDlg 对话框class CAboutDlg : public CDialogEx
{
public:CAboutDlg();// 对话框数据
#ifdef AFX_DESIGN_TIMEenum { IDD = IDD_ABOUTBOX };
#endifprotected:virtual void DoDataExchange(CDataExchange* pDX);    // DDX/DDV 支持// 实现
protected:DECLARE_MESSAGE_MAP()
};CAboutDlg::CAboutDlg() : CDialogEx(IDD_ABOUTBOX)
{
}void CAboutDlg::DoDataExchange(CDataExchange* pDX)
{CDialogEx::DoDataExchange(pDX);
}BEGIN_MESSAGE_MAP(CAboutDlg, CDialogEx)
END_MESSAGE_MAP()// CMFCRABBITMQDlg 对话框CMFCRABBITMQDlg::CMFCRABBITMQDlg(CWnd* pParent /*=nullptr*/): CDialogEx(IDD_MFCRABBITMQ_DIALOG, pParent)
{m_hIcon = AfxGetApp()->LoadIcon(IDR_MAINFRAME);
}void CMFCRABBITMQDlg::DoDataExchange(CDataExchange* pDX)
{CDialogEx::DoDataExchange(pDX);DDX_Control(pDX, IDC_EDIT_LOG, m_ctrlLog);
}BEGIN_MESSAGE_MAP(CMFCRABBITMQDlg, CDialogEx)ON_WM_SYSCOMMAND()ON_WM_PAINT()ON_WM_QUERYDRAGICON()ON_BN_CLICKED(IDC_BUTTON_PRODUCT, &CMFCRABBITMQDlg::OnBnClickedButtonProduct)ON_BN_CLICKED(IDC_BUTTON_CONSUMER, &CMFCRABBITMQDlg::OnBnClickedButtonConsumer)
END_MESSAGE_MAP()// CMFCRABBITMQDlg 消息处理程序BOOL CMFCRABBITMQDlg::OnInitDialog()
{CDialogEx::OnInitDialog();// 将“关于...”菜单项添加到系统菜单中。// IDM_ABOUTBOX 必须在系统命令范围内。ASSERT((IDM_ABOUTBOX & 0xFFF0) == IDM_ABOUTBOX);ASSERT(IDM_ABOUTBOX < 0xF000);CMenu* pSysMenu = GetSystemMenu(FALSE);if (pSysMenu != nullptr){BOOL bNameValid;CString strAboutMenu;bNameValid = strAboutMenu.LoadString(IDS_ABOUTBOX);ASSERT(bNameValid);if (!strAboutMenu.IsEmpty()){pSysMenu->AppendMenu(MF_SEPARATOR);pSysMenu->AppendMenu(MF_STRING, IDM_ABOUTBOX, strAboutMenu);}}// 设置此对话框的图标。  当应用程序主窗口不是对话框时，框架将自动//  执行此操作SetIcon(m_hIcon, TRUE);			// 设置大图标SetIcon(m_hIcon, FALSE);		// 设置小图标// TODO: 在此添加额外的初始化代码return TRUE;  // 除非将焦点设置到控件，否则返回 TRUE
}void CMFCRABBITMQDlg::OnSysCommand(UINT nID, LPARAM lParam)
{if ((nID & 0xFFF0) == IDM_ABOUTBOX){CAboutDlg dlgAbout;dlgAbout.DoModal();}else{CDialogEx::OnSysCommand(nID, lParam);}
}// 如果向对话框添加最小化按钮，则需要下面的代码
//  来绘制该图标。  对于使用文档/视图模型的 MFC 应用程序，
//  这将由框架自动完成。void CMFCRABBITMQDlg::OnPaint()
{if (IsIconic()){CPaintDC dc(this); // 用于绘制的设备上下文SendMessage(WM_ICONERASEBKGND, reinterpret_cast<WPARAM>(dc.GetSafeHdc()), 0);// 使图标在工作区矩形中居中int cxIcon = GetSystemMetrics(SM_CXICON);int cyIcon = GetSystemMetrics(SM_CYICON);CRect rect;GetClientRect(&rect);int x = (rect.Width() - cxIcon + 1) / 2;int y = (rect.Height() - cyIcon + 1) / 2;// 绘制图标dc.DrawIcon(x, y, m_hIcon);}else{CDialogEx::OnPaint();}
}//当用户拖动最小化窗口时系统调用此函数取得光标
//显示。
HCURSOR CMFCRABBITMQDlg::OnQueryDragIcon()
{return static_cast<HCURSOR>(m_hIcon);
}void CMFCRABBITMQDlg::sendMessageToRabbitMQ(const std::string& message)
{amqp_connection_state_t conn;amqp_socket_t* socket = NULL;int status;// 初始化连接conn = amqp_new_connection();socket = amqp_tcp_socket_new(conn);if (!socket) {std::cerr << "Failed to create TCP socket" << std::endl;return;}// 连接到 RabbitMQ 服务器status = amqp_socket_open(socket, "localhost", 5672);if (status) {std::cerr << "Failed to open TCP socket" << std::endl;amqp_destroy_connection(conn);return;}// 登录到 RabbitMQ 服务器amqp_rpc_reply_t loginReply = amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest");if (loginReply.reply_type != AMQP_RESPONSE_NORMAL) {std::cerr << "Failed to login to RabbitMQ" << std::endl;amqp_destroy_connection(conn);return;}// 打开通道amqp_channel_open(conn, 1);amqp_rpc_reply_t openReply = amqp_get_rpc_reply(conn);if (openReply.reply_type != AMQP_RESPONSE_NORMAL) {std::cerr << "Failed to open channel" << std::endl;amqp_destroy_connection(conn);return;}// 声明队列amqp_queue_declare(conn, 1, amqp_cstring_bytes("my_queue"), 0, 0, 0, 0, amqp_empty_table);amqp_rpc_reply_t queueDeclareReply = amqp_get_rpc_reply(conn);if (queueDeclareReply.reply_type != AMQP_RESPONSE_NORMAL) {std::cerr << "Failed to declare queue" << std::endl;amqp_destroy_connection(conn);return;}// 发布消息到队列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; // 持久化消息amqp_basic_publish(conn,1,amqp_cstring_bytes(""),  // 默认交换器amqp_cstring_bytes("my_queue"),0,0,&props,amqp_cstring_bytes(message.c_str()));// 检查发布结果amqp_rpc_reply_t publishReply = amqp_get_rpc_reply(conn);if (publishReply.reply_type != AMQP_RESPONSE_NORMAL) {std::cerr << "Failed to publish message" << std::endl;}else {std::cout << "Message sent: " << message << std::endl;}// 关闭通道和连接amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS);amqp_connection_close(conn, AMQP_REPLY_SUCCESS);amqp_destroy_connection(conn);
}void CMFCRABBITMQDlg:: receiveMessagesFromRabbitMQ()
{amqp_connection_state_t conn;amqp_socket_t* socket = NULL;int status;// 初始化连接conn = amqp_new_connection();socket = amqp_tcp_socket_new(conn);if (!socket) {std::cerr << "Failed to create TCP socket" << std::endl;return;}// 连接到 RabbitMQ 服务器status = amqp_socket_open(socket, "localhost", 5672);if (status) {std::cerr << "Failed to open TCP socket" << std::endl;amqp_destroy_connection(conn);return;}// 登录到 RabbitMQ 服务器amqp_rpc_reply_t loginReply = amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest");if (loginReply.reply_type != AMQP_RESPONSE_NORMAL) {std::cerr << "Failed to login to RabbitMQ" << std::endl;amqp_destroy_connection(conn);return;}// 打开通道amqp_channel_open(conn, 1);amqp_rpc_reply_t openReply = amqp_get_rpc_reply(conn);if (openReply.reply_type != AMQP_RESPONSE_NORMAL) {std::cerr << "Failed to open channel" << std::endl;amqp_destroy_connection(conn);return;}// 声明队列amqp_queue_declare(conn, 1, amqp_cstring_bytes("my_queue"), 0, 0, 0, 0, amqp_empty_table);amqp_rpc_reply_t queueDeclareReply = amqp_get_rpc_reply(conn);if (queueDeclareReply.reply_type != AMQP_RESPONSE_NORMAL) {std::cerr << "Failed to declare queue" << std::endl;amqp_destroy_connection(conn);return;}// 开始消费消息amqp_basic_consume(conn, 1, amqp_cstring_bytes("my_queue"), amqp_empty_bytes, 0, 1, 0, amqp_empty_table);amqp_rpc_reply_t consumeReply = amqp_get_rpc_reply(conn);if (consumeReply.reply_type != AMQP_RESPONSE_NORMAL) {std::cerr << "Failed to start consuming messages" << std::endl;amqp_destroy_connection(conn);return;}std::cout << "Waiting for messages. To exit press CTRL+C" << std::endl;showLog(_T("消费者开始监听:"));while (true) {amqp_rpc_reply_t reply;amqp_envelope_t envelope;amqp_maybe_release_buffers(conn);reply = amqp_consume_message(conn, &envelope, NULL, 0);if (AMQP_RESPONSE_NORMAL != reply.reply_type) {break;}static int t_count = 0;t_count++;std::cout << "Received message: ";char* p1 = static_cast<char*>(envelope.message.body.bytes);char* p = new char[envelope.message.body.len+1]();// strcpy_s(p , envelope.message.body.len, "11");for (size_t i = 0; i < envelope.message.body.len; ++i){p[i] = p1[i];}std::cout << std::endl;CString t_Cstr = (CString)p;CString t_CstrCount;t_CstrCount.Format(_T("%d  ->"), t_count);CString t_tep = t_CstrCount + t_Cstr;showLog(t_tep);delete[]p;amqp_destroy_envelope(&envelope);}// 关闭通道和连接amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS);amqp_connection_close(conn, AMQP_REPLY_SUCCESS);amqp_destroy_connection(conn);
}// ch:取流线程 | en:Grabbing thread
unsigned int __stdcall SendThread(void* pUser)
{if (pUser){CMFCRABBITMQDlg* Gthis = (CMFCRABBITMQDlg*)pUser;CString t_CstrSend;for (int i = 0; i < 1000000; ++i){Gthis->GetDlgItemText(IDC_EDIT_TEXT, t_CstrSend);std::string t_strSend = (CT2A)t_CstrSend;Gthis->sendMessageToRabbitMQ(t_strSend);Sleep(1);}return 0;}return -1;
}void CMFCRABBITMQDlg::OnBnClickedButtonProduct()
{unsigned int nThreadID = 0;void* m_hGrabThread = (void*)_beginthreadex(NULL, 0, SendThread, this, 0, &nThreadID);if (NULL == m_hGrabThread){showLog(_T("消费者创建线程失败"));return;}}// ch:取流线程 | en:Grabbing thread
unsigned int __stdcall ConsumThread(void* pUser)
{if (pUser){CMFCRABBITMQDlg* Gthis = (CMFCRABBITMQDlg*)pUser;Gthis->receiveMessagesFromRabbitMQ();return 0;}return -1;
}void CMFCRABBITMQDlg::OnBnClickedButtonConsumer()
{unsigned int nThreadID = 0;void* m_hGrabThread = (void*)_beginthreadex(NULL, 0, ConsumThread, this, 0, &nThreadID);if (NULL == m_hGrabThread){showLog(_T("消费者创建线程失败"));return;}
}void CMFCRABBITMQDlg::showLog(CString p_str_log)
{int t_intLine = m_ctrlLog.GetLineCount();if (t_intLine == 1000){m_CstrLog = _T("");}SYSTEMTIME timeCur;GetLocalTime(&timeCur);char t_logbuffer[1024] = { 0 };sprintf_s(t_logbuffer, ("[%04d%02d%02d_%02d:%02d:%02d:%03d] "), timeCur.wYear, timeCur.wMonth, timeCur.wDay, timeCur.wHour, timeCur.wMinute, timeCur.wSecond, timeCur.wMilliseconds);p_str_log += "\r\n";m_CstrLog += t_logbuffer;m_CstrLog += p_str_log;m_ctrlLog.SetWindowTextW(m_CstrLog);m_ctrlLog.LineScroll(m_ctrlLog.GetLineCount() - 1, 0);}