博客

消息队列设计精要

解耦是消息队列要解决的最本质问题。

所谓解耦，简单点讲就是一个事务，只关心核心的流程。而需要依赖其他系统但不那么重要的事情，有通知即可，无需等待结果。

如果下游有很多系统关心你的系统发出的通知的时候，果断地使用消息队列吧。

掰开了揉碎了看，最简单的消息队列可以做成一个消息转发器，把一次RPC做成两次RPC。发送者把消息投递到服务端，服务端再将消息转发一手到接收端，就是这么简单。

该资源(本人整理制作，知识付费，请尊重劳动者的成果)。需要的话扫描联系本文博客头像

关于服务端的安装与配置，请参考：

Windows RabbitMQ服务端安装与配置详解

消息队列的基本功能之一是进行广播。

RabbitMQ是持久化的MQ。

随时更新rabbitmq-c最新版本！Debug | Release 版本！一个发送接口 一个接收接口 简单至极！

持续更新最新rabbitmq-c

最近更新：

Windows VS2017

登录首页：

http://localhost:15672/

如果是异地服务器，需要填写服务器的IP才能访问

RabbitMQ服务器队列

全部资源

百度网盘

最新rabbitmq-c的源码编译生成的库文件

开发者可以直接使用CRabbitMQ类来解决上传下载，接口更方便好用

发送和接收的示例代码

#include <iostream>
#include <string>
#include <sstream>
#include <vector>
#include "MessageQueue/RabbitMQ.h"
#include "MessageQueue/MessageBody.h"
#include <ctime>
using namespace std;


//--------------------------需要开发者自己填写的信息begin--------------------------------------
//服务端的一个Exchange，客户端发送的时候，如果没有则会创建；客户端取的时候，如果没有则取不到数据
CExchange my_test_exchange("MyTestExchange");
//服务端的一个消息队列，客户端发送的时候，如果没有则会创建；客户端取的时候，如果没有则取不到数据
CQueue my_test_queue("MyTestQueue");
//服务端Exchange与消息队列之间的绑定key，客户端发送的时候，如果没有则不能发送；客户端取的时候，如果没有则取不到数据
string my_test_bind_key("MyTestBindKey");
//RabbitMQ服务端IP
string rabbitmq_server_ip = "127.0.0.1";
//RabbitMQ服务端端口
int rabbitmq_server_port = 5672;
//RabbitMQ服务端登录用户名
string rabbitmq_server_user = "guest";
//RabbitMQ服务端登录密码
string rabbitmq_server_psw = "guest";
//--------------------------需要开发者自己填写的信息end--------------------------------------


//--------------------------建议开发者设置批量操作begin--------------------------------------
//建议将此值设置成500，或者1000.这样可以节省时间和网络资源。没有必要一条一条的发送或者取
int test_send_number = 10;
//--------------------------建议开发者设置批量操作end--------------------------------------
//引用全局变量

//发送数据到RabbitMQ服务的全部过程如下
void send_message_to_mq(void)
{
		vector<CMessage> message_array;
		CMessage message("");
		//制作test_send_number个消息，用于发送到RabbitMQ服务器
		for (int i=0;i<test_send_number;++i)
		{
			ostringstream oss;
			oss<<i;
			//这里创建你要发送的数据，放到CMessage的m_data成员变量中，从服务端取的时候也是一样从这个成员中取string
			message.m_data = "Hello"+oss.str();
			message_array.push_back(message);
		}
		string& err = GetErr();    
		//RabbitMQ服务器的连接信息，这里改成你自己的IP/PORT
		CRabbitMQ pro(rabbitmq_server_ip,rabbitmq_server_port,rabbitmq_server_user,rabbitmq_server_psw);

		if(pro.Connect(err)<0)
		{
			cout<<"连接失败！"<<endl;
			return ;
		}
		else
		{
			cout<<"连接成功！"<<endl;
		}

		//声明一个交换机，声明一次，下次不需要再次声明
		if(pro.exchange_declare(my_test_exchange, err) < 0)
		{
			cout<<"声明交换机失败！"<<endl;
			return ;
		}
		else
		{
			cout<<"声明交换机成功！"<<endl;
		}
		//声明一个队列，声明一次，下次不需要再次声明
		if( (pro.queue_declare(my_test_queue, err) < 0) ) 
		{
			cout<<"声明队列失败！"<<endl;
			return ;
		}
		else
		{
			cout<<"声明队列成功！"<<endl;
		}
		//将交换机绑定到队列， 绑定一次，后面不需要再绑定
		if( (pro.queue_bind(my_test_queue,my_test_exchange,my_test_bind_key, err)<0) ) 
		{
			cout<<"绑定队列失败！"<<endl;
			return ;
		}
		else
		{
			cout<<"绑定队列成功！"<<endl;
		}
		string my_test_root_key=my_test_bind_key;
		//发送数据到RabbitMQ服务器，可以反复调用从而实现连续发送数据
		int flag = pro.publish(message_array,my_test_root_key,err);

		if(flag<0)
		{
			cout<<"投递消息到MQ失败"<<endl;
		}
		else
		{
			cout<<"投递消息到MQ成功"<<endl;
			for (vector<CMessage>::iterator itr = message_array.begin(); itr != message_array.end(); ++itr)
			{
				cout<<"投递了："<<(*itr).m_data<<endl;
			}
		}
		pro.__sleep(10);
		pro.Disconnect();
}
//从RabbitMQ取数据的全部过程如下
void get_message_from_mq(void)
{
		vector<CMessage> message_array;
		string& err = GetErr();    

		//RabbitMQ服务器的连接信息，这里改成你自己的IP/PORT
		CRabbitMQ pro(rabbitmq_server_ip,rabbitmq_server_port,rabbitmq_server_user,rabbitmq_server_psw);

		if(pro.Connect(err)<0)
		{
			cout<<"取消息时连接失败！"<<endl;
			return ;
		}
		else
		{
			cout<<"取消息时连接成功！"<<endl;
		}
		message_array.clear();
		//一次性取1000個消息，不足1000也无妨，有多少取到多少
		int get_number= test_send_number;
		::timeval tvb={0,10};
		//从RabbitMQ服务器取消息
		if(pro.consumer(my_test_queue,message_array,get_number,&tvb,err)<0)
		{
			cout<<"取消息失败！"<<endl;
		}
		else
		{
			cout<<"取消息成功！取到了"<<message_array.size()<<"个消息"<<endl;
			for (int i=0;i<message_array.size();i++)
			{
				cout<<"取到的消息："<<message_array[i].m_data<<endl;
			}
		}
		pro.__sleep(10);
		pro.Disconnect();
}

int main(int ,char**)
{
	cout<<"Hello UseMQ!"<<endl;
	//发送消息
	send_message_to_mq();
	//取消息
	get_message_from_mq();
	cout<<"Hello UseMQ end!";
	return 0;
};

Windows运行效果：

Linux运行效果：

用户反馈