|
|
@@ -0,0 +1,981 @@
|
|
|
+
|
|
|
+#include "rabbitmq_base.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() {
|
|
|
+ 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 (loadProtobufFile(prototxt_path, t_rabbitmq_parameter_all) != SUCCESS) {
|
|
|
+ return Error_manager(RABBITMQ_READ_PROTOBUF_ERROR, MINOR_ERROR,
|
|
|
+ "rabbitmq_init_from_protobuf read_proto_param failed");
|
|
|
+ }
|
|
|
+ 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;
|
|
|
+
|
|
|
+ for (auto &queue:rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_reciever_vector()) {
|
|
|
+ mp_rabbitmq_reciever.insert(std::pair<const std::string, Rabbitmq_proto::Rabbitmq_channel_queue_consume>(queue.routing_key(), queue));
|
|
|
+ }
|
|
|
+ for (auto &queue:rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_request_vector()) {
|
|
|
+ mp_rabbitmq_reciever.insert(std::pair<const std::string, Rabbitmq_proto::Rabbitmq_channel_queue_consume>(queue.routing_key(), queue));
|
|
|
+ }
|
|
|
+ for (auto &queue:rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_status_vector()) {
|
|
|
+ mp_rabbitmq_reciever.insert(std::pair<const std::string, Rabbitmq_proto::Rabbitmq_channel_queue_consume>(queue.routing_key(), queue));
|
|
|
+ }
|
|
|
+
|
|
|
+ //amqp_new_connection 新建amqp的连接配置,里面只有连接状态参数
|
|
|
+ // 返回amqp_connection_state_t_ *, 函数内部分配内存, amqp_destroy_connection()可以释放内存, 内存不为空则成功
|
|
|
+ mp_connect = amqp_new_connection();
|
|
|
+ if (mp_connect == nullptr) {
|
|
|
+ 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 == nullptr) {
|
|
|
+ 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");
|
|
|
+ LOG(WARNING) << " 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) {
|
|
|
+ LOG(ERROR) << " m_rabbitmq_status error ";
|
|
|
+ return Error_manager(Error_code::ERROR, Error_level::MINOR_ERROR,
|
|
|
+ " m_rabbitmq_status error ");
|
|
|
+ }
|
|
|
+// LOG(INFO) << exchange_name << " " << routing_key;
|
|
|
+ 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) {
|
|
|
+ int channel = m_rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_status_vector(vector_index).channel();
|
|
|
+ std::string exchange_name = m_rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_status_vector(
|
|
|
+ vector_index).exchange_name();
|
|
|
+ std::string routing_key = m_rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_status_vector(
|
|
|
+ vector_index).routing_key();
|
|
|
+ int timeout_ms = m_rabbitmq_parameter_all.rabbitmq_parameters().rabbitmq_sender_status_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;
|
|
|
+}
|
|
|
+
|
|
|
+//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;
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+
|
Jeston