Measure/Modules/MeasureNode/rabbitmq/rabbitmq_base.cpp

#include "rabbitmq_base.h"
#include "tool/time_tool.h"

Rabbitmq_base::Rabbitmq_base()
{
	m_rabbitmq_status = RABBITMQ_STATUS_UNKNOW;

	mp_connect = NULL;
	mp_socket = NULL;
	m_port = 0;

	mp_receive_analysis_thread = NULL;
	mp_send_thread = NULL;
	mp_encapsulate_status_thread = NULL;
	m_encapsulate_status_cycle_time = 1000;//默认1000ms,就自动封装一次状态信息

	check_msg_callback = NULL;
	check_executer_callback = NULL;
	execute_msg_callback = NULL;
	encapsulate_status_callback = NULL;

}

Rabbitmq_base::~Rabbitmq_base()
{
	rabbitmq_uninit();
}

//初始化 通信 模块。如下三选一
Error_manager Rabbitmq_base::rabbitmq_init()
{
    LOG(INFO) << RABBITMQ_PARAMETER_PATH;
	return  rabbitmq_init_from_protobuf(RABBITMQ_PARAMETER_PATH);
}
//初始化 通信 模块。从文件读取
Error_manager Rabbitmq_base::rabbitmq_init_from_protobuf(std::string prototxt_path)
{
	Rabbitmq_proto::Rabbitmq_parameter_all t_rabbitmq_parameter_all;
	if(!proto_tool::read_proto_param(prototxt_path,t_rabbitmq_parameter_all) )
	{
		return Error_manager(RABBITMQ_READ_PROTOBUF_ERROR,MINOR_ERROR,
							 "rabbitmq_init_from_protobuf read_proto_param  failed");
	}
    LOG(INFO) << "read rabbitmq params success.";
	return rabbitmq_init_from_protobuf(t_rabbitmq_parameter_all);
}
//初始化 通信 模块。从protobuf读取
Error_manager Rabbitmq_base::rabbitmq_init_from_protobuf(Rabbitmq_proto::Rabbitmq_parameter_all &rabbitmq_parameter_all)
{
	LOG(INFO) << " ---Rabbitmq_base::rabbitmq_init_from_protobuf() run--- "<< this;

	int t_status=0;		//状态
	amqp_rpc_reply_t t_reply;	//reply答复结果
	Error_manager t_error;
	m_rabbitmq_parameter_all = rabbitmq_parameter_all;

	//amqp_new_connection 新建amqp的连接配置,里面只有连接状态参数
	// 返回amqp_connection_state_t_ *, 函数内部分配内存, amqp_destroy_connection()可以释放内存, 内存不为空则成功
	mp_connect = amqp_new_connection();
	if ( mp_connect == NULL )
	{
		return Error_manager(Error_code::RABBITMQ_AMQP_NEW_CONNECTION_ERROR, Error_level::MINOR_ERROR,
							 "amqp_new_connection fun error ");
	}

	//amqp_tcp_socket_new 新建tcp_socket连接
	// 返回amqp_socket_t *, 函数内部分配内存, amqp_connection_close()可以释放内存, 内存不为空则成功
	mp_socket = amqp_tcp_socket_new(mp_connect);
	if ( mp_socket == NULL )
	{
		return Error_manager(Error_code::RABBITMQ_AMQP_TCP_SOCKET_NEW_ERROR, Error_level::MINOR_ERROR,
							 "amqp_tcp_socket_new fun error ");
	}

	//载入外部参数
	if (rabbitmq_parameter_all.rabbitmq_parameters().has_ip() &&
		rabbitmq_parameter_all.rabbitmq_parameters().has_port() &&
		rabbitmq_parameter_all.rabbitmq_parameters().has_user() &&
		rabbitmq_parameter_all.rabbitmq_parameters().has_password() )
	{
		m_ip = rabbitmq_parameter_all.rabbitmq_parameters().ip();
		m_port = rabbitmq_parameter_all.rabbitmq_parameters().port();
		m_user = rabbitmq_parameter_all.rabbitmq_parameters().user();
		m_password = rabbitmq_parameter_all.rabbitmq_parameters().password();
	}
	else
	{
		return Error_manager(Error_code::RABBITMQ_PROTOBUF_LOSS_ERROR, Error_level::MINOR_ERROR,
							 " rabbitmq_parameter_all.rabbitmq_parameters() The data is not complete   ");
	}

	//amqp_socket_open 打开socket连接, 输入ip和port,
	// 成功返回AMQP_STATUS_OK = 0x0, 失败返回错误状态码, 详见 enum amqp_status_enum_
	//只需要设置配置服务器的ip和port, 不需要配置子节点客户端的ip和port, 在后面配置channel通道时,进行设置.
	t_status = amqp_socket_open(mp_socket, m_ip.c_str(), m_port);
	if ( t_status != AMQP_STATUS_OK )
	{
		return Error_manager(Error_code::RABBITMQ_AMQP_SOCKET_OPEN_ERROR, Error_level::MINOR_ERROR,
							 amqp_error_to_string(t_status, "amqp_socket_open") );
	}

	//amqp_login() 登录代理服务器,
	//输入 连接参数结构体 amqp_connection_state_t,
	//输入 连接地址, 前面 amqp_socket_open() 已经输入了,这里默认写"/"
	//输入 连接通道最大值, 默认值0表示没有限制
	//输入 连接帧率最大值, 默认值是131072 (128KB)
	//输入 心跳帧之间的秒数, 默认值0禁用心跳
	//输入 身份验证模式, 	AMQP_SASL_METHOD_PLAIN, 追加用户名和密码
	//					AMQP_SASL_METHOD_EXTERNAL, 追加身份证
	//返回 结果的结构体 amqp_rpc_reply_t
	//	amqp_response_type_enum reply_type 登录成功是 AMQP_RESPONSE_NORMAL
	//		失败:如果是 reply_type == AMQP_RESPONSE_SERVER_EXCEPTION, 服务器连接错误, 错误信息在 amqp_method_t reply
	//		失败:如果是 reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION, 库函数错误, 错误信息在 int library_error
	t_reply = amqp_login(mp_connect, "/", 0, 131072, 0,
						 AMQP_SASL_METHOD_PLAIN, m_user.c_str(), m_password.c_str());
	if ( t_reply.reply_type != AMQP_RESPONSE_NORMAL )
	{
		return Error_manager(Error_code::RABBITMQ_AMQP_LOGIN_ERROR, Error_level::MINOR_ERROR,
							 amqp_error_to_string(t_reply, "amqp_login") );
	}

	//清除channel_map, 通道的缓存,防止重复开启, (channel允许重复使用, 但是不能重复初始化)
	m_channel_map.clear();

	//创建通道队列消费者, (交换机和永久队列不在代码里创建,请在服务器上手动创建)
	t_error = rabbitmq_new_channel_queue_consume(rabbitmq_parameter_all);
	if ( t_error != Error_code::SUCCESS )
	{
		return t_error;
	}

	//启动通信, 开启线程, run thread
	t_error = rabbitmq_run();
	if ( t_error != Error_code::SUCCESS )
	{
		return t_error;
	}
	return Error_code::SUCCESS;
}

//创建通道队列消费者, (交换机和永久队列不在代码里创建,请在服务器上手动创建)
Error_manager Rabbitmq_base::rabbitmq_new_channel_queue_consume(Rabbitmq_proto::Rabbitmq_parameter_all &rabbitmq_parameter_all)
{
	int t_status=0;		//状态
	amqp_rpc_reply_t t_reply;	//reply答复结果
	Error_manager t_error;

	///Rabbitmq 接受的通道,队列和消费者, 多个
	for(int i=0;i<rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_reciever_vector_size();++i)
	{
		//Rabbitmq 配置的通道,队列和消费者,
		Rabbitmq_proto::Rabbitmq_channel_queue_consume t_inf =
		rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_reciever_vector(i);

		//通道查重,防止重复开启(channel允许重复使用, 但是不能重复初始化)
		if ( m_channel_map.find(t_inf.channel()) == m_channel_map.end() )
		{
			//amqp_channel_open() 打开连接通道, 同一台电脑可以多个进程和线程进行连接服务器, 每个连接需要自己独特的通道.
			amqp_channel_open(mp_connect, t_inf.channel());
			//amqp_get_rpc_reply() 获取当前网络连接的状态结果.
			t_reply = amqp_get_rpc_reply(mp_connect);
			if ( t_reply.reply_type != AMQP_RESPONSE_NORMAL )
			{
				return Error_manager(Error_code::RABBITMQ_AMQP_CHANNEL_OPEN_ERROR, Error_level::MINOR_ERROR,
									 amqp_error_to_string(t_reply, "amqp_channel_open") );
			}
			if ( t_inf.consume_no_ack() == 0 )
			{
				//amqp_basic_qos设置通道每次只能接受一条消息, 直到该消息被ack,才能接受下一条.状态消息可以继续接受
				//uint16_t prefetch_count 同时接受消息的个数, 这里固定写1,
				//配合 amqp_basic_qos 和 amqp_basic_ack , 来阻塞这个通道的接受消息
				//注:请求消息no_ack==0, 当接受一条指令后,该通道被阻塞,其他通道仍然正常接受, 等到任务被执行完,手动调用amqp_basic_ack函数, 则可以继续接受请求消息.
				//注:状态消息no_ack==1, 当接受一条指令后,该状态消息立刻被删除,然后可以继续接受下一条状态消息.
				amqp_basic_qos(mp_connect, t_inf.channel(), 0, PREFETCH_COUNT, 0);
			}
			m_channel_map[t_inf.channel()] = true;
		}

		//临时队列需要代码创建, 永久队列需要在服务器上提前手动创建
		if ( t_inf.queue_durable() == 0 )
		{
			//目前只填充超时时间,  x-message-ttl 队列接受消息 的超时时间 (单位毫秒)
			if ( t_inf.queue_meassage_ttl() != 0 )
			{
				amqp_table_t t_arguments;		//队列的扩展属性 num_entries 是map长度, amqp_table_entry_t_ 是map指针
				//目前只填充超时时间,  x-message-ttl 队列接受消息 的超时时间 (单位毫秒)
				t_arguments.num_entries = 1;
				amqp_table_entry_t_ t_map_arg;
				t_map_arg.key = amqp_cstring_bytes("x-message-ttl");		//需要配置的参数
				t_map_arg.value.kind = AMQP_FIELD_KIND_U16;								//需要配置的数据类型, 如果是字符串, 写 AMQP_FIELD_KIND_UTF8
				t_map_arg.value.value.u16 = t_inf.queue_meassage_ttl();					//需要配置的数值
				t_arguments.entries = &t_map_arg;

				//amqp_queue_declare() 队列声明, 就是创建新的队列.
				//输入 amqp_connection_state_t state 连接状态参数的结构体
				//输入 amqp_channel_t channel 连接通道的编号
				//输入 amqp_bytes_t queue 队列名称,可以手动命名,如果写空,系统就会自动分配, 手动写amqp_cstring_bytes("abcdefg"), 默认空 amqp_empty_bytes
				//输入 amqp_boolean_t passive 是否被动,默认0
				//输入 amqp_boolean_t durable 是否持久,默认0, 节点代码可以创建临时队列(所有权归节点), 服务器手动创建永久队列(所有权归服务器)
				//		1表示永久队列,当节点死掉,队列在服务器保留,仍然可以接受数据,节点上线后,可以接受掉线期间的所有数据
				//		0表示临时队列,当节点死掉,队列消失,不再接受数据,直到下次恢复正常
				//输入 amqp_boolean_t exclusive 是否独立,默认0
				//输入 amqp_boolean_t auto_delete 是否自动删除,默认0, 1表示消息被消费者接受后,就自动删除消息, 当接收端断连后,队列也会才删除,
				// 													一般情况下设为0,然后让接受者手动删除.
				//输入 amqp_table_t arguments 预留参数,默认空 amqp_empty_table
				//返回 amqp_queue_declare_ok_t *	返回结果
				amqp_queue_declare(mp_connect, t_inf.channel(), amqp_cstring_bytes(t_inf.queue_name().c_str()),
								   t_inf.queue_passive(), t_inf.queue_durable(), t_inf.queue_exclusive(),
								   t_inf.queue_auto_delete(), t_arguments);
			}
			else
			{
				//amqp_queue_declare() 队列声明, 就是创建新的队列.
				//输入 amqp_connection_state_t state 连接状态参数的结构体
				//输入 amqp_channel_t channel 连接通道的编号
				//输入 amqp_bytes_t queue 队列名称,可以手动命名,如果写空,系统就会自动分配, 手动写amqp_cstring_bytes("abcdefg"), 默认空 amqp_empty_bytes
				//输入 amqp_boolean_t passive 是否被动,默认0
				//输入 amqp_boolean_t durable 是否持久,默认0, 节点代码可以创建临时队列(所有权归节点), 服务器手动创建永久队列(所有权归服务器)
				//		1表示永久队列,当节点死掉,队列在服务器保留,仍然可以接受数据,节点上线后,可以接受掉线期间的所有数据
				//		0表示临时队列,当节点死掉,队列消失,不再接受数据,直到下次恢复正常
				//输入 amqp_boolean_t exclusive 是否独立,默认0
				//输入 amqp_boolean_t auto_delete 是否自动删除,默认0, 1表示消息被消费者接受后,就自动删除消息, 当接收端断连后,队列也会才删除,
				// 													一般情况下设为0,然后让接受者手动删除.
				//输入 amqp_table_t arguments 预留参数,默认空 amqp_empty_table
				//返回 amqp_queue_declare_ok_t *	返回结果
				amqp_queue_declare(mp_connect, t_inf.channel(), amqp_cstring_bytes(t_inf.queue_name().c_str()),
								   t_inf.queue_passive(), t_inf.queue_durable(), t_inf.queue_exclusive(),
								   t_inf.queue_auto_delete(), amqp_empty_table);
			}


			//amqp_queue_bind 队列绑定, 将队列加载到服务器的交换机下面, 交换机收到消息后,就会检查key,然后放到指定的队列.
			//输入 amqp_connection_state_t state 连接状态参数的结构体
			//输入 amqp_channel_t channel 连接通道的编号
			//输入 amqp_bytes_t queue 队列名称,
			//输入 amqp_bytes_t exchange 交换机模式字符串
			//输入 amqp_bytes_t bindingkey 绑定密钥字符串, 交换机的判断规则. 发送端的 routingkey 和 接收端的 bindingkey 需要保持一致
			//输入 amqp_table_t arguments 预留参数,默认空 amqp_empty_table
			//返回 amqp_queue_bind_ok_t *	返回结果
			//注注注注注意了, 队列绑定交换机时,必须保证交换机是有效的.否则报错
			amqp_queue_bind(mp_connect, t_inf.channel(), amqp_cstring_bytes(t_inf.queue_name().c_str()),
							amqp_cstring_bytes(t_inf.exchange_name().c_str()),
							amqp_cstring_bytes(t_inf.binding_key().c_str()), amqp_empty_table);

			amqp_rpc_reply_t t_reply = amqp_get_rpc_reply(mp_connect);
			if ( t_reply.reply_type != AMQP_RESPONSE_NORMAL )
			{
				return Error_manager(Error_code::RABBITMQ_AMQP_QUEUE_BIND_ERROR, Error_level::MINOR_ERROR,
									 amqp_error_to_string(t_reply, "amqp_queue_bind") );
			}
		}

		//amqp_basic_consume 创建基本类型的消费者,就是接收端, 消费者绑定队列,只能接受一个队列里面的消息
		//输入 amqp_connection_state_t state 连接状态参数的结构体
		//输入 amqp_channel_t channel 连接通道的编号
		//输入 amqp_bytes_t queue 队列名称,
		//输入 amqp_bytes_t consumer_tag 消费者名称
		//输入 amqp_boolean_t no_local 是否非本地, 默认0,表示本地
		//输入 amqp_boolean_t no_ack, 是否确认应答,默认0,表示接收后需要应答
		//输入 amqp_boolean_t exclusive 是否独立,默认0
		//输入 amqp_table_t arguments 预留参数,默认空 amqp_empty_table
		//返回 amqp_basic_consume_ok_t *	返回结果
		//注注注注注意了, 接受端绑定队列时,必须保证队列是有效的,否则报错,
		amqp_basic_consume(mp_connect, t_inf.channel(), amqp_cstring_bytes(t_inf.queue_name().c_str()),
						   amqp_cstring_bytes(t_inf.consume_name().c_str()), t_inf.consume_no_local(),
						   t_inf.consume_no_ack(), t_inf.consume_exclusive(), amqp_empty_table);
		amqp_rpc_reply_t t_reply = amqp_get_rpc_reply(mp_connect);
		if ( t_reply.reply_type != AMQP_RESPONSE_NORMAL )
		{
			return Error_manager(Error_code::RABBITMQ_AMQP_NEW_CONSUME_ERROR, Error_level::MINOR_ERROR,
								 amqp_error_to_string(t_reply, "amqp_basic_consume") );
		}
	}

	//Rabbitmq 发送请求的通道
	for(int i=0;i<rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_request_vector_size();++i)
	{
		//Rabbitmq 配置发送通道
		Rabbitmq_proto::Rabbitmq_channel_queue_consume t_inf1 =
		rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_request_vector(i);

		//通道查重,防止重复开启(channel允许重复使用, 但是不能重复初始化)
		if ( m_channel_map.find(t_inf1.channel()) == m_channel_map.end() )
		{
			//amqp_channel_open() 打开连接通道, 同一台电脑可以多个进程和线程进行连接服务器, 每个连接需要自己独特的通道.
			amqp_channel_open(mp_connect, t_inf1.channel());
			//amqp_get_rpc_reply() 获取当前网络连接的状态结果.
			t_reply = amqp_get_rpc_reply(mp_connect);
			if ( t_reply.reply_type != AMQP_RESPONSE_NORMAL )
			{
				return Error_manager(Error_code::RABBITMQ_AMQP_CHANNEL_OPEN_ERROR, Error_level::MINOR_ERROR,
									 amqp_error_to_string(t_reply, "amqp_channel_open") );
			}
			m_channel_map[t_inf1.channel()] = true;
		}
	}

	//Rabbitmq 发送状态的通道
	for(int i=0;i<rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_status_vector_size();++i)
	{
		//Rabbitmq 配置发送通道
		Rabbitmq_proto::Rabbitmq_channel_queue_consume t_inf2 =
		rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_status_vector(i);

		//通道查重,防止重复开启(channel允许重复使用, 但是不能重复初始化)
		if ( m_channel_map.find(t_inf2.channel()) == m_channel_map.end() )
		{
			//amqp_channel_open() 打开连接通道, 同一台电脑可以多个进程和线程进行连接服务器, 每个连接需要自己独特的通道.
			amqp_channel_open(mp_connect, t_inf2.channel());
			//amqp_get_rpc_reply() 获取当前网络连接的状态结果.
			t_reply = amqp_get_rpc_reply(mp_connect);
			if ( t_reply.reply_type != AMQP_RESPONSE_NORMAL )
			{
				return Error_manager(Error_code::RABBITMQ_AMQP_CHANNEL_OPEN_ERROR, Error_level::MINOR_ERROR,
									 amqp_error_to_string(t_reply, "amqp_channel_open") );
			}
			m_channel_map[t_inf2.channel()] = true;
		}
	}

	return Error_code::SUCCESS;
}

//启动通信, 开启线程, run thread
Error_manager Rabbitmq_base::rabbitmq_run()
{
	//启动 线程。
	//接受线程默认循环, 内部的 amqp_consume_message 进行等待, 超时1ms
	m_receive_analysis_condition.reset(false, true, false);
	mp_receive_analysis_thread = new std::thread(&Rabbitmq_base::receive_analysis_thread, this);
	//发送线程默认循环, 内部的wait_and_pop进行等待,
	m_send_condition.reset(false, true, false);
	mp_send_thread = new std::thread(&Rabbitmq_base::send_thread, this);
	//封装线程默认等待, ...., 超时1秒, 超时后主动 封装心跳和状态信息,
	m_encapsulate_status_condition.reset(false, false, false);
	mp_encapsulate_status_thread = new std::thread(&Rabbitmq_base::encapsulate_status_thread, this);

	m_rabbitmq_status = RABBITMQ_STATUS_READY;
	return Error_code::SUCCESS;
}

//反初始化 通信 模块。
Error_manager Rabbitmq_base::rabbitmq_uninit()
{
	LOG(INFO) << " ---Rabbitmq_base::rabbitmq_uninit() run--- "<< this;

	//终止list，防止 wait_and_pop 阻塞线程。
	m_send_list.termination_list();

	//杀死线程，强制退出
	if (mp_receive_analysis_thread)
	{
		m_receive_analysis_condition.kill_all();
	}
	if (mp_send_thread)
	{
		m_send_condition.kill_all();
	}
	if (mp_encapsulate_status_thread)
	{
		m_encapsulate_status_condition.kill_all();
	}


	//回收线程的资源
	if (mp_receive_analysis_thread)
	{
		mp_receive_analysis_thread->join();
		delete mp_receive_analysis_thread;
		mp_receive_analysis_thread = NULL;
	}
	if (mp_send_thread)
	{
		mp_send_thread->join();
		delete mp_send_thread;
		mp_send_thread = NULL;
	}
	if (mp_encapsulate_status_thread)
	{
		mp_encapsulate_status_thread->join();
		delete mp_encapsulate_status_thread;
		mp_encapsulate_status_thread = NULL;
	}

	//清空list
	m_send_list.clear_and_delete();

	if ( m_rabbitmq_status == RABBITMQ_STATUS_READY )
	{
		for (auto iter = m_channel_map.begin(); iter != m_channel_map.end(); ++iter)
		{
			amqp_channel_close(mp_connect, iter->first, AMQP_REPLY_SUCCESS);
		}
		amqp_connection_close(mp_connect, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(mp_connect);
	}

	m_rabbitmq_status = RABBITMQ_STATUS_UNKNOW;
	return Error_code::SUCCESS;
}

//重连, 快速uninit, init
Error_manager Rabbitmq_base::rabbitmq_reconnnect()
{
	//重连全程加锁,防止其他线程运行.
	std::unique_lock<std::mutex> lk(m_mutex);
	m_rabbitmq_status = RABBITMQ_STATUS_RECONNNECT;

	//断开连接
	for (auto iter = m_channel_map.begin(); iter != m_channel_map.end(); ++iter)
	{
		amqp_channel_close(mp_connect, iter->first, AMQP_REPLY_SUCCESS);
	}
	amqp_connection_close(mp_connect, AMQP_REPLY_SUCCESS);
	amqp_destroy_connection(mp_connect);

	//重新连接,线程不需要重启
	LOG(INFO) << " ---Rabbitmq_base::rabbitmq_reconnnect() run--- "<< this;

	int t_status=0;		//状态
	amqp_rpc_reply_t t_reply;	//reply答复结果
	Error_manager t_error;

	//amqp_new_connection 新建amqp的连接配置,里面只有连接状态参数
	// 返回amqp_connection_state_t_ *, 函数内部分配内存, amqp_destroy_connection()可以释放内存, 内存不为空则成功
	mp_connect = amqp_new_connection();
	if ( mp_connect == NULL )
	{
		return Error_manager(Error_code::RABBITMQ_AMQP_NEW_CONNECTION_ERROR, Error_level::MINOR_ERROR,
							 "amqp_new_connection fun error ");
	}

	//amqp_tcp_socket_new 新建tcp_socket连接
	// 返回amqp_socket_t *, 函数内部分配内存, amqp_connection_close()可以释放内存, 内存不为空则成功
	mp_socket = amqp_tcp_socket_new(mp_connect);
	if ( mp_socket == NULL )
	{
		return Error_manager(Error_code::RABBITMQ_AMQP_TCP_SOCKET_NEW_ERROR, Error_level::MINOR_ERROR,
							 "amqp_tcp_socket_new fun error ");
	}

	//载入外部参数
	if (m_rabbitmq_parameter_all.rabbitmq_parameters().has_ip() &&
		m_rabbitmq_parameter_all.rabbitmq_parameters().has_port() &&
		m_rabbitmq_parameter_all.rabbitmq_parameters().has_user() &&
		m_rabbitmq_parameter_all.rabbitmq_parameters().has_password() )
	{
		m_ip = m_rabbitmq_parameter_all.rabbitmq_parameters().ip();
		m_port = m_rabbitmq_parameter_all.rabbitmq_parameters().port();
		m_user = m_rabbitmq_parameter_all.rabbitmq_parameters().user();
		m_password = m_rabbitmq_parameter_all.rabbitmq_parameters().password();
	}
	else
	{
		return Error_manager(Error_code::RABBITMQ_PROTOBUF_LOSS_ERROR, Error_level::MINOR_ERROR,
							 " rabbitmq_parameter_all.rabbitmq_parameters() The data is not complete   ");
	}

	//amqp_socket_open 打开socket连接, 输入ip和port,
	// 成功返回AMQP_STATUS_OK = 0x0, 失败返回错误状态码, 详见 enum amqp_status_enum_
	//只需要设置配置服务器的ip和port, 不需要配置子节点客户端的ip和port, 在后面配置channel通道时,进行设置.
	t_status = amqp_socket_open(mp_socket, m_ip.c_str(), m_port);
	if ( t_status != AMQP_STATUS_OK )
	{
		return Error_manager(Error_code::RABBITMQ_AMQP_SOCKET_OPEN_ERROR, Error_level::MINOR_ERROR,
							 amqp_error_to_string(t_status, "amqp_socket_open") );
	}

	//amqp_login() 登录代理服务器,
	//输入 连接参数结构体 amqp_connection_state_t,
	//输入 连接地址, 前面 amqp_socket_open() 已经输入了,这里默认写"/"
	//输入 连接通道最大值, 默认值0表示没有限制
	//输入 连接帧率最大值, 默认值是131072 (128KB)
	//输入 心跳帧之间的秒数, 默认值0禁用心跳
	//输入 身份验证模式, 	AMQP_SASL_METHOD_PLAIN, 追加用户名和密码
	//					AMQP_SASL_METHOD_EXTERNAL, 追加身份证
	//返回 结果的结构体 amqp_rpc_reply_t
	//	amqp_response_type_enum reply_type 登录成功是 AMQP_RESPONSE_NORMAL
	//		失败:如果是 reply_type == AMQP_RESPONSE_SERVER_EXCEPTION, 服务器连接错误, 错误信息在 amqp_method_t reply
	//		失败:如果是 reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION, 库函数错误, 错误信息在 int library_error
	t_reply = amqp_login(mp_connect, "/", 0, 131072, 0,
						 AMQP_SASL_METHOD_PLAIN, m_user.c_str(), m_password.c_str());
	if ( t_reply.reply_type != AMQP_RESPONSE_NORMAL )
	{
		return Error_manager(Error_code::RABBITMQ_AMQP_LOGIN_ERROR, Error_level::MINOR_ERROR,
							 amqp_error_to_string(t_reply, "amqp_login") );
	}

	//清除channel_map, 通道的缓存,防止重复开启, (channel允许重复使用, 但是不能重复初始化)
	m_channel_map.clear();

	//创建通道队列消费者, (交换机和永久队列不在代码里创建,请在服务器上手动创建)
	t_error = rabbitmq_new_channel_queue_consume(m_rabbitmq_parameter_all);
	if ( t_error != Error_code::SUCCESS )
	{
		return t_error;
	}

	//不用重启线程
	return Error_code::SUCCESS;

}

//设置 自动封装状态的时间周期
void Rabbitmq_base::set_encapsulate_status_cycle_time(unsigned int encapsulate_status_cycle_time)
{
	m_encapsulate_status_cycle_time = encapsulate_status_cycle_time;
}

//设置回调函数check_msg_callback
void Rabbitmq_base::set_check_msg_callback(Error_manager (*callback)(Rabbitmq_message* p_msg))
{
	check_msg_callback = callback;
}
//设置回调函数check_executer_callback
void Rabbitmq_base::set_check_executer_callback(Error_manager (*callback)(Rabbitmq_message* p_msg))
{
	check_executer_callback = callback;
}
//设置回调函数execute_msg_callback
void Rabbitmq_base::set_execute_msg_callback(Error_manager (*callback)(Rabbitmq_message* p_msg))
{
	execute_msg_callback = callback;
}
//设置回调函数encapsulate_status_callback
void Rabbitmq_base::set_encapsulate_status_callback(Error_manager (*callback)())
{
	encapsulate_status_callback = callback;
}

//mp_receive_analysis_thread 接受解析 执行函数，
void Rabbitmq_base::receive_analysis_thread()
{
	LOG(INFO) << " Rabbitmq_base::receive_analysis_thread start "<< this;

	//通信接受线程, 负责接受socket消息, 并存入 m_receive_data_list
	while (m_receive_analysis_condition.is_alive())
	{
		//这里就不需要超时等待了, rabbitmq的接受函数可以配置等待超时....
//		m_receive_analysis_condition.wait_for_ex(std::chrono::microseconds(1));
		m_receive_analysis_condition.wait();
		if ( m_receive_analysis_condition.is_alive() )
		{
            std::this_thread::sleep_for(std::chrono::microseconds(100));
            std::this_thread::yield();

			amqp_rpc_reply_t t_reply;		//运行结果
			amqp_envelope_t t_envelope;		//数据包, 含有一些包裹属性和数据内容
			//接受消息等待超时,默认1000us, 当收到消息后,立刻通过阻塞,否则等待超时后通过阻塞
			struct timeval t_timeout;		//超时时间, 默认1ms
			t_timeout.tv_sec = 0;
			t_timeout.tv_usec = 1000;

			{//这个大括号表示只对 recv 和 send 加锁, 不要因为后面的复杂逻辑影响通信效率
				std::unique_lock<std::mutex> lk(m_mutex);
				//允许释放连接参数状态的内存,
				// 因为这个连接是底层分配的内存,是全局的. 为了开启多个连接,就要重复使用
				//这里释放之后,其他代码就开启多线程开启新的连接了.
				amqp_maybe_release_buffers(mp_connect);
				//amqp_consume_message 接受消息, 阻塞函数,可以设置超时.
				//输入 amqp_connection_state_t state 连接状态参数的结构体
				//输入 amqp_envelope_t *envelope 接受数据包的指针, 成功接收到数据后,数据包会覆盖
				//输入 const struct timeval *timeout 超时时间, 防止阻塞. 传入NULL就是完全阻塞.
				//输入 int flags 未使用, 默认0
				//输入 amqp_connection_state_t state 连接状态参数的结构体
				//返回 状态结果的结构体 amqp_rpc_reply_t
				//	amqp_response_type_enum reply_type 成功是 AMQP_RESPONSE_NORMAL
				//		失败:如果是 reply_type == AMQP_RESPONSE_SERVER_EXCEPTION, 服务器连接错误, 错误信息在 amqp_method_t reply
				//		失败:如果是 reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION, 库函数错误, 错误信息在 int library_error
				t_reply = amqp_consume_message(mp_connect, &t_envelope, &t_timeout, 0);
			}

			if ( AMQP_RESPONSE_NORMAL == t_reply.reply_type )//正常接受到消息
			{


				m_rabbitmq_status = RABBITMQ_STATUS_READY;
				//从t_envelope数据包里面提取信息
				std::string t_receive_string = std::string((char*)t_envelope.message.body.bytes, t_envelope.message.body.len);
				int t_channel = t_envelope.channel;
				int t_delivery_tag = t_envelope.delivery_tag;
				std::string t_exchange_name = std::string((char*)t_envelope.exchange.bytes, t_envelope.exchange.len);
				std::string t_routing_key = std::string((char*)t_envelope.routing_key.bytes, t_envelope.routing_key.len);
				//如果这里接受到了消息, 在这提前解析消息最前面的Base_msg (消息公共内容), 用于后续的check
				message::Base_msg t_base_msg;
//				if( t_base_msg.ParseFromString(t_receive_string) )

				//删除 message::Base_msg 里面的 message::Base_info的机制,完全依赖服务器来分发消息
				if( true )
				{
					//第一次解析之后转化为, Communication_message, 自定义的通信消息格式
					Rabbitmq_message  t_rabbitmq_message;
					t_rabbitmq_message.reset(t_base_msg.base_info(), t_receive_string, t_channel, t_delivery_tag, t_exchange_name, t_routing_key);
					//检查消息是否有效, 主要检查消息类型和接受者, 判断这条消息是不是给我的.
					if ( check_msg(&t_rabbitmq_message) == SUCCESS )
					{
						//这里直接就用当前线程进行处理,
						//检查消息是否可以被处理
						if ( check_executer(&t_rabbitmq_message) == SUCCESS )
						{
							//处理消息
							if ( execute_msg(&t_rabbitmq_message) == SUCCESS )
							{

							}
							//else不做处理
						}
						//else不做处理
					}
					//else不做处理
				}
					//else解析失败, 就当做什么也没发生, 认为接收消息无效,
				else
				{
					std::cout << " huli test :::: " << " t_receive_string = " << t_receive_string << std::endl;
					if ( t_channel == 401 )
					{
						amqp_basic_ack(mp_connect, t_channel, t_delivery_tag, 0);
					}
				}

				//amqp_destroy_envelope 销毁数据包, 只有接受成功, t_envelope才有内存
				amqp_destroy_envelope(&t_envelope);
			}
			else//没有接受到消息
			{
				//超时报错,不做处理, continue
				//注注注注注意了, 没有收到消息会超时报错, res.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION, res.library_error = -13, (-0x000D request timed out)
				if (t_reply.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION && t_reply.library_error == -13)
				{
					m_rabbitmq_status = RABBITMQ_STATUS_READY;
					continue;
				}
				else//其他报错,特殊处理
				{
					//need

					std::string error_description = amqp_error_to_string(t_reply, "amqp_consume_message");
					std::cout << " huli test 123123123:::: " << " error_description = " << error_description << std::endl;
//					return Error_manager(Error_code::RABBITMQ_AMQP_CONSUME_MESSAGE_ERROR, Error_level::MINOR_ERROR,
//											 amqp_error_to_string(t_reply, "amqp_consume_message") );
					//重启
					rabbitmq_reconnnect();

				}
			}
		}
	}

	LOG(INFO) << " Rabbitmq_base::receive_analysis_thread end "<< this;
	return;
}

//检查消息是否有效, 主要检查消息类型和接受者, 判断这条消息是不是给我的., 需要子类重载
Error_manager Rabbitmq_base::check_msg(Rabbitmq_message* p_msg)
{
	if ( check_msg_callback != NULL )
	{
		return check_msg_callback(p_msg);
	}
	return Error_code::SUCCESS;
}

//检查执行者的状态, 判断能否处理这条消息, 需要子类重载
Error_manager Rabbitmq_base::check_executer(Rabbitmq_message* p_msg)
{
	if ( check_executer_callback != NULL )
	{
		return check_executer_callback(p_msg);
	}
	return Error_code::SUCCESS;
}

//处理消息, 需要子类重载
Error_manager Rabbitmq_base::execute_msg(Rabbitmq_message* p_msg)
{
	if ( execute_msg_callback != NULL )
	{
		return execute_msg_callback(p_msg);
	}
	else
	{
		//需要子类重载
		std::cout << " huli test :::: " << " execute_msg Rabbitmq_message = " << p_msg->get_message_buf() << std::endl;

		//如果是请求消息,那么在子节点继承的时候一定要记得调用
		//配置rabbitmq.proto时,  如果consume_no_ack == 0 , 一定要手动调用 amqp_basic_ack
		int consume_no_ack = 1;
		if(consume_no_ack == 0 || p_msg->m_channel == 401)
		{
			//amqp_basic_ack 确认消息, 通知服务器队列手动删除消息.
			//输入 amqp_connection_state_t state 连接状态参数的结构体
			//输入 amqp_channel_t channel 连接通道的编号
			//输入 uint64_t delivery_tag  消息传递编号,
			//输入 amqp_boolean_t multiple 多个标记位, 默认0,   1表示删除1~delivery_tag的所有消息, 不删除大于delivery_tag的, 0表示只删除这一条
			int ack_result = amqp_basic_ack(mp_connect, p_msg->m_channel, p_msg->m_delivery_tag, 0);
		}
	}
	return Error_code::SUCCESS;
}

//ack_msg 处理完消息后, 手动确认消息, 通知服务器队列删除消息.
//执行者在execute_msg里面可以调用这个函数, 或者回调也行.
Error_manager Rabbitmq_base::ack_msg(Rabbitmq_message* p_msg)
{
	//amqp_basic_ack 确认消息, 通知服务器队列手动删除消息.
	//输入 amqp_connection_state_t state 连接状态参数的结构体
	//输入 amqp_channel_t channel 连接通道的编号
	//输入 uint64_t delivery_tag  消息传递编号,
	//输入 amqp_boolean_t multiple 多个标记位, 默认0,   1表示删除1~delivery_tag的所有消息, 不删除大于delivery_tag的, 0表示只删除这一条
	int ack_result = amqp_basic_ack(mp_connect, p_msg->m_channel, p_msg->m_delivery_tag, 0);
	if ( ack_result != 0 )
	{
		return Error_manager(Error_code::RABBITMQ_AMQP_BASIC_ACK_ERROR, Error_level::MINOR_ERROR,
							 amqp_error_to_string(ack_result, "amqp_basic_ack") );
	}
	return Error_code::SUCCESS;
}

//mp_send_thread 发送线程执行函数，
void Rabbitmq_base::send_thread()
{
	LOG(INFO) << " Rabbitmq_base::send_thread start "<< this;

	//通信发送线程, 负责巡检m_send_list, 并发送消息
	while (m_send_condition.is_alive())
	{
		m_send_condition.wait();
		if ( m_send_condition.is_alive() )
		{
			std::this_thread::yield();

			Rabbitmq_message* tp_msg = NULL;
			int t_result = 0;
			//这里 wait_and_pop 会使用链表内部的 m_data_cond 条件变量来控制等待,
			//封装线程使用push的时候, 会唤醒线程并通过等待, 此时 m_send_data_condition 是一直通过的.
			//如果需要退出, 那么就要 m_send_data_list.termination_list();	和 m_send_data_condition.kill_all();
			bool is_pop = m_send_list.wait_and_pop(tp_msg);
			if ( is_pop )
			{
				if ( tp_msg != NULL )
				{
					//amqp_basic_properties_t 消息数据的基本属性,里面有15个成员.
					amqp_basic_properties_t props;

					//判断是否要设置发送消息的超时时间, 如果配置10秒,超时后,服务器会自动删除消息
					if ( tp_msg->m_timeout_ms == std::chrono::milliseconds(0) )
					{
						//amqp_flags_t _flags	一个uint32_t, 按位 表示这15个属性的修改开关.
						//例如:     _flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG = 0b 1001 0000 0000 0000;
						//就表示   content-type  和  delivery-mode  是有效属性. 接下来的设置就会生效.
						props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
						//amqp_bytes_t content_type 消息数据的类型 "text/plain"是 普通文本格式
						//注意了,需要使用 amqp_cstring_bytes() 将char*转为amqp_bytes_t(自定义的字符串, 类似于std::string)
						props.content_type = amqp_cstring_bytes("text/plain");
						//uint8_t delivery_mode 配送模式  2表示持续发送模式
						props.delivery_mode = AMQP_DELIVERY_PERSISTENT;
					}
					else
					{
						//amqp_flags_t _flags	一个uint32_t, 按位 表示这15个属性的修改开关.
						//例如:     _flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG = 0b 1001 0000 0000 0000;
						//就表示   content-type  和  delivery-mode  是有效属性. 接下来的设置就会生效.
						props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG | AMQP_BASIC_EXPIRATION_FLAG;
						//amqp_bytes_t content_type 消息数据的类型 "text/plain"是 普通文本格式
						//注意了,需要使用 amqp_cstring_bytes() 将char*转为amqp_bytes_t(自定义的字符串, 类似于std::string)
						props.content_type = amqp_cstring_bytes("text/plain");
						//uint8_t delivery_mode 配送模式  2表示持续发送模式
						props.delivery_mode = AMQP_DELIVERY_PERSISTENT;
						char buf[256] = {0};
						sprintf(buf, "%d", (int)tp_msg->m_timeout_ms.count());
						props.expiration = amqp_cstring_bytes(buf);//超时, 单位ms;
					}

					{//这个大括号表示只对 recv 和 send 加锁, 不要因为后面的复杂逻辑影响通信效率
						std::unique_lock<std::mutex> lk(m_mutex);
//						std::cout << " huli test :::: " << " tp_msg->m_message_buf = " << tp_msg->m_message_buf << std::endl;

						//amqp_basic_publish() 发布消息给代理服务器, 在交换器上发布一个带有路由密钥的消息。交换机会根据路由密钥匹配,放到对应的队列里面
						//输入 amqp_connection_state_t state 连接状态参数的结构体
						//输入 amqp_channel_t channel 连接通道的编号
						//输入 amqp_bytes_t exchange 交换机模式字符串
						//输入 amqp_bytes_t routing_key 路由密钥字符串, 交换机的判断规则. 发送端的 routingkey 和 接收端的 bindingkey 需要保持一致
						//输入 amqp_boolean_t mandatory 强制服务器必须通过路由密钥才能存到队列, 默认为0
						//输入 amqp_boolean_t immediate 表示服务器必须立刻转发消息给接受者, 默认为0
						//输入 struct amqp_basic_properties_t_ const *properties 消息数据的基本属性
						//输入 amqp_bytes_t body 消息数据内容
						//返回错误码 成功返回AMQP_STATUS_OK = 0x0, 失败返回错误状态码, 详见 enum amqp_status_enum_
						//注注注注注意了::amqp_basic_publish()是异步通信,
						// return AMQP_STATUS_OK 也只是表示消息成功发送到服务器. 无法确认 接收端是否正常接受消息
						t_result = amqp_basic_publish(mp_connect, tp_msg->m_channel,
													  amqp_cstring_bytes(tp_msg->m_exchange_name.c_str()),
													  amqp_cstring_bytes(tp_msg->m_routing_key.c_str()), 0, 0,
													  &props, amqp_cstring_bytes(tp_msg->m_message_buf.c_str()) );
					}


					if ( t_result == AMQP_STATUS_OK )
					{
						m_rabbitmq_status = RABBITMQ_STATUS_READY;
						delete(tp_msg);
						tp_msg = NULL;
//							std::string re = amqp_error_to_string(t_result, "amqp_basic_publish");
//							std::cout << " huli test :::: " << " re = " << re << std::endl;
//							return Error_manager(Error_code::RABBITMQ_AMQP_BASIC_PUBLISH_ERROR, Error_level::MINOR_ERROR,
//												 amqp_error_to_string(t_result, "amqp_basic_publish") );
					}
					else
					{
						std::string re = amqp_error_to_string(t_result, "amqp_basic_publish");
						std::cout << " huli test :::: " << " re = " << re << std::endl;

						//重启
						m_rabbitmq_status = RABBITMQ_STATUS_RECONNNECT;
						m_send_list.push(tp_msg);	//重新加入队列,下一次再发
						tp_msg = NULL;
						rabbitmq_reconnnect();
					}


				}
			}
			else
			{
				//没有取出, 那么应该就是 m_termination_flag 结束了
//				return Error_manager(Error_code::CONTAINER_IS_TERMINATE, Error_level::MINOR_ERROR,
//									 " Communication_socket_base::send_data_thread() error ");
			}
		}
	}

	LOG(INFO) << " Rabbitmq_base::send_thread end "<< this;
	return;
}
//手动封装消息,需要手动写入参数channel,exchange_name,routing_key
Error_manager Rabbitmq_base::encapsulate_msg(std::string message, int channel, std::string exchange_name, std::string routing_key, int timeout_ms=0)
{
	if ( m_rabbitmq_status != RABBITMQ_STATUS_READY )
	{
	    return Error_manager(Error_code::ERROR, Error_level::MINOR_ERROR,
	    					" m_rabbitmq_status error ");
	}
	Rabbitmq_message* tp_msg = new Rabbitmq_message(message, channel, exchange_name, routing_key, timeout_ms);
	bool is_push = m_send_list.push(tp_msg);
	if ( is_push == false )
	{
		delete(tp_msg);
		tp_msg = NULL;
		return Error_manager(Error_code::CONTAINER_IS_TERMINATE, Error_level::MINOR_ERROR,
							 " Communication_socket_base::encapsulate_msg error ");
	}
	return Error_code::SUCCESS;
}
//手动封装消息,需要手动写入参数channel,exchange_name,routing_key
Error_manager Rabbitmq_base::encapsulate_msg(Rabbitmq_message* p_msg)
{
	Rabbitmq_message* tp_msg = new Rabbitmq_message(*p_msg);
	bool is_push = m_send_list.push(tp_msg);
	if ( is_push == false )
	{
		delete(tp_msg);
		tp_msg = NULL;
		return Error_manager(Error_code::CONTAINER_IS_TERMINATE, Error_level::MINOR_ERROR,
							 " Communication_socket_base::encapsulate_msg error ");
	}
	return Error_code::SUCCESS;
}
//手动封装任务消息(请求和答复), 系统会使用rabbitmq.proto的配置参数,
Error_manager Rabbitmq_base::encapsulate_task_msg(std::string message, int vector_index)
{
	int channel = m_rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_request_vector(vector_index).channel();
	std::string exchange_name = m_rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_request_vector(vector_index).exchange_name();
	std::string routing_key = m_rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_request_vector(vector_index).routing_key();
	int timeout_ms = m_rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_request_vector(vector_index).timeout_ms();
	Rabbitmq_message* tp_msg = new Rabbitmq_message(message, channel, exchange_name, routing_key, timeout_ms);
	bool is_push = m_send_list.push(tp_msg);
	if ( is_push == false )
	{
		delete(tp_msg);
		tp_msg = NULL;
		return Error_manager(Error_code::CONTAINER_IS_TERMINATE, Error_level::MINOR_ERROR,
							 " Communication_socket_base::encapsulate_msg error ");
	}
	return Error_code::SUCCESS;
}
//手动封装状态消息, 系统会使用rabbitmq.proto的配置参数,
Error_manager Rabbitmq_base::encapsulate_status_msg(std::string message, int vector_index)
{
    for(auto &iter : m_rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_status_vector()) {
        if (iter.routing_key().find(std::to_string(vector_index)) != iter.routing_key().npos) {
            int channel = iter.channel();
            std::string exchange_name = iter.exchange_name();
            std::string routing_key = iter.routing_key();
            int timeout_ms = iter.timeout_ms();
            Rabbitmq_message* tp_msg = new Rabbitmq_message(message, channel, exchange_name, routing_key, timeout_ms);
            bool is_push = m_send_list.push(tp_msg);
            if (!is_push)
            {
                delete(tp_msg);
                tp_msg = nullptr;
                return Error_manager(Error_code::CONTAINER_IS_TERMINATE, Error_level::MINOR_ERROR,
                                     " Communication_socket_base::encapsulate_msg error ");
            }
            return Error_code::SUCCESS;
        }
    }

    return Error_manager(Error_code::CONTAINER_IS_TERMINATE, Error_level::MINOR_ERROR,
                         " Communication_socket_base::encapsulate_msg error ");
}

//mp_encapsulate_stauts_thread 自动封装线程执行函数，
void Rabbitmq_base::encapsulate_status_thread()
{
	LOG(INFO) << " Rabbitmq_base::encapsulate_status_thread start "<< this;

	//通信封装线程, 负责定时封装消息, 并存入 m_send_data_list
	while (m_encapsulate_status_condition.is_alive())
	{
		bool t_pass_flag = m_encapsulate_status_condition.wait_for_millisecond(m_encapsulate_status_cycle_time);

		if ( m_encapsulate_status_condition.is_alive() )
		{
			std::this_thread::yield();
			//如果封装线程被主动唤醒, 那么就表示 需要主动发送消息,
			if ( t_pass_flag )
			{
				//主动发送消息,
			}
				//如果封装线程超时通过, 那么就定时封装心跳和状态信息
			else
			{
				//只有通信正常的时候,才封装发送状态消息
				if ( m_rabbitmq_status == RABBITMQ_STATUS_READY )
				{
					auto_encapsulate_status();
				}
			}
		}
	}

	LOG(INFO) << " Rabbitmq_base::encapsulate_status_thread end "<< this;
	return;
}
//定时封装发送消息, 一般为心跳和状态信息, 需要子类重载
Error_manager Rabbitmq_base::auto_encapsulate_status()
{
	if ( encapsulate_status_callback != NULL )
	{
		return encapsulate_status_callback();
	}
	return Error_code::SUCCESS;
}



//把rabbitmq的错误信息转化为string, amqp_status就是enum amqp_status_enum_, amqp_error_string2()函数可以把他翻译为string
std::string Rabbitmq_base::amqp_error_to_string(int amqp_status)
{
	char buf[256] = {0};
	sprintf(buf, "amqp_status = 0x%x, %s", amqp_status, amqp_error_string2(amqp_status));
	return buf;
}
//把rabbitmq的错误信息转化为string, amqp_status就是enum amqp_status_enum_, amqp_error_string2()函数可以把他翻译为string
std::string Rabbitmq_base::amqp_error_to_string(int amqp_status, std::string amqp_fun_name)
{
	char buf[256] = {0};
	sprintf(buf, "amqp_fun_name = %s, amqp_status = 0x%x, %s", amqp_fun_name.c_str(), amqp_status, amqp_error_string2(amqp_status));
	return buf;
}
//把rabbitmq的错误信息转化为string, amqp_rpc_reply_t就是amqp函数运行的结果
std::string Rabbitmq_base::amqp_error_to_string(amqp_rpc_reply_t amqp_rpc_reply)
{
	char buf[256] = {0};
	//	amqp_response_type_enum reply_type 登录成功是 AMQP_RESPONSE_NORMAL
	//		失败:如果是 reply_type == AMQP_RESPONSE_SERVER_EXCEPTION, 服务器连接错误, 错误信息在 amqp_method_t reply
	//		失败:如果是 reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION, 库函数错误, 错误信息在 int library_error
	switch ( amqp_rpc_reply.reply_type )
	{
		case AMQP_RESPONSE_NORMAL:
		{
			sprintf(buf, "SUCCESS");
			break;
		}
		case AMQP_RESPONSE_NONE:
		{
			sprintf(buf, " reply_type = AMQP_RESPONSE_NONE " );
			break;
		}
		case AMQP_RESPONSE_LIBRARY_EXCEPTION:
		{
			sprintf(buf, " reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION, library_error = %s, ",amqp_error_string2(amqp_rpc_reply.library_error) );
			break;
		}
		case AMQP_RESPONSE_SERVER_EXCEPTION:
		{
			if ( amqp_rpc_reply.reply.id == AMQP_CONNECTION_CLOSE_METHOD )
			{
				amqp_connection_close_t * p_decoded = (amqp_connection_close_t *)amqp_rpc_reply.reply.decoded;
				sprintf(buf, " reply.id = AMQP_CONNECTION_CLOSE_METHOD, reply = %u, %.*s ",
						p_decoded->reply_code, (int)p_decoded->reply_text.len, (char *)p_decoded->reply_text.bytes);
			}
			else if ( amqp_rpc_reply.reply.id == AMQP_CHANNEL_CLOSE_METHOD )
			{
				amqp_channel_close_t * p_decoded = (amqp_channel_close_t *)amqp_rpc_reply.reply.decoded;
				sprintf(buf, " reply.id = AMQP_CHANNEL_CLOSE_METHOD, reply = %u, %.*s ",
						p_decoded->reply_code, (int)p_decoded->reply_text.len, (char *)p_decoded->reply_text.bytes);
			}
			else
			{
				sprintf(buf, " reply_type = AMQP_RESPONSE_SERVER_EXCEPTION " );
			}
			break;
		}
		default:
		{
			sprintf(buf, " reply_type = unknown, reply.id = 0x%08X, ",
					amqp_rpc_reply.reply.id );
			break;
		}
	}

	return buf;
}
//把rabbitmq的错误信息转化为string, amqp_rpc_reply_t就是amqp函数运行的结果
std::string Rabbitmq_base::amqp_error_to_string(amqp_rpc_reply_t amqp_rpc_reply, std::string amqp_fun_name)
{
	char buf[256] = {0};
	//	amqp_response_type_enum reply_type 登录成功是 AMQP_RESPONSE_NORMAL
	//		失败:如果是 reply_type == AMQP_RESPONSE_SERVER_EXCEPTION, 服务器连接错误, 错误信息在 amqp_method_t reply
	//		失败:如果是 reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION, 库函数错误, 错误信息在 int library_error
	switch ( amqp_rpc_reply.reply_type )
	{
		case AMQP_RESPONSE_NORMAL:
		{
			sprintf(buf, "SUCCESS");
			break;
		}
		case AMQP_RESPONSE_NONE:
		{
			sprintf(buf, "amqp_fun_name = %s, reply_type = AMQP_RESPONSE_NONE ", amqp_fun_name.c_str() );
			break;
		}
		case AMQP_RESPONSE_LIBRARY_EXCEPTION:
		{
			// huli test 123123123::::  error_description = amqp_fun_name = amqp_consume_message, reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION, library_error = unexpected protocol state,
			sprintf(buf, "amqp_fun_name = %s, reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION, library_error = %s, ", amqp_fun_name.c_str(),amqp_error_string2(amqp_rpc_reply.library_error) );
			break;
		}
		case AMQP_RESPONSE_SERVER_EXCEPTION:
		{
			if ( amqp_rpc_reply.reply.id == AMQP_CONNECTION_CLOSE_METHOD )
			{
				amqp_connection_close_t * p_decoded = (amqp_connection_close_t *)amqp_rpc_reply.reply.decoded;
				sprintf(buf, "amqp_fun_name = %s, reply.id = AMQP_CONNECTION_CLOSE_METHOD, reply = %u, %.*s ",
						amqp_fun_name.c_str(), p_decoded->reply_code, (int)p_decoded->reply_text.len, (char *)p_decoded->reply_text.bytes);
			}
			else if ( amqp_rpc_reply.reply.id == AMQP_CHANNEL_CLOSE_METHOD )
			{
				amqp_channel_close_t * p_decoded = (amqp_channel_close_t *)amqp_rpc_reply.reply.decoded;
				sprintf(buf, "amqp_fun_name = %s, reply.id = AMQP_CHANNEL_CLOSE_METHOD, reply = %u, %.*s ",
						amqp_fun_name.c_str(), p_decoded->reply_code, (int)p_decoded->reply_text.len, (char *)p_decoded->reply_text.bytes);
			}
			else
			{
				sprintf(buf, "amqp_fun_name = %s, reply_type = AMQP_RESPONSE_SERVER_EXCEPTION ", amqp_fun_name.c_str() );
			}
			break;
		}
		default:
		{
			sprintf(buf, "amqp_fun_name = %s, reply_type = unknown, reply.id = 0x%08X, ",
					amqp_fun_name.c_str(),amqp_rpc_reply.reply.id );
			break;
		}
	}

	return buf;
}