全屏网站源码,宝格丽官网,网店推广方式有哪些,wordpress新闻小程序目录 1. 项目介绍和服务器功能设计2. 基础工具安装3. gflags的安装与使用3.1 gflags的介绍3.2 gflags的安装3.3 gflags的认识3.4 gflags的使用 4. gtest的安装与使用4.1 gtest的介绍4.2 gtest的安装4.3 gtest的使用 5 Spdlog日志组件的安装与使用5.1 Spdlog的介绍5.2 Spdlog的安… 目录 1. 项目介绍和服务器功能设计2. 基础工具安装3. gflags的安装与使用3.1 gflags的介绍3.2 gflags的安装3.3 gflags的认识3.4 gflags的使用 4. gtest的安装与使用4.1 gtest的介绍4.2 gtest的安装4.3 gtest的使用 5 Spdlog日志组件的安装与使用5.1 Spdlog的介绍5.2 Spdlog的安装5.3 Spdlog的认识5.4 Spdlog的使用5.5 对Spdlog的进行封装5.6 spdlog 与 glog 组件对比 6. etcd的安装与使用6.1 etcd的介绍6.2 etcd的安装6.3 客户端类的接口介绍6.4 客户端类的使用6.5 封装服务发现与注册功能 7 brpc的安装与使用7.1 brpc的介绍7.2 brpc的安装7.3 brpc的介绍7.4 brpc的使用7.5 brpc的封装 8. ES的安装与使用8.1 ES的介绍8.2 ES和kibana的安装8.3 ES客户端的安装8.4 ES的核心概念8.4 ES的封装 9. cpp-httplib的安装与使用9.1 cpp-httplib的介绍9.2 cpp-httplib的安装及接口介绍9.3 cpp-httplib的使用 10. websocketpp的安装与使用10.1 websocketpp的介绍和原理10.2 websocketpp的使用 11. redis的安装与使用11.1 redis的介绍11.2 redis的安装11.3 redis的部分接口介绍11.4 redis的使用 12. ODB安装与使用12.1 ODB的介绍12.2 ODB的安装 13. RabbitMq的安装和使用13.1 RabbitMq的介绍13.2 RabbitMq的安装13.3 AMQP-CPP 库的简单使用13.4 RabbitMQ的二次封装 14. 短信验证码SDK15.1 短信验证码的安装14.2 接口的使用14.3 短信接口的封装 15. 语音识别SDK15.1 语音识别的安装15.2 语音识别的简单使用15.3 语音识别的封装 16. cmake搭建过程和介绍精简版 1. 项目介绍和服务器功能设计
1项目简介见微服务即时通讯系统的实现客户端博客https://blog.csdn.net/m0_65558082/article/details/143168742?spm1001.2014.3001.5502
2在本项目当中服务器的主要包含了以下功能
用户注册用户输入用户名(昵称)以及密码进行用户名的注册。用户登录用户通过用户名和密码进行登录。短信验证码获取当用户通过手机号注册或登录的时候需要获取短信验证码。手机号注册用户输入手机号和短信验证码进行手机号的用户注册。手机号登录用户输入手机号和短信验证码进行手机号的用户登录。用户信息获取当用户登录之后获取个人信息进行展示。头像修改设置用户头像。昵称修改设置用户昵称。签名修改设置用户签名。手机号修改修改用户的绑定手机号。好友列表的获取当用户登录成功之后获取自己好友列表进行展示。申请好友搜索用户之后点击申请好友向对方发送好友申请。待处理申请的获取当用户登录成功之后会获取离线的好友申请请求以待处理。好友申请的处理针对收到的好友申请进行同意/拒绝的处理。删除好友删除当前好友列表中的好友。用户搜索可以进行用户的搜索用于申请好友。聊天会话列表的获取每个单人/多人聊天都有一个聊天会话在登录成功后可以获取聊天会话查看历史的消息以及对方的各项信息。多人聊天会话的创建单人聊天会话在对方同意好友时创建而多人会话需要调用该接口进行手动创建。聊天成员列表的获取多人聊天会话中可以点击查看群成员按钮查看群成员信息。发送聊天消息在聊天框输入内容后点击发送则向服务器发送消息聊天请求。获取历史消息 获取最近 N 条消息用于登录成功后点击对方头像打开聊天框时显示最近的消息。获取指定时间段内的消息用户可以进行聊天消息的按时间搜索。 消息搜索用户可以进行聊天消息的关键字搜索。文件的上传 单个文件的上传这个接口基本用于后台部分收到文件消息后将文件数据转发给文件子服务进行存储。多个文件的上传这个接口基本用于后台部分收到文件消息后将文件数据转发给文件子服务进行存储。 文件的下载 单个文件的下载在后台用于获取用户头像文件数据以及客户端用于获取文件/语音/图片消息的文件数据。多个文件的下载在后台用于大批量获取用户头像数据比如获取用户列表的时候以及前端的批量文件下载。 语音消息的文字转换客户端进行语音消息的文字转换。
3除了以上的与客户端之间交互的功能之外还包含一些服务器后台内部所需的功能
消息的存储用于将文本消息进行存储起来以便于进行消息的搜索以及离线消息的存储。文件的存储用于存储用户的头像文件以及消息中的文件/图片/语音文件数据。各项用户好友会话数据的存储管理。
2. 基础工具安装
1编辑器安装
C
[xiaomakeralibaba ~]$ sudo apt-get install vim2编译器安装
C
[xiaomakeralibaba ~]$ sudo apt-get install gcc g3调试器安装
C
[xiaomakeralibaba ~]$ sudo apt-get install gdb4项目构建工具安装
C
[xiaomakeralibaba ~]$ sudo apt-get install make cmake5传输工具安装
C
[xiaomakeralibaba ~]$ sudo apt-get install lrzsz6版本管理工具安装
C
[xiaomakeralibaba ~]$ sudo apt-get install git3. gflags的安装与使用
3.1 gflags的介绍
gflags是Google 开发的一个开源库用于 C 应用程序中命令行参数的声明、定义和解析。gflags 库提供了一种简单的方式来添加、解析和文档化命令行标志flags使得程序可以根据不同的运行时配置进行调整。它具有如下几个特点
易于使用gflags 提供了一套简单直观的 API 来定义和解析命令行标志使得开 发者可以轻松地为应用程序添加新的参数。自动帮助和文档gflags 可以自动生成每个标志的帮助信息和文档这有助于用 户理解如何使用程序及其参数。类型安全gflags 支持多种数据类型的标志包括布尔值、整数、字符串等并 且提供了类型检查和转换。多平台支持gflags 可以在多种操作系统上使用包括 Windows、Linux 和macOS。可扩展性gflags 允许开发者自定义标志的注册和解析逻辑提供了强大的扩展性。
官方文档https://gflags.github.io/gflags/ 代码仓库https://github.com/gflags/gflags.git
3.2 gflags的安装
1直接命令行安装
C
[xiaomakeralibaba gflags]$ sudo apt-get install libgflags-dev2原码安装
# 下载源码
git clone https://github.com/gflags/gflags.git
# 切换目录
cd gflags/
# 创建build并进入
mkdir build cd build/
# 生成 Makefile
cmake ..
# 编译代码
make
# 安装
make install3.3 gflags的认识
1使用 gflags 库来定义/解析命令行参数必须包含如下头文件
C
#include gflags/gflags.h2定义参数
利用 gflag 提供的宏定义来定义参数。该宏的 3 个参数分别为命令行参数名参数默认值参数的帮助信息。
C
DEFINE_bool(reuse_addr, true, 是否开始网络地址重用选项);
DEFINE_int32(log_level, 1, 日志等级1-DEBUG, 2-WARN, 3-ERROR);
DEFINE_string(log_file, stdout, 日志输出位置设置默认为标准输出);gflags支持定义多种类型的宏函数
C
DEFINE_bool
DEFINE_int32
DEFINE_int64
DEFINE_uint64
DEFINE_double
DEFINE_string3访问参数
我们可以在程序中通过 FLAGS_name 像正常变量一样访问标志参数。比如在上面的例子中我们可以通过 FLAGS_big_menu 和 FLAGS_languages 变量来访问命令行参数。
4不同文件访问参数
如果想再另外一个文件访问当前文件的参数以参数 FLAGS_big_menu 为例我们可以使用用宏 DECLARE_bool(big_menu)来声明引入这个参数。其实这个宏就相当于做了 extern FLAGS_big_menu 定义外部链接属性。
5初始化所有参数
当我们定义好参数后需要告诉可执行程序去处理解析命令行传入的参数使得FLAGS_*变量能得到正确赋值。我们需要在 main 函数中调用下面的函数来解决命令行传入的所有参数。
C
google::ParseCommandLineFlags(argc, argv, true);参数解析 argc 和 argv 就是 main 的入口参数第三个参数被称为 remove_flags。如果它为 true, 表示
ParseCommandLineFlags 会从 argv 中移除标识和它们的参数相应减少 argc 的值。如果它为 false,ParseCommandLineFlags 会保留 argc 不变但将会重新调整它们的顺序使得标识再前面。
6运行参数设置flags 为我们提供了多种命令行设置参数的方式。
string 和 int 设置参数
Shell
exec --log_file./main.log
exec -log_file./main.log
exec --log_file ./main.log
exec -log_file ./main.logbool 设置参数
Shell
exec --reuse_addr
exec --noreuse_addr
exec --reuse_addrtrue
exec --reuse_addrfalse符号- -将会终止标识的处理。比如在 exec -f1 1 – -f2 2 中 f1 被认为是一个标 识但 f2 不会
7配置文件的使用
配置文件的使用其实就是为了让程序的运行参数配置更加标准化不需要每次运行的时候都手动收入每个参数的数值而是通过配置文件一次编写永久使用。需要注意的是配置文件中选项名称必须与代码中定义的选项名称一致。样例
C
-reuse_addrtrue,
-log_level3
-log_file./log/main.log8特殊参数标识
gflags 也默认为我们提供了几个特殊的标识。
Shell
--help # 显示文件中所有标识的帮助信息
--helpfull # 和-help 一样, 帮助信息更全面一些
--helpshort # 只显示当前执行文件里的标志
--helpxml # 以 xml 方式打印方便处理
--version # 打印版本信息由 google::SetVersionString()设定
--flagfile -flagfilef #从文件 f 中读取命令行参数3.4 gflags的使用
1编写样例代码main.cpp
#include iostream
#include gflags/gflags.hDEFINE_string(ip, 127.0.0.1, 这是服务器的监听IP地址格式127.0.0.1);
DEFINE_int32(port, 8080, 这是服务器的监听端口, 格式: 8080);
DEFINE_bool(debug_enable, true, 是否启用调试模式, 格式true/false);int main(int argc, char *argv[])
{google::ParseCommandLineFlags(argc, argv, true);std::cout FLAGS_ip std::endl;std::cout FLAGS_port std::endl;std::cout FLAGS_debug_enable std::endl;return 0;
}2配置文件编写main.conf
-ip192.168.2.2
-port10001
-debug_enabletrue3Makefile 编写
main:main.cppg $^ -o $ -stdc17 -lgflags.PHONY:clean
clean:rm -f main4运行结果 4. gtest的安装与使用
4.1 gtest的介绍
gtest是一个跨平台的 C单元测试框架由 google 公司发布。gtest 是为了在不同平台上为编写 C单元测试而生成的。它提供了丰富的断言、致命和非致命判断、参数化等等测试所需的宏以及全局测试单元测试组件。
4.2 gtest的安装
1命令行安装
C
[xiaomakeralibaba ~]$ sudo apt-get install libgtest-dev4.3 gtest的使用
1头文件包含
C
#include gtest/gtest.h2框架初始化接口
C
testing::InitGoogleTest(argc, argv);3调用测试样例
C
RUN_ALL_TESTS(); 4TEST 宏
C
//这里不需要双引号且同测试下多个测试样例不能同名
TEST(测试名称, 测试样例名称)
TEST_F(test_fixture,test_name)TEST主要用来创建一个简单测试 它定义了一个测试函数 在这个函数中可以使用任何 C代码并且使用框架提供的断言进行检查。TEST_F主要用来进行多样测试适用于多个测试场景如果需要相同的数据配置的情况 即相同的数据测不同的行为。
5GTest 中的断言的宏可以分为两类
ASSERT_系列如果当前点检测失败则退出当前函数EXPECT_系列如果当前点检测失败则继续往下执行
下面是经常使用的断言介绍
C
// bool 值检查
ASSERT_TRUE(参数)期待结果是 true
ASSERT_FALSE(参数)期待结果是 false
//数值型数据检查
ASSERT_EQ(参数 1参数 2)传入的是需要比较的两个数 equal
ASSERT_NE(参数 1参数 2)not equal不等于才返回 true
ASSERT_LT(参数 1参数 2)less than小于才返回 true
ASSERT_GT(参数 1参数 2)greater than大于才返回 true
ASSERT_LE(参数 1参数 2)less equal小于等于才返回 true
ASSERT_GE(参数 1参数 2)greater equal大于等于才返回 true6测试样例
#include iostream
#include gtest/gtest.hint Add(int num1, int num2)
{return num1 num2;
}TEST(测试名称, 加法用例名称1)
{ASSERT_EQ(Add(10, 20), 30);ASSERT_LT(Add(20, 20), 50);
}TEST(测试名称, 字符串比较测试)
{std::string str Hello;EXPECT_EQ(str, hello);printf(断言失败后的打印\n);EXPECT_EQ(str, Hello);
}int main(int argc, char *argv[])
{//单元测试框架的初始化testing::InitGoogleTest(argc, argv);//开始所有的单元测试return RUN_ALL_TESTS();
}7运行结果 5 Spdlog日志组件的安装与使用
5.1 Spdlog的介绍
1spdlog 是一个高性能、超快速、零配置的 C 日志库它旨在提供简洁的 API 和丰富的功能同时保持高性能的日志记录。它支持多种输出目标、格式化选项、线程安全以及异步日志记录。以下是对 spdlog 的详细介绍和使用方法。github链接https://github.com/gabime/spdlog
2Spdlog的特点
高性能spdlog 专为速度而设计即使在高负载情况下也能保持良好的性能。零配置无需复杂的配置只需包含头文件即可在项目中使用。异步日志支持异步日志记录减少对主线程的影响。格式化支持自定义日志消息的格式化包括时间戳、线程 ID、日志级别等。多平台跨平台兼容支持 Windows、Linux、macOS 等操作系统。丰富的 API提供丰富的日志级别和操作符重载方便记录各种类型的日志。
5.2 Spdlog的安装
1命令行安装
C
[xiaomakeralibaba spdlog]$ sudo apt-get install libspdlog-dev2原码安装
Bash
[xiaomakeralibaba spdlog]$ git clone https://github.com/gabime/spdlog.git
[xiaomakeralibaba spdlog]$ cd spdlog/
[xiaomakeralibaba spdlog]$ mkdir build cd build
[xiaomakeralibaba spdlog]$ cmake -DCMAKE_INSTALL_PREFIX/usr ..
[xiaomakeralibaba spdlog]$ make sudo make install5.3 Spdlog的认识
1在你的 C 源文件中包含 spdlog 的头文件
C
#include spdlog/spdlog.h2日志输出等级枚举
C
namespace level
{enum level_enum : int{trace SPDLOG_LEVEL_TRACE,debug SPDLOG_LEVEL_DEBUG,info SPDLOG_LEVEL_INFO,warn SPDLOG_LEVEL_WARN,err SPDLOG_LEVEL_ERROR,critical SPDLOG_LEVEL_CRITICAL,off SPDLOG_LEVEL_OFF,n_levels};
}3日志输出可以自定义日志消息的格式
C
logger-set_pattern(%Y-%m-%d %H:%M:%S [%t] [%-7l] %v);
%t - 线程 IDThread ID。
%n - 日志器名称Logger name。
%l - 日志级别名称Level name如 INFO, DEBUG, ERROR 等。
%v - 日志内容message。
比特就业课
%Y - 年Year。
%m - 月Month。
%d - 日Day。
%H - 小时24-hour format。
%M - 分钟Minute。
%S - 秒Second。4日志记录器类创建一个基本的日志记录器并设置日志级别和输出模式
C
namespace spdlog
{class logger{logger(std::string name);logger(std::string name, sink_ptr single_sink);logger(std::string name, sinks_init_list sinks);void set_level(level::level_enum log_level);void set_formatter(std::unique_ptrformatter f);template typename... Argsvoid trace(fmt::format_stringArgs... fmt, Args ...args);template typename... Argsvoid debug(fmt::format_stringArgs... fmt, Args ...args);template typename... Argsvoid info(fmt::format_stringArgs... fmt, Args ...args); template typename... Argsvoid warn(fmt::format_stringArgs... fmt, Args ...args); template typename... Argsvoid error(fmt::format_stringArgs... fmt, Args ...args); template typename... Argsvoid critical(fmt::format_stringArgs... fmt, Args ...args);void flush(); // 刷新日志// 策略刷新--触发指定等级日志的时候立即刷新日志的输出void flush_on(level::level_enum log_level);};
}5异步日志记录类为了异步记录日志可以使用 spdlog::async_logger
C
class async_logger final : public logger
{async_logger(std::string logger_name,sinks_init_list sinks_list,std::weak_ptrdetails::thread_pool tp,async_overflow_policy overflow_policy async_overflow_policy::block);async_logger(std::string logger_name,sink_ptr single_sink,std::weak_ptrdetails::thread_pool tp,async_overflow_policy overflow_policy async_overflow_policy::block);// 异步日志输出需要异步工作线程的支持这里是线程池类class SPDLOG_API thread_pool{thread_pool(size_t q_max_items,size_t threads_n,std::functionvoid() on_thread_start,std::functionvoid() on_thread_stop);thread_pool(size_t q_max_items, size_t threads_n,std::functionvoid() on_thread_start);thread_pool(size_t q_max_items, size_t threads_n);};
}
std::shared_ptrspdlog::details::thread_pool thread_pool()
{return details::registry::instance().get_tp();
}// 默认线程池的初始化接口
inline void init_thread_pool(size_t q_size, size_t thread_count) auto async_logger spdlog::async_logger_mt(async_logger,logs/async_log.txt);
async_logger-info(This is an asynchronous info message);6日志记录器工厂类
C
using async_factory async_factory_implasync_overflow_policy::block;
template typename Sink, typename... SinkArgs
inline std::shared_ptrspdlog::logger create_async(std::string logger_name,SinkArgs ...sink_args)// 创建一个彩色输出到标准输出的日志记录器默认工厂创建同步日志记录器template typename Factory spdlog::synchronous_factorystd::shared_ptrlogger stdout_color_mt(const std::string logger_name,color_mode mode color_mode::automatic);
// 标准错误
template typename Factory spdlog::synchronous_factory
std::shared_ptrlogger stderr_color_mt(const std::string logger_name,color_mode mode color_mode::automatic);
// 指定文件
template typename Factory spdlog::synchronous_factory
std::shared_ptrlogger basic_logger_mt(const std::string logger_name,const filename_t filename,bool truncate false,const file_event_handlers event_handlers {})// 循环文件template typename Factory spdlog::synchronous_factorystd::shared_ptrlogger rotating_logger_mt(const std::string logger_name,const filename_t filename,size_t max_file_size,size_t max_files,bool rotate_on_open false)...7日志落地类
C
namespace spdlog
{namespace sinks{class SPDLOG_API sink{public:virtual ~sink() default;virtual void log(const details::log_msg msg) 0;virtual void flush() 0;virtual void set_pattern(const std::string pattern) 0;virtual voidset_formatter(std::unique_ptrspdlog::formatter sink_formatter) 0;void set_level(level::level_enum log_level);};using stderr_color_sink_mt;using stderr_sink_mt;using stdout_color_sink_mt;using stdout_sink_mt;// 滚动日志文件-超过一定大小则自动重新创建新的日志文件sink_ptr rotating_file_sink(filename_t base_filename,std::size_t max_size,std::size_t max_files,bool rotate_on_open false,const file_event_handlers event_handlers {});using rotating_file_sink_mt rotating_file_sinkstd::mutex;// 普通的文件落地对啊 ingsink_ptr basic_file_sink(const filename_t filename,bool truncate false,const file_event_handlers event_handlers {});using basic_file_sink_mt basic_file_sinkstd::mutex;using kafka_sink_mt kafka_sinkstd::mutex;using mongo_sink_mt mongo_sinkstd::mutex;using tcp_sink_mt tcp_sinkstd::mutex;using udp_sink_mt udp_sinkstd::mutex;.....//*_st单线程版本不用加锁效率更高。//*_mt多线程版本用于多线程程序是线程安全的。}
}8全局接口
C
//输出等级设置接口
void set_level(level::level_enum log_level);
//日志刷新策略-每隔 N 秒刷新一次
void flush_every(std::chrono::seconds interval)
//日志刷新策略-触发指定等级立即刷新
void flush_on(level::level_enum log_level);9使用日志记录器记录不同级别的日志
C
logger-trace(This is a trace message);
logger-debug(This is a debug message);
logger-info(This is an info message);
logger-warn(This is a warning message);
logger-error(This is an error message);
logger-critical(This is a critical message);5.4 Spdlog的使用
1同步日志器sync.cpp
#include iostream
#include spdlog/spdlog.h
#include spdlog/sinks/stdout_color_sinks.h
#include spdlog/sinks/basic_file_sink.hint main()
{//设置全局的刷新策略//每秒刷新spdlog::flush_every(std::chrono::seconds());//遇到debug以上等级的日志立即刷新spdlog::flush_on(spdlog::level::level_enum::debug);//设置全局的日志输出等级 -- 无所谓 --每个日志器可以独立进行设置spdlog::set_level(spdlog::level::level_enum::debug);//创建同步日志器标准输出/文件 -- 工厂接口默认创建的就是同步日志器auto logger spdlog::stdout_color_mt(default-logger);//auto logger spdlog::basic_logger_mt(file-logger, sync.log);//设置日志器的刷新策略以及设置日志器的输出等级// logger-flush_on(spdlog::level::level_enum::debug);// logger-set_level(spdlog::level::level_enum::debug);//设置日志输出格式logger-set_pattern([%n][%H:%M:%S][%t][%-8l] %v);//进行简单的日志输出logger-trace(你好{}, 小明);logger-debug(你好{}, 小明);logger-info(你好{}, 小明);logger-warn(你好{}, 小明);logger-error(你好{}, 小明);logger-critical(你好{}, 小明);std::cout 日志输出演示完毕\n;return 0;
}2运行结果 3异步日志输出器async.cpp
#include iostream
#include spdlog/spdlog.h
#include spdlog/sinks/stdout_color_sinks.h
#include spdlog/sinks/basic_file_sink.h
#include spdlog/async.hint main()
{//设置全局的刷新策略//每秒刷新spdlog::flush_every(std::chrono::seconds(1));spdlog::flush_on(spdlog::level::level_enum::debug);spdlog::set_level(spdlog::level::level_enum::debug);//初始化异步日志输出线程配置//void init_thread_pool(size_t q_size, size_t thread_count)spdlog::init_thread_pool(3072, 1);//创建异步日志器标准输出/文件 -- 工厂接口默认创建的就是同步日志器auto logger spdlog::stdout_color_mtspdlog::async_factory(async-logger);//设置日志输出格式logger-set_pattern([%n][%H:%M:%S][%t][%-8l] %v);//进行简单的日志输出logger-trace(你好{}, 小明);logger-debug(你好{}, 小明);logger-info(你好{}, 小明);logger-warn(你好{}, 小明);logger-error(你好{}, 小明);logger-critical(你好{}, 小明);std::cout 日志输出演示完毕\n;return 0;
}4运行结果 5.5 对Spdlog的进行封装
1因为 spdlog 的日志输出对文件名和行号并不是很友好也有可能是调研不到位…以及因为 spdlog 本身实现了线程安全如果使用默认日志器每次进行单例获取效率会有降低因此进行二次封装简化使用。日志的初始化封装接口以及日志的输出接口封装宏。
2封装成logger.hpp
#pragma once
#include iostream
#include spdlog/spdlog.h
#include spdlog/sinks/stdout_color_sinks.h
#include spdlog/sinks/basic_file_sink.h
#include spdlog/async.h// mode - 运行模式 true-发布模式 false调试模式
std::shared_ptrspdlog::logger g_default_logger;
void init_logger(bool mode, const std::string file, int32_t level)
{if(mode false) {//如果是调试模式则创建标准输出日志器输出等级为最低g_default_logger spdlog::stdout_color_mt(default-logger);g_default_logger-set_level(spdlog::level::level_enum::trace);g_default_logger-flush_on(spdlog::level::level_enum::trace);}else {//否则是发布模式则创建文件输出日志器输出等级根据参数而定g_default_logger spdlog::basic_logger_mt(default-logger, file);g_default_logger-set_level((spdlog::level::level_enum)level);g_default_logger-flush_on((spdlog::level::level_enum)level);}g_default_logger-set_pattern([%n][%H:%M:%S][%t][%-8l]%v);
}#define LOG_TRACE(format, ...) g_default_logger-trace(std::string([{}:{}] ) format, __FILE__, __LINE__, ##__VA_ARGS__)
#define LOG_DEBUG(format, ...) g_default_logger-debug(std::string([{}:{}] ) format, __FILE__, __LINE__, ##__VA_ARGS__)
#define LOG_INFO(format, ...) g_default_logger-info(std::string([{}:{}] ) format, __FILE__, __LINE__, ##__VA_ARGS__)
#define LOG_WARN(format, ...) g_default_logger-warn(std::string([{}:{}] ) format, __FILE__, __LINE__, ##__VA_ARGS__)
#define LOG_ERROR(format, ...) g_default_logger-error(std::string([{}:{}] ) format, __FILE__, __LINE__, ##__VA_ARGS__)
#define LOG_FATAL(format, ...) g_default_logger-critical(std::string([{}:{}] ) format, __FILE__, __LINE__, ##__VA_ARGS__)3封装测试
#include gflags/gflags.h
#include logger.hppDEFINE_bool(run_mode, false, 程序的运行模式false-调试 true-发布);
DEFINE_string(log_file, , 发布模式下用于指定日志的输出文件);
DEFINE_int32(log_level, 0, 发布模式下用于指定日志输出等级);int main(int argc, char *argv[])
{google::ParseCommandLineFlags(argc, argv, true);init_logger(FLAGS_run_mode, FLAGS_log_file, FLAGS_log_level);LOG_DEBUG(你好{}, 小明);LOG_INFO(你好{}, 小明);LOG_WARN(你好{}, 小明);LOG_ERROR(你好{}, 小明);LOG_FATAL(你好{}, 小明);LOG_DEBUG(这是一个测试);return -1;
}4运行结果 5整体Makefile
all:sync async main
main:main.cppg $^ -o $ -stdc17 -lspdlog -lfmt -lgflags -lpthread
sync:sync.cppg $^ -o $ -stdc17 -lspdlog -lfmt -lpthread
async:async.cppg $^ -o $ -stdc17 -lspdlog -lfmt -lpthread.PHONY:clean
clean:rm -f main sync async5.6 spdlog 与 glog 组件对比
1glog 和 spdlog 都是流行的 C 日志库它们各自具有不同的特点和优势。以下是对这两个库的对比分析包括性能测试的结果和使用场景的考量。
2使用场景的考量
glog 是由 Google 开发的一个开源 C 日志库它提供了丰富的日志功能包括多种日志级别、条件日志记录、日志文件管理、信号处理、自定义日志格式等。glog 默认情况下是同步记录日志的这意味着每次写日志操作都会阻塞直到日志数据被写入磁盘。spdlog 是一个开源的、高性能的 C 日志库它支持异步日志记录允许在不影响主线程的情况下进行日志写入。spdlog 旨在提供零配置的用户体验只需包含头文件即可使用。它还支持多种输出目标、格式化选项和线程安全。
3性能方面
根据性能对比测试分析glog 在同步调用的场景下的性能较 spdlog 慢。在一台低配的服务器上glog 耗时 1.027 秒处理十万笔日志数据而在固态硬盘上的耗时为 0.475 秒。在同样的性能测试中spdlog 在同步调用的场景下比 glog 快。在低配服务器上的耗时为 0.135 秒而在固态硬盘上的耗时为 0.057 秒。此外spdlog 还提供了异步日志记录的功能其简单异步模式的耗时为 0.158 秒。
4对比总结
性能从性能测试结果来看spdlog 在同步调用场景下的性能优于 glog。当涉及到大量日志数据时spdlog 显示出更快的处理速度。异步日志spdlog 支持异步日志记录这在处理高负载应用程序时非常有用可以减少日志操作对主线程的影响。易用性spdlog 提供了更简单的集成和配置方式只需包含头文件即可使用而glog 可能需要额外的编译和配置步骤。功能glog 提供了一些特定的功能如条件日志记录和信号处理这些在某些场景下可能非常有用。使用场景glog 可能更适合那些对日志性能要求不是特别高但需要一些特定功能的场景。而 spdlog 则适合需要高性能日志记录和异步日志能力的应用程序。
在选择日志库时开发者应根据项目的具体需求和性能要求来决定使用哪个库。如果项目对日志性能有较高要求或者需要异步日志记录来避免阻塞主线程spdlog 可能是更好的选择。如果项目需要一些特定的日志功能或者已经在使用 glog 且没有显著的性能问题那么继续使用 glog 也是合理的。
5总结
spdlog 是一个功能强大且易于使用的 C 日志库它提供了丰富的功能和高性能的日志记录能力。通过简单的 API开发者可以快速地在项目中实现日志记录同时保持代码的清晰和可维护性。无论是在开发阶段还是生产环境中spdlog 都能提供稳定和高效的日志服务。
6. etcd的安装与使用
6.1 etcd的介绍
Etcd 是一个 golang 编写的分布式、高可用的一致性键值存储系统用于配置共享和服务发现等。它使用 Raft 一致性算法来保持集群数据的一致性且客户端通过长连接watch 功能能够及时收到数据变化通知相较于 Zookeeper 框架更加轻量化。以下是关于 etcd 的安装与使用方法的详细介绍。
6.2 etcd的安装
1首先需要在你的系统中安装 Etcd。Etcd 是一个分布式键值存储通常用于服务发现和配置管理。以下是在 Linux 系统上安装 Etcd 的基本步骤
安装 Etcd
Bash
sudo apt-get install etcd当安装的时候系统显示找不到etcd的时候可以分别安装etcd-client、etcd-discovery 和 etcd-server。因为它们都是 etcd 项目的一部分。系统可能将etcd拆分成这三个只需要依次安装即可。
启动 Etcd 服务
Bash
sudo systemctl start etcd设置 Etcd 开机自启
Bash
sudo systemctl enable etcd2节点配置如果是单节点集群其实就可以不用进行配置默认 etcd 的集群节点通信端口为 2380客户端访问端口为 2379。若需要修改则可以配置/etc/default/etcd
Bash
#节点名称默认为 default
ETCD_NAMEetcd1
#数据目录默认为 ${name}.etcd
ETCD_DATA_DIR/var/lib/etcd/default.etcd
#用于客户端连接的 URL。
ETCD_LISTEN_CLIENT_URLShttp://192.168.65.132:2379,http://127.0.0.1:2379
#用于客户端访问的公开也就是提供服务的 URL
ETCD_ADVERTISE_CLIENT_URLShttp://192.168.65.132:2379,http://127.
0.0.1:2379
#用于集群节点间通信的 URL。
ETCD_LISTEN_PEER_URLShttp://192.168.65.132:2380
ETCD_INITIAL_ADVERTISE_PEER_URLShttp://192.168.65.132:2380
#心跳间隔时间-毫秒
ETCD_HEARTBEAT_INTERVAL100
#选举超时时间-毫秒
ETCD_ELECTION_TIMEOUT1000
#以下为集群配置若无集群则需要注销
#初始集群状态和配置--集群中所有节点
#ETCD_INITIAL_CLUSTERetcd1http://192.168.65.132:2380,etcd2http
://192.168.65.132:2381,etcd3http://192.168.65.132:2382
#初始集群令牌-集群的 ID
#ETCD_INITIAL_CLUSTER_TOKENetcd-cluster
#ETCD_INITIAL_CLUSTER_STATEnew
#以下为安全配置如果要求 SSL 连接 etcd 的话把下面的配置启用并修改文件
路径
#ETCD_CERT_FILE/etc/ssl/client.pem
#ETCD_KEY_FILE/etc/ssl/client-key.pem
#ETCD_CLIENT_CERT_AUTHtrue
#ETCD_TRUSTED_CA_FILE/etc/ssl/ca.pem
#ETCD_AUTO_TLStrue
#ETCD_PEER_CERT_FILE/etc/ssl/member.pem
#ETCD_PEER_KEY_FILE/etc/ssl/member-key.pem
#ETCD_PEER_CLIENT_CERT_AUTHfalse
#ETCD_PEER_TRUSTED_CA_FILE/etc/ssl/ca.pem
#ETCD_PEER_AUTO_TLStrue3单节点运行示例
Bash
etcd --name etcd1 --initial-advertise-peer-urls
http://192.168.65.132:2380 \ --listen-peer-urls http://192.168.65.132:2380 \ --listen-client-urls http://192.168.65.132:2379 \ --advertise-client-urls http://192.168.65.132:2379 \--initial-cluster-token etcd-cluster \ --initial-cluster
etcd1http://192.168.65.132:2380,etcd2http://192.168.65.132:2381,
etcd3http://192.168.65.132:2382 \ --initial-cluster-state new nohup1.out 4运行验证
C
etcdctl put mykey this is awesome如果出现报错
Bash
[xiaomakeralibaba ~]$ etcdctl put mykey this is awesome No help topic for put则 sudo vi /etc/profile 在末尾声明环境变量 ETCDCTL_API3 以确定 etcd 版本。
Bash
export ETCDCTL_API3完毕后加载配置文件并重新执行测试指令
Bash
[xiaomakeralibaba ~]$ source /etc/profile
[xiaomakeralibaba ~]$ etcdctl put mykey this is awesome
OK
[xiaomakeralibaba ~]$ etcdctl get mykey
mykey
this is awesome
[xiaomakeralibaba ~]$ etcdctl del mykey在命令行中输入以下命令来停止etcd服务
bash
[xiaomakeralibaba ~]$ sudo systemctl stop etcd为了确保etcd服务已经成功停止可以运行以下命令来检查etcd服务的状态
bash
[xiaomakeralibaba ~]$ sudo systemctl status etcd5搭建服务注册发现中心
使用 Etcd 作为服务注册发现中心你需要定义服务的注册和发现逻辑。这通常涉及到以下几个操作 服务注册服务启动时向 Etcd 注册自己的地址和端口。服务发现客户端通过 Etcd 获取服务的地址和端口用于远程调用。健康检查服务定期向 Etcd 发送心跳以维持其注册信息的有效性。
etcd 采用 golang 编写v3 版本通信采用 grpc API即(HTTP2protobuf) 官方只维护了 go 语言版本的 client 库因此需要找到 C/C 非官方的 client 开发库。 etcd-cpp-apiv3 etcd-cpp-apiv3是一个 etcd 的 C版本客户端 API。它依赖于 mipsasm, boost, protobuf, gRPC, cpprestsdk 等库。etcd-cpp-apiv3 的 GitHub 地址是https://github.com/etcd-cpp-apiv3/etcd-cpp-apiv3 依赖安装
C
sudo apt-get install libboost-all-dev libssl-dev
sudo apt-get install libprotobuf-dev protobuf-compiler-grpc
sudo apt-get install libgrpc-dev libgrpc-dev
sudo apt-get install libcpprest-devetcd-cpp-apiv3框架安装
Bash
# 1. 克隆官方库
git clone https://github.com/etcd-cpp-apiv3/etcd-cpp-apiv3.git
# 2. 进入目录
cd etcd-cpp-apiv3
# 3. 创建build目录
mkdir build cd build
# 4. 构建make
cmake .. -DCMAKE_INSTALL_PREFIX/usr
# 5. 运行安装
make -j$(nproc) sudo make install6.3 客户端类的接口介绍
1客户端类简单的接口展示
// pplx::task 并行库异步结果对象
// 阻塞方式 get(): 阻塞直到任务执行完成并获取任务结果
// 非阻塞方式 wait(): 等待任务到达终止状态然后返回任务状态
namespace etcd
{class Value{bool is_dir(); //判断是否是一个目录std::string const key(); // 键值对的 key 值std::string const as_string(); // 键值对的 val 值int64_t lease(); // 用于创建租约的响应中返回租约 ID};// etcd 会监控所管理的数据的变化一旦数据产生变化会通知客户端// 在通知客户端的时候会返回改变前的数据和改变后的数据class Event{enum class EventType{PUT, // 键值对新增或数据发生改变DELETE_, // 键值对被删除INVALID,};enum EventType event_type();const Value kv();const Value prev_kv();}; class Response{bool is_ok();std::string const error_message();Value const value(); // 当前的数值 或者 一个请求的处理结果Value const prev_value(); // 之前的数值Value const value(int index); //std::vectorEvent const events(); // 触发的事件};class KeepAlive{KeepAlive(Client const client, int ttl, int64_t lease_id 0);// 返回租约 IDint64_t Lease();// 停止保活动作void Cancel();}; class Client{// etcd_url http://127.0.0.1:2379Client(std::string const etcd_url,std::string const load_balancer round_robin);// Put a new key-value pair 新增一个键值对pplx::taskResponse put(std::string const key,std::string const value);// 新增带有租约的键值对 一定时间后如果没有续租数据自动删除pplx::taskResponse put(std::string const key,std::string const value,const int64_t leaseId);// 获取一个指定 key 目录下的数据列表pplx::taskResponse ls(std::string const key);// 创建并获取一个存活 ttl 时间的租约pplx::taskResponse leasegrant(int ttl);// 获取一个租约保活对象其参数 ttl 表示租约有效时间pplx::taskstd::shared_ptrKeepAlive leasekeepalive(intttl);// 撤销一个指定的租约pplx::taskResponse leaserevoke(int64_t lease_id);// 数据锁pplx::taskResponse lock(std::string const key);};class Watcher{Watcher(Client const client,std::string const key, // 要监控的键值对 keystd::functionvoid(Response) callback, // 发生改变后的回调bool recursive false); // 是否递归监控目录下的所有数据改变Watcher(std::string const address,std::string const key,std::functionvoid(Response) callback,bool recursive false);// 阻塞等待直到监控任务被停止bool Wait();bool Cancel();};
}6.4 客户端类的使用
1服务注册
#include etcd/Client.hpp
#include etcd/KeepAlive.hpp
#include etcd/Response.hpp
#include threadint main(int argc, char *argv[])
{std::string etcd_host http://127.0.0.1:2379;// 实例化客户端对象etcd::Client client(etcd_host);// 获取租约保活对象--伴随着创建一个指定有效时长的租约auto keep_alive client.leasekeepalive(3).get();// 获取租约IDauto lease_id keep_alive-Lease();// 向etcd新增数据auto resp1 client.put(/service/user, 127.0.0.1:8080, lease_id).get();if (resp1.is_ok() false){std::cout 新增数据失败 resp1.error_message() std::endl;return -1;}auto resp2 client.put(/service/friend, 127.0.0.1:9090).get();if (resp2.is_ok() false){std::cout 新增数据失败 resp2.error_message() std::endl;return -1;}std::this_thread::sleep_for(std::chrono::seconds(10));return 0;
}2服务发现
#include etcd/Client.hpp
#include etcd/KeepAlive.hpp
#include etcd/Response.hpp
#include etcd/Watcher.hpp
#include etcd/Value.hpp
#include threadvoid callback(const etcd::Response resp)
{if (resp.is_ok() false){std::cout 收到一个错误的事件通知: resp.error_message() std::endl;return;}for (auto const ev : resp.events()){if (ev.event_type() etcd::Event::EventType::PUT){std::cout 服务信息发生了改变:\n;std::cout 当前的值 ev.kv().key() - ev.kv().as_string() std::endl;std::cout 原来的值 ev.prev_kv().key() - ev.prev_kv().as_string() std::endl;}else if (ev.event_type() etcd::Event::EventType::DELETE_){std::cout 服务信息下线被删除:\n;std::cout 当前的值 ev.kv().key() - ev.kv().as_string() std::endl;std::cout 原来的值 ev.prev_kv().key() - ev.prev_kv().as_string() std::endl;}}
}int main(int argc, char *argv[])
{std::string etcd_host http://127.0.0.1:2379;// 实例化客户端对象etcd::Client client(etcd_host);// 获取指定的键值对信息auto resp client.ls(/service).get();if (resp.is_ok() false){std::cout 获取键值对数据失败: resp.error_message() std::endl;return -1;}int sz resp.keys().size();for (int i 0; i sz; i){std::cout resp.value(i).as_string() 可以提供 resp.key(i) 服务\n;}// 实例化一个键值对事件监控对象etcd::Watcher watcher(client, /service, callback, true);watcher.Wait();return 0;
}3Makefile
all:put get
put:put.cppg -stdc17 $^ -o $ -letcd-cpp-api -lcpprest
get:get.cppg -stdc17 $^ -o $ -letcd-cpp-api -lcpprest.PHONY:clean
clean:rm -rf put get4运行结果 6.5 封装服务发现与注册功能
1服务注册 主要是在 etcd 服务器上存储一个租期 ns 的保活键值对表示所能提供指定服务的节点主机比如 /service/user/instance-1 的 key且对应的 val 为提供服务的主机节点地址key, val – /service/user/instance-1, 127.0.0.1:9000 /service 是主目录其下会有不同服务的键值对存储/user 是服务名称表示该键值对是一个用户服务的节点/instance-1 是节点实例名称提供用户服务可能会有很多节点每个节点都应该有自己独立且唯一的实例名称。 当这个键值对注册之后服务发现方可以基于目录进行键值对的发现。且一旦注册节点退出保活失败则 3s 后租约失效键值对被删除etcd 会通知发现方数据的失效进而实现服务下线通知的功能。
2服务发现 服务发现分为两个过程 刚启动客户端的时候进行 ls 目录浏览进行/service 路径下所有键值对的获取。对关心的服务进行 watcher 观测一旦数值发生变化新增/删除收到通知进行节点的管理。 如果 ls 的路径为/service则会获取到 /service/user, /service/firend, …等其路径下的所有能够提供服务的实例节点数据。 如果 ls 的路径为 /service/user 则会获取到 /service/user/instancd-1, /service/user/instance-2,…等所有提供用户服务的实例节点数据。 客户端可以将发现的所有实例 - 地址管理起来以便于进行节点的管理 收到新增数据通知则向本地管理添加新增的节点地址 – 服务上线。收到删除数据通知则从本地管理删除对应的节点地址 – 服务下线。 因为管理了所有的能够提供服务的节点主机的地址因此当需要进行 rpc 调用的时候则根据服务名称获取一个能够提供服务的主机节点地址进行访问就可以了而这里的获取策略我们采用 RR 轮转策略。
3封装思想
将 etcd 的操作全部封装起来也不需要管理数据只需要向外四个基础操作接口 进行服务注册也就是向 etcd 添加 服务-主机地址的数据进行服务发现获取当前所有能提供服务的信息设置服务上线的处理回调接口设置服务下线的处理回调接口
这样封装之后外部的 rpc 调用模块可以先获取所有的当前服务信息建立通信连接进行 rpc 调用也能在有新服务上线的时候新增连接以及下线的时候移除连接。
4具体实现
#pragma once
#include etcd/Client.hpp
#include etcd/KeepAlive.hpp
#include etcd/Response.hpp
#include etcd/Watcher.hpp
#include etcd/Value.hpp
#include thread
#include memory
#include logger.hppnamespace Test
{// 服务注册客户端类class Registry{public:using ptr std::shared_ptrRegistry;Registry(const std::string host):_client(std::make_sharedetcd::Client(host)),_keep_clive(_client-leasekeepalive(3).get()),_lease_id(_keep_clive-Lease()){}bool registry(const std::string key, const std::string val){auto resp _client-put(key, val, _lease_id).get();if(resp.is_ok() false){LOG_ERROR(注册数据失败{}, resp.error_message());return false;}return true;}private:std::shared_ptretcd::Client _client;std::shared_ptretcd::KeepAlive _keep_clive;uint64_t _lease_id;};//服务发现客户端类class Discovery {public:using ptr std::shared_ptrDiscovery;using NotifyCallback std::functionvoid(std::string, std::string);Discovery(const std::string host, const std::string basedir,const NotifyCallback put_cb,const NotifyCallback del_cb):_client(std::make_sharedetcd::Client(host)),_put_cb(put_cb),_del_cb(del_cb){//先进行服务发现,先获取到当前已有的数据auto resp _client-ls(basedir).get();if(resp.is_ok() false) {LOG_ERROR(获取服务信息数据失败{}, resp.error_message());}int sz resp.keys().size();for(int i 0; i sz; i) {if (_put_cb) _put_cb(resp.key(i), resp.value(i).as_string());}//然后进行事件监控监控数据发生的改变并调用回调进行处理_watcher std::make_sharedetcd::Watcher(*_client.get(), basedir,std::bind(Discovery::callback, this, std::placeholders::_1), true);}private:void callback(const etcd::Response resp){if(resp.is_ok() false){LOG_ERROR(收到一个错误的事件通知: {}, resp.error_message());return;}for(auto const ev : resp.events()){if(ev.event_type() etcd::Event::EventType::PUT){if (_put_cb)_put_cb(ev.kv().key(), ev.kv().as_string());LOG_DEBUG(新增服务{}-{}, ev.kv().key(), ev.kv().as_string());}else if(ev.event_type() etcd::Event::EventType::DELETE_){if(_del_cb){_del_cb(ev.prev_kv().key(), ev.prev_kv().as_string());}LOG_DEBUG(下线服务{}-{}, ev.prev_kv().key(), ev.prev_kv().as_string());}}}private:NotifyCallback _put_cb;NotifyCallback _del_cb;std::shared_ptretcd::Client _client;std::shared_ptretcd::Watcher _watcher;};
}7 brpc的安装与使用
7.1 brpc的介绍
1brpc 是用 c语言编写的工业级 RPC 框架常用于搜索、存储、机器学习、广告、推荐等高性能系统。你可以使用它
搭建能在一个端口支持多协议的服务, 或访问各种服务 restful http/https, h2/gRPC。使用 brpc 的 http 实现比 libcurl 方便多了。从其他语言通过 HTTP/h2json 访问基于 protobuf 的协议。redis 和 memcached线程安全比官方 client 更方便。rtmp/flv/hls可用于搭建流媒体服务。支持 thrift 线程安全比官方 client 更方便。各种百度内使用的协议: baidu_stdstreaming_rpchulu_pbrpcsofa_pbrpcnova_pbrpcpublic_pbrpcubrpc 和使用 nshead 的各种协议。基于工业级的 RAFT 算法实现搭建高可用分布式系统已在 braft 开源。 Server 能同步或异步处理请求。Client 支持同步、异步、半同步或使用组合 channels 简化复杂的分库或并发访问。通过 http 界面调试服务, 使用 cpuheapcontention profilers。获得更好的延时和吞吐。把你组织中使用的协议快速地加入 brpc或定制各类组件包括命名服务 (dns, zk, etcd)负载均衡 (rr、random、consistent hashing)。
7.2 brpc的安装
1先安装依赖
C
[xiaomakeralibaba brpc]$ sudo apt-get install -y git g make libssl-dev libprotobuf-dev libprotoc-dev protobuf-compiler libleveldb-dev2安装 brpc
C
[xiaomakeralibaba brpc]$ git clone https://github.com/apache/brpc.git
[xiaomakeralibaba brpc]$ cd brpc/
[xiaomakeralibaba brpc]$ mkdir build cd build
[xiaomakeralibaba build]$ cmake -DCMAKE_INSTALL_PREFIX/usr .. cmake --build . -j6
[xiaomakeralibaba build]$ make sudo make install7.3 brpc的介绍
1日志输出类与接口
包含头文件 #include butil/logging.h 日志输出这里本质上我们其实用不着 brpc 的日志输出因此在这里主要介绍如何关闭日志输出。
C
namespace logging
{enum LoggingDestination{LOG_TO_NONE 0};struct BUTIL_EXPORT LoggingSettings{LoggingSettings();LoggingDestination logging_dest;};bool InitLogging(const LoggingSettings settings);
}2protobuf 类与接口
C
namespace google
{namespace protobuf{class PROTOBUF_EXPORT Closure{public:Closure() {}virtual ~Closure();virtual void Run() 0;};inline Closure *NewCallback(void (*function)());class PROTOBUF_EXPORT RpcController{bool Failed();std::string ErrorText();}}
}3服务端类与接口这里只介绍主要用到的成员与接口。
C
namespace brpc
{struct ServerOptions{// 无数据传输则指定时间后关闭连接int idle_timeout_sec; // Default: -1 (disabled)int num_threads; // Default: #cpu-cores//....};enum ServiceOwnership {// 添加服务失败时服务器将负责删除服务对象SERVER_OWNS_SERVICE,// 添加服务失败时服务器也不会删除服务对象SERVER_DOESNT_OWN_SERVICE};class Server{int AddService(google::protobuf::Service *service,ServiceOwnership ownership);int Start(int port, const ServerOptions *opt);int Stop(int closewait_ms /*not used anymore*/);int Join();// 休眠直到 ctrlc 按下或者 stop 和 join 服务器void RunUntilAskedToQuit();};class ClosureGuard{explicit ClosureGuard(google::protobuf::Closure *done);~ClosureGuard(){if (_done)_done-Run();}}; class HttpHeader{void set_content_type(const std::string type);const std::string *GetHeader(const std::string key);void SetHeader(const std::string key, const std::string value); const URI uri() const { return _uri; }HttpMethod method() const { return _method; }void set_method(const HttpMethod method); int status_code(); void set_status_code(int status_code);};class Controller : public google::protobuf::RpcController{void set_timeout_ms(int64_t timeout_ms);void set_max_retry(int max_retry);google::protobuf::Message *response();HttpHeader http_response();HttpHeader http_request();bool Failed();std::string ErrorText();using AfterRpcRespFnType std::functionvoid(Controller *cntl, const google::protobuf::Message *req, const google::protobuf::Message *res);void set_after_rpc_resp_fn(AfterRpcRespFnType fn);};
}4客户端类与接口
C
namespace brpc
{struct ChannelOptions{// 请求连接超时时间int32_t connect_timeout_ms; // Default: 200 (milliseconds)// rpc 请求超时时间int32_t timeout_ms; // Default: 500 (milliseconds)// 最大重试次数int max_retry; // Default: 3// 序列化协议类型 options.protocol baidu_std;AdaptiveProtocolType protocol;//....};class Channel : public ChannelBase{// 初始化接口成功返回 0int Init(const char *server_addr_and_port,const ChannelOptions *options);};
}7.4 brpc的使用
1同步调用
同步调用是指客户端会阻塞收到 server 端的响应或发生错误。创建 proto 文件 - main.proto
syntaxproto3;package example;option cc_generic_services true;message EchoRequest {string message 1;
}message EchoResponse {string message 1;
}service EchoService {rpc Echo(EchoRequest) returns (EchoResponse);
}之后运行如下命令也可以放入Makefile当中
protoc --cpp_out./ main.proto创建服务端代码server.cpp
#include brpc/server.h
#include butil/logging.h
#include main.pb.h// 1. 继承于EchoService创建一个子类并实现rpc调用的业务功能
class EchoServiceImpl : public example::EchoService
{
public:EchoServiceImpl(){}void Echo(google::protobuf::RpcController* controller,const ::example::EchoRequest* request,::example::EchoResponse* response,::google::protobuf::Closure* done){brpc::ClosureGuard rpc_guard(done);std::cout 收到消息: request-message() std::endl;std::string str request-message() --这是响应;response-set_message(str);}~EchoServiceImpl(){}
};int main(int argc, char *argv[])
{// 关闭brpc的默认日志输出logging::LoggingSettings settings;settings.logging_dest logging::LoggingDestination::LOG_TO_NONE;logging::InitLogging(settings);// 2. 构造服务器对象brpc::Server server;// 3. 向服务器对象中新增EchoService服务EchoServiceImpl echo_service;int ret server.AddService(echo_service, brpc::ServiceOwnership::SERVER_DOESNT_OWN_SERVICE);if(ret -1){std::cout 添加Rpc服务失败\n;return -1;}// 4. 启动服务器brpc::ServerOptions options;options.idle_timeout_sec -1; //连接空闲超时时间-超时后连接被关闭options.num_threads 1; // IO线程数量ret server.Start(8080, options);if(ret -1){std::cout 启动服务器失败\n;return -1;}server.RunUntilAskedToQuit();//修改等待运行结束return 0;
}创建客户端源码client.cpp
#include brpc/channel.h
#include thread
#include main.pb.hint main()
{//1. 构造Channel信道连接服务器brpc::ChannelOptions options;options.connect_timeout_ms -1;// 连接等待超时时间-1表示一直等待options.timeout_ms -1; //rpc请求等待超时时间-1表示一直等待options.max_retry 3;//请求重试次数options.protocol baidu_std; //序列化协议默认使用baidu_stdbrpc::Channel channel;int ret channel.Init(127.0.0.1:8080, options);if(ret -1){std::cout 初始化信道失败\n;return -1;}//2. 构造EchoService_Stub对象用于进行rpc调用example::EchoService_Stub stub(channel);//3. 进行Rpc调用/example::EchoRequest req;req.set_message(你好~小明~);brpc::Controller *cntl new brpc::Controller();example::EchoResponse *rsp new example::EchoResponse();stub.Echo(cntl, req, rsp, nullptr);if (cntl-Failed() true) {std::cout Rpc调用失败 cntl-ErrorText() std::endl;return -1;}std::cout 收到响应: rsp-message() std::endl;delete cntl;delete rsp;std::cout 异步调用结束\n;std::this_thread::sleep_for(std::chrono::seconds(3));return 0;
}编写Makefile
all:server client
server:server.cpp main.pb.ccg $^ -o $ -stdc17 -lbrpc -lgflags -lssl -lcrypto -lprotobuf -lleveldb
client:client.cpp main.pb.ccg $^ -o $ -stdc17 -lbrpc -lgflags -lssl -lcrypto -lprotobuf -lleveldb.PHONY:clean
clean:rm -rf server client运行结果 2异步调用
异步调用是指客户端注册一个响应处理回调函数 当调用一个 RPC 接口时立即返回不会阻塞等待响应 当 server 端返回响应时会调用传入的回调函数处理响应。具体的做法给 CallMethod 传递一个额外的回调对象 doneCallMethod 在发出request 后就结束了而不是在 RPC 结束后。当 server 端返回 response 或发生错误包括超时时done-Run()会被调用。对 RPC 的后续处理应该写在 done-Run()里而不是 CallMethod 后。由于 CallMethod 结束不意味着 RPC 结束response/controller 仍可能被框架及 done-Run()使用它们一般得创建在堆上并在 done-Run()中删除。如果提前删除了它们那当 done-Run()被调用时将访问到无效内存。
#include brpc/channel.h
#include thread
#include main.pb.hvoid callback(brpc::Controller* cntl, ::example::EchoResponse* response)
{std::unique_ptrbrpc::Controller cntl_guard(cntl);std::unique_ptrexample::EchoResponse resp_guard(response);if(cntl-Failed() true){std::cout Rpc调用失败 cntl-ErrorText() std::endl;return;}std::cout 收到响应: response-message() std::endl;
}int main()
{//1. 构造Channel信道连接服务器brpc::ChannelOptions options;options.connect_timeout_ms -1;// 连接等待超时时间-1表示一直等待options.timeout_ms -1; //rpc请求等待超时时间-1表示一直等待options.max_retry 3;//请求重试次数options.protocol baidu_std; //序列化协议默认使用baidu_stdbrpc::Channel channel;int ret channel.Init(127.0.0.1:8080, options);if(ret -1){std::cout 初始化信道失败\n;return -1;}//2. 构造EchoService_Stub对象用于进行rpc调用example::EchoService_Stub stub(channel);//3. 进行Rpc调用/example::EchoRequest req;req.set_message(你好~小明~);brpc::Controller *cntl new brpc::Controller();example::EchoResponse *rsp new example::EchoResponse();auto clusure google::protobuf::NewCallback(callback, cntl, rsp);stub.Echo(cntl, req, rsp, clusure); //异步调用std::cout 异步调用结束\n;std::this_thread::sleep_for(std::chrono::seconds(3));return 0;
}7.5 brpc的封装
1封装思想
rpc 调用这里的封装因为不同的服务调用使用的是不同的 Stub这个封装起来的意义不大因此我们只需要封装通信所需的 Channel 管理即可这样当需要进行什么样的服务调用的时候只需要通过服务名称获取对应的 channel然后实例化 Stub 进行调用即可。封装 Channel 的管理每个不同的服务可能都会有多个主机提供服务因此一个服务可能会对应多个 Channel需要将其管理起来并提供获取指定服务 channel 的接口。进行 rpc 调用时获取 channel目前以 RR 轮转的策略选择 channel。提供进行服务声明的接口因为在整个系统中提供的服务有很多但是当前可能并不一定会用到所有的服务因此通过声明来告诉模块哪些服务是自己关心的需要建立连接管理起来没有添加声明的服务即使上线也不需要进行连接的建立。提供服务上线时的处理接口也就是新增一个指定服务的 channel。提供服务下线时的处理接口也就是删除指定服务下的指定 channel。
2具体实现channel.hpp
#pragma once
#include brpc/channel.h
#include string
#include vector
#include memory
#include unordered_map
#include unordered_set
#include mutex
#include logger.hppnamespace Test
{// 封装单个服务的信道管理类:class ServiceChannel{public:using ChannelPtr std::shared_ptrbrpc::Channel;using Ptr std::shared_ptrServiceChannel;ServiceChannel(const std::string name):_service_name(name),_index(0){}// 服务上线了一个节点则调用append新增信道void append(const std::string host){auto channel std::make_sharedbrpc::Channel();brpc::ChannelOptions options;options.connect_timeout_ms -1;options.timeout_ms -1;options.max_retry 3;options.protocol baidu_std;int ret channel-Init(host.c_str(), options);if(ret -1){LOG_ERROR(初始化{}-{}信道失败!, _service_name, host);return;}std::unique_lockstd::mutex lock(_mutex);_hosts.insert(std::make_pair(host, channel));_channels.push_back(channel);}// 服务下线了一个节点则调用remove释放信道void remove(const std::string host){std::unique_lockstd::mutex lock(_mutex);auto iter _hosts.find(host);if(iter _hosts.end()){LOG_WARN({}-{}节点删除信道时没有找到信道信息, _service_name, host);return;}for(auto vit _channels.begin(); vit ! _channels.end(); vit){if (*vit iter-second){_channels.erase(vit);break;}}_hosts.erase(iter);}// 通过RR轮转策略获取一个Channel用于发起对应服务的Rpc调用ChannelPtr choose(){std::unique_lockstd::mutex lock(_mutex);if(_channels.size() 0){LOG_ERROR(当前没有能够提供 {} 服务的节点, _service_name);return ChannelPtr();}int32_t idx _index % _channels.size();return _channels[idx];}private:std::mutex _mutex;int32_t _index; // 当前轮转下标计数器std::string _service_name; // 服务名称std::vectorChannelPtr _channels; // 当前服务对应的信道集合std::unordered_mapstd::string, ChannelPtr _hosts; // 主机地址与信道映射关系};// 总体的服务信道管理类class ServiceManager{public:using Ptr std::shared_ptrServiceManager;ServiceManager(){}// 获取指定服务的节点信道ServiceChannel::ChannelPtr choose(const std::string service_name){std::unique_lockstd::mutex lock(_mutex);auto iter _services.find(service_name);if(iter _services.end()){LOG_ERROR(当前没有能够提供 {} 服务的节点, service_name);return ServiceChannel::ChannelPtr();}return iter-second-choose();}// 先声明我关注哪些服务的上下线不关心的就不需要管理了void declared(const std::string service_name){std::unique_lockstd::mutex lock(_mutex);_follow_services.insert(service_name);}// 服务上线时调用的回调接口将服务节点管理起来void onServiceOnline(const std::string service_instance, const std::string host){std::string service_name getServiceName(service_instance);ServiceChannel::Ptr service;{std::unique_lockstd::mutex lock(_mutex);auto fit _follow_services.find(service_name);if(fit _follow_services.end()){LOG_DEBUG({}-{} 服务上线了但是当前并不关心, service_name, host);return;}// 先获取管理对象没有则创建有则添加节点auto sit _services.find(service_name);if(sit _services.end()){service std::make_sharedServiceChannel(service_name);_services.insert(std::make_pair(service_name, service));}else{service sit-second;}}if(!service) {LOG_ERROR(新增 {} 服务管理节点失败, service_name);return;}service-append(host);LOG_DEBUG({}-{} 服务上线新节点进行添加管理, service_name, host);}// 服务下线时调用的回调接口从服务信道管理中删除指定节点信道void onServiceOffline(const std::string service_instance, const std::string host){std::string service_name getServiceName(service_instance);ServiceChannel::Ptr service;{std::unique_lockstd::mutex lock(_mutex);auto fit _follow_services.find(service_name);if(fit _follow_services.end()){LOG_DEBUG({}-{} 服务下线了但是当前并不关心, service_name, host);return;}// 先获取管理对象没有则创建有则添加节点auto sit _services.find(service_name);if(sit _services.end()){LOG_WARN(删除{}服务节点时没有找到管理对象, service_name);return;}service sit-second;}service-remove(host);LOG_DEBUG({}-{} 服务下线节点进行删除管理, service_name, host);}private:std::string getServiceName(const std::string service_instance){auto pos service_instance.find_last_of(/);if(pos std::string::npos){return service_instance;}return service_instance.substr(0, pos);}private:std::mutex _mutex;std::unordered_setstd::string _follow_services;std::unordered_mapstd::string, ServiceChannel::Ptr _services;};
}8. ES的安装与使用
8.1 ES的介绍
Elasticsearch 简称 ES它是个开源分布式搜索引擎它的特点有分布式零配置自动发现索引自动分片索引副本机制restful 风格接口多数据源自动搜索负载等。它可以近乎实时的存储、检索数据本身扩展性很好可以扩展到上百台服务器处理 PB 级别的数据。es 也使用 Java 开发并使用 Lucene 作为其核心来实现所有索引和搜索的功能但是它的目的是通过简单的 RESTful API 来隐藏 Lucene 的复杂性从而让全文搜索变得简单。Elasticsearch 是面向文档(document oriented)的这意味着它可以存储整个对象或文档(document)。然而它不仅仅是存储还会索引(index)每个文档的内容使之可以被搜索。在 Elasticsearch 中你可以对文档而非成行成列的数据进行索引、搜索、排序、过滤。
8.2 ES和kibana的安装
1安装ES
Shell
# 添加仓库秘钥
wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -
# 上边的添加方式会导致一个 apt-key 的警告如果不想报警告使用下边这个
curl -s https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo gpg --no-default-keyring --keyring gnupgring:/etc/apt/trusted.gpg.d/icsearch.gpg --import
# 可以添加镜像源仓库
echo deb https://artifacts.elastic.co/packages/7.x/apt stable main | sudo tee /etc/apt/sources.list.d/elasticsearch.list
# 更新软件包列表
sudo apt update
# 安装 es
sudo apt-get install elasticsearch7.17.21
# 启动 es
sudo systemctl start elasticsearch
# 安装 ik 分词器插件
sudo /usr/share/elasticsearch/bin/elasticsearch-plugin install https://get.infini.cloud/elasticsearch/analysis-ik/7.17.21启动 es
Shell
sudo systemctl start elasticsearch启动 es 的时候报错 2解决办法
Shell
#调整 ES 虚拟内存虚拟内存默认最大映射数为 65530无法满足 ES 系统要求需要调整为 262144 以上
sysctl -w vm.max_map_count262144
#增加虚拟机内存配置
vim /etc/elasticsearch/jvm.options
#新增如下内容
-Xms512m
-Xmx512mes 服务的状态
Shell
sudo systemctl status elasticsearch.service验证 es 是否安装成功
Shell
curl -X GET http://localhost:9200/设置外网访问如果新配置完成的话默认只能在本机进行访问。
Shell
# 打开如下文件
vim /etc/elasticsearch/elasticsearch.yml
# 新增以下配置
network.host: 0.0.0.0
http.port: 9200
cluster.initial_master_nodes: [node-1]浏览器访问http://47.100.90.125:9200/
3安装 kibana
Shell
# 使用 apt 命令安装 Kibana。
sudo apt install kibana
# 配置 Kibana可选根据需要配置 Kibana。配置文件通常位于 /etc/kibana/kibana.yml。可能需要设置如服务器地址、端口、Elasticsearch URL 等。
sudo vim /etc/kibana/kibana.yml
# 例如你可能需要设置 Elasticsearch 服务的 URL 大概 32 行左右elasticsearch.host: http://localhost:9200
# 启动 Kibana 服务
sudo systemctl start kibana
# 设置开机自启可选如果你希望 Kibana 在系统启动时自动启动可以使用以下命令来启用自启动。
sudo systemctl enable kibana
# 验证安装使用以下命令检查 Kibana 服务的状态。
sudo systemctl status kibana
# 访问 Kibana在浏览器中访问 Kibana通常是 http://your-ip:56018.3 ES客户端的安装
代码https://github.com/seznam/elasticlient 官网https://seznam.github.io/elasticlient/index.html
ES C的客户端选择并不多 我们这里使用 elasticlient 库, 下面进行安装。
Shell
# 克隆代码
git clone https://github.com/seznam/elasticlient
# 切换目录
cd elasticlient
# 更新子模块
git submodule update --init --recursive
# 编译代码
mkdir build cd build
# cmake生成Makefile
cmake -DCMAKE_INSTALL_PREFIX/usr ..
# 安装
make make installcmake 生成 makefile 的过程会遇到一个问题 需要安装 MicroHTTPD 库
Shell
sudo apt-get install libmicrohttpd-dev8.4 ES的核心概念
1索引Index
一个索引就是一个拥有几分相似特征的文档的集合。比如说你可以有一个客户数据的索引一个产品目录的索引还有一个订单数据的索引。一个索引由一个名字来标识必须全部是小写字母的并且当我们要对应于这个索引中的文档进行索引、搜索、更新和删除的时候都要使用到这个名字。在一个集群中可以定义任意多的索引。
2类型Type
在一个索引中你可以定义一种或多种类型。一个类型是你的索引的一个逻辑上的分类/分区其语义完全由你来定。通常会为具有一组共同字段的文档定义一个类型。比如说我们假设你运营一个博客平台并且将你所有的数据存储到一个索引中。在这个索引中你可以为用户数据定义一个类型为博客数据定义另一个类型为评论数据定义另一个类型…
3字段Field
字段相当于是数据表的字段对文档数据根据不同属性进行的分类标识。
分类类型备注字符串text, keywordtext 会被分词生成索引;keyword 不会被分词生成索引只能精确值搜索整形integer, long, short, byte浮点double float逻辑booleantrue 或 false日期date, date_nanos“2018-01-13” 或 “2018-01-13 12:10:30”或者时间戳即 1970 到现在的秒数/毫秒数二进制binary二进制通常只存储不索引范围range
4映射mapping
映射是在处理数据的方式和规则方面做一些限制如某个字段的数据类型、默认值、分析器、是否被索引等等这些都是映射里面可以设置的其它就是处理 es 里面数据的一些使用规则设置也叫做映射按着最优规则处理数据对性能提高很大因此才需要建立映射并且需要思考如何建立映射才能对性能更好。
名称数值备注enabledtrue(默认) false是否仅作存储不做搜索和分析indextrue(默认) false是否构建倒排索引决定了是否分词是否被索引index_optiondynamictrue缺省 false 控制mapping 的自动更新doc_valuetrue(默认) false是否开启 doc_value用户聚合和排序分析分词字段不能使用fielddatafielddata: {“format”: “disabled”}是否为 text 类型启动 fielddata实现排序和聚合分析针对分词字段参与排序或聚合时能提高性能不分词字段统一建议使用 doc_valuestoretrue false(默认)是否单独设置此字段的是否存储而从_source 字段中分离只能搜索不能获取值coercetrue(默认) false是否开启自动数据类型转换功能比如字符串转数字浮点转整型analyzer“analyzer”: “ik”指定分词器默认分词器为 standard analyzerboost“boost”: 1.23字段级别的分数加权默认值是 1.0fields“fields”: {“raw”: {“type”: “text”, “index”: “not_analyzed”}}对一个字段提供多种索引模式同一个字段的值一个分词一个不分词data_detectiontrue(默认) false是否自动识别日期类型
5文档 document
一个文档是一个可被索引的基础信息单元。比如你可以拥有某一个客户的文档某一个产品的一个文档或者某个订单的一个文档。文档以 JSONJavascript Object Notation格式来表示而 JSON 是一个到处存在的互联网数据交互格式。在一个index/type 里面你可以存储任意多的文档。一个文档必须被索引或者赋予一个索引的 type。Elasticsearch 与传统关系型数据库相比如下 8.4 ES的封装
1封装客户端 api 主要是因为客户端只提供了基础的数据存储获取调用功能无法根据我们的思想完成索引的构建以及查询正文的构建需要使用者自己组织好 json 进行序列化后才能作为正文进行接口的调用。
2而封装的目的就是简化用户的操作将索引的 json 正文构造以及查询搜索的正文构造操作给封装起来使用者调用接口添加字段就行不用关心具体的 json 数据格式。所以封装的主要内容 索引构造过程的封装 索引正文构造过程大部分正文都是固定的唯一不同的地方是各个字段不同的名称以及是否只存储不索引这些选项因此重点关注以下几个点即可 字段类型type : text / keyword (目前只用到这两个类型)。是否索引enable : true/false。索引的话分词器类型 analyzer : ik_max_word / standard。 新增文档构造过程的封装 新增文档其实在常规下都是单条新增并非批量新增因此直接添加字段和值就行 文档搜索构造过程的封装 搜索正文构造过程我们默认使用条件搜索我们主要关注的以下几点 应该遵循的条件是什么should 中有什么。条件的匹配方式是什么match 还是 term/terms还是 wildcard。过滤的条件字段是什么must_not 中有什么。过滤的条件字段匹配方式是什么match 还是 wildcard还是 term/terms。
整个封装的过程其实就是对 Json::Value 对象的一个组织的过程。
3具体封装结果
#include icsearch.hpp
#include user.hxx
#include message.hxxnamespace Test
{class ESClientFactory{public:static std::shared_ptrelasticlient::Client create(const std::vectorstd::string host_list){return std::make_sharedelasticlient::Client(host_list);}};class ESUser{public:using ptr std::shared_ptrESUser;ESUser(const std::shared_ptrelasticlient::Client client) : _es_client(client) {}bool createIndex(){bool ret ESIndex(_es_client, user).append(user_id, keyword, standard, true).append(nickname).append(phone, keyword, standard, true).append(description, text, standard, false).append(avatar_id, keyword, standard, false).create();if (ret false){LOG_INFO(用户信息索引创建失败!);return false;}LOG_INFO(用户信息索引创建成功!);return true;}bool appendData(const std::string uid,const std::string phone,const std::string nickname,const std::string description,const std::string avatar_id){bool ret ESInsert(_es_client, user).append(user_id, uid).append(nickname, nickname).append(phone, phone).append(description, description).append(avatar_id, avatar_id).insert(uid);if (ret false){LOG_ERROR(用户数据插入/更新失败!);return false;}LOG_INFO(用户数据新增/更新成功!);return true;}std::vectorUser search(const std::string key, const std::vectorstd::string uid_list){std::vectorUser res;Json::Value json_user ESSearch(_es_client, user).append_should_match(phone.keyword, key).append_should_match(user_id.keyword, key).append_should_match(nickname, key).append_must_not_terms(user_id.keyword, uid_list).search();if (json_user.isArray() false){LOG_ERROR(用户搜索结果为空或者结果不是数组类型);return res;}int sz json_user.size();LOG_DEBUG(检索结果条目数量{}, sz);for (int i 0; i sz; i){User user;user.user_id(json_user[i][_source][user_id].asString());user.nickname(json_user[i][_source][nickname].asString());user.description(json_user[i][_source][description].asString());user.phone(json_user[i][_source][phone].asString());user.avatar_id(json_user[i][_source][avatar_id].asString());res.push_back(user);}return res;}private:// const std::string _uid_key user_id;// const std::string _desc_key user_id;// const std::string _phone_key user_id;// const std::string _name_key user_id;// const std::string _avatar_key user_id;std::shared_ptrelasticlient::Client _es_client;};class ESMessage{public:using ptr std::shared_ptrESMessage;ESMessage(const std::shared_ptrelasticlient::Client es_client) : _es_client(es_client) {}bool createIndex(){bool ret ESIndex(_es_client, message).append(user_id, keyword, standard, false).append(message_id, keyword, standard, false).append(create_time, long, standard, false).append(chat_session_id, keyword, standard, true).append(content).create();if (ret false){LOG_INFO(消息信息索引创建失败!);return false;}LOG_INFO(消息信息索引创建成功!);return true;}bool appendData(const std::string user_id,const std::string message_id,const long create_time,const std::string chat_session_id,const std::string content){bool ret ESInsert(_es_client, message).append(message_id, message_id).append(create_time, create_time).append(user_id, user_id).append(chat_session_id, chat_session_id).append(content, content).insert(message_id);if (ret false){LOG_ERROR(消息数据插入/更新失败!);return false;}LOG_INFO(消息数据新增/更新成功!);return true;}bool remove(const std::string mid){bool ret ESRemove(_es_client, message).remove(mid);if (ret false){LOG_ERROR(消息数据删除失败!);return false;}LOG_INFO(消息数据删除成功!);return true;}std::vectorbite_im::Message search(const std::string key, const std::string ssid){std::vectorbite_im::Message res;Json::Value json_user ESSearch(_es_client, message).append_must_term(chat_session_id.keyword, ssid).append_must_match(content, key).search();if (json_user.isArray() false){LOG_ERROR(用户搜索结果为空或者结果不是数组类型);return res;}int sz json_user.size();LOG_DEBUG(检索结果条目数量{}, sz);for (int i 0; i sz; i){bite_im::Message message;message.user_id(json_user[i][_source][user_id].asString());message.message_id(json_user[i][_source][message_id].asString());boost::posix_time::ptime ctime(boost::posix_time::from_time_t(json_user[i][_source][create_time].asInt64()));message.create_time(ctime);message.session_id(json_user[i][_source][chat_session_id].asString());message.content(json_user[i][_source][content].asString());res.push_back(message);}return res;}private:std::shared_ptrelasticlient::Client _es_client;};
}9. cpp-httplib的安装与使用
9.1 cpp-httplib的介绍
1C HTTP 库cpp-httplib是一个轻量级的 C HTTP 客户端/服务器库它提供了简单的 API 来创建 HTTP 服务器和客户端支持同步和异步操作。以下是一些关于cpp-httplib 的主要特点
轻量级cpp-httplib 的设计目标是简单和轻量只有一个头文件包含即可不依赖于任何外部库。跨平台它支持多种操作系统包括 Windows、Linux 和 macOS。同步和异步操作库提供了同步和异步两种操作方式允许开发者根据需要选择。支持 HTTP/1.1它实现了 HTTP/1.1 协议包括持久连接和管道化。Multipart form-data支持发送和接收 multipart/form-data 类型的请求这对于文件上传非常有用。SSL/TLS 支持通过使用 OpenSSL 或 mbedTLS 库cpp-httplib 支持 HTTPS 和 WSS。简单易用API 设计简洁易于学习和使用。性能尽管是轻量级库但性能表现良好适合多种应用场景。社区活跃cpp-httplib 有一个活跃的社区不断有新的功能和改进被加入。
9.2 cpp-httplib的安装及接口介绍
1cpp-httplib安装
C
[xiaomakeralibaba ~]$ git clone https://github.com/yhirose/cpp-httplib.git而且我们只需要cpp-httplib目录下的httplib.h头文件就可以将此头文件移动其它目录下即可将cpp-httplib目录删除。
2类与接口的介绍
namespace httplib
{struct Request{std::string method;std::string path;Headers headers;std::string body;Params params;}; struct Response{std::string version;int status -1;std::string reason;Headers headers;std::string body;void set_content(const std::string s,const std::string content_type);void set_header(const std::string key,const std::string val);};class Server{using Handler std::functionvoid(const Request , Response);Server Get(const std::string pattern, Handler handler);Server Post(const std::string pattern, Handler handler);Server Put(const std::string pattern, Handler handler);Server Delete(const std::string pattern, Handler handler);bool listen(const std::string host, int port);};class Client{explicit Client(const std::string host, int port);Result Get(const std::string path, const Headers headers);Result Post(const std::string path, const std::string body,const std::string content_type);Result Put(const std::string path, const std::string body,const std::string content_type);Result Delete(const std::string path, const std::string body,const std::string content_type);};
}9.3 cpp-httplib的使用
1main.cpp
#include iostream
#include ../../third/include/httplib.hint main()
{// 1. 实例化服务器对象httplib::Server server;// 2. 注册回调函数 void(const httplib::Request , httplib::Response )server.Get(/hi, [](const httplib::Request req, httplib::Response rsp){std::cout req.method std::endl;std::cout req.path std::endl;for(auto it : req.headers) {std::cout it.first : it.second std::endl;}std::string body htmlbodyh1Hello World/h1/body/html;rsp.set_content(body, text/html);rsp.status 200;});// 3. 启动服务器server.listen(0.0.0.0, 8080);return 0;
}2运行结果 3服务端结果 10. websocketpp的安装与使用
10.1 websocketpp的介绍和原理
1WebSocketpp 是一个跨平台的开源BSD 许可证头部专用 C库它实现了RFC6455WebSocket 协议和 RFC7692WebSocketCompression Extensions。它允许将 WebSocket 客户端和服务器功能集成到 C程序中。在最常见的配置中全功能网络 I/O 由 Asio 网络库提供。WebSocketpp 的主要特性包括
事件驱动的接口。支持 HTTP/HTTPS、WS/WSS、IPv6。灵活的依赖管理 — Boost 库/C11 标准库。可移植性Posix/Windows、32/64bit、Intel/ARM。线程安全。
2WebSocketpp 同时支持 HTTP 和 Websocket 两种网络协议 比较适用于我们本次的项目 所以我们选用该库作为项目的依赖库用来搭建 HTTP 和WebSocket 服务器。下面是该项目的一些常用网站
githubhttps://github.com/zaphoyd/websocketpp用户手册 http://docs.websocketpp.org/官网http://www.zaphoyd.com/websocketpp
3Websocket 协议介绍
WebSocket 是从 HTML5 开始支持的一种网页端和服务端保持长连接的 消息推送机制。传统的 web 程序都是属于 “一问一答” 的形式即客户端给服务器发送了一个HTTP 请求服务器给客户端返回一个 HTTP 响应。这种情况下服务器是属于被动的一方如果客户端不主动发起请求服务器就无法主动给客户端响应像网页即时聊天或者我们做的五子棋游戏这样的程序都是非常依赖 “消息推送” 的即需要服务器主动推动消息到客户端。如果只是使用原生的 HTTP 协议要想实现消息推送一般需要通过 “轮询” 的方式实现 而轮询的成本比较高并且也不能及时的获取到消息的响应。基于上述两个问题 就产生了 WebSocket 协议。WebSocket 更接近于 TCP 这种级别的通信方式一旦连接建立完成客户端或者服务器都可以主动的向对方发送数据。
4原理解析
WebSocket 协议本质上是一个基于 TCP 的协议。为了建立一个 WebSocket 连接客户端浏览器首先要向服务器发起一个 HTTP 请求这个请求和通常的 HTTP 请求不同包含了一些附加头信息通过这个附加头信息完成握手过程并升级协议的过程。 具体协议升级的过程如下 报文格式 报文字段比较多我们重点关注这几个字段
FINWebSocket 传输数据以消息为概念单位一个消息有可能由一个或多个帧组成FIN 字段为 1 表示末尾帧。RSV1~3保留字段只在扩展时使用若未启用扩展则应置 1若收到不全为 0的数据帧且未协商扩展则立即终止连接。opcode: 标志当前数据帧的类型。 0x0: 表示这是个延续帧当 opcode 为 0 表示本次数据传输采用了数据分片当前收到的帧为其中一个分片0x1: 表示这是文本帧。0x2: 表示这是二进制帧。0x3-0x7: 保留暂未使用。0x8: 表示连接断开。0x9: 表示 ping 帧。0xa: 表示 pong 帧。0xb-0xf: 保留暂未使用。 mask表示 Payload 数据是否被编码若为 1 则必有 Mask-Key用于解码Payload 数据。仅客户端发送给服务端的消息需要设置。Payload length数据载荷的长度单位是字节 有可能为 7 位、716 位、764位。假设 Payload length x x 为 0~126数据的长度为 x 字节。x 为 126后续 2 个字节代表一个 16 位的无符号整数该无符号整数的值为数据的长度。x 为 127后续 8 个字节代表一个 64 位的无符号整数最高位为 0该无符号整数的值为数据的长度。 Mask-Key当 mask 为 1 时存在长度为 4 字节解码规则 DECODED[i] ENCODED[i] ^ MASK[i % 4]。Payload data: 报文携带的载荷数据。
10.2 websocketpp的使用
1main.cpp
#include websocketpp/config/asio_no_tls.hpp
#include websocketpp/server.hpp
#include iostream// 定义server类型
typedef websocketpp::serverwebsocketpp::config::asio server_t;void onOpen(websocketpp::connection_hdl hdl)
{std::cout websocket长连接建立成功\n;
}void onClose(websocketpp::connection_hdl hdl)
{std::cout websocket长连接断开\n;
}void onMessage(server_t *server, websocketpp::connection_hdl hdl, server_t::message_ptr msg)
{//1. 获取有效消息载荷数据进行业务处理std::string body msg-get_payload();std::cout 收到消息 body std::endl;//2. 对客户端进行响应//获取通信连接auto conn server-get_con_from_hdl(hdl);//发送数据conn-send(body -Hello!, websocketpp::frame::opcode::value::text);
}int main()
{// 1. 实例化服务器对象server_t server;// 2. 初始化日志输出 --- 关闭日志输出server.set_access_channels(websocketpp::log::alevel::none);// 3. 初始化asio框架server.init_asio();// 4. 设置消息处理/连接握手成功/连接关闭回调函数server.set_open_handler(onOpen);server.set_close_handler(onClose);auto msg_hadler std::bind(onMessage, server, std::placeholders::_1, std::placeholders::_2);server.set_message_handler(msg_hadler);// 5. 启用地址重用server.set_reuse_addr(true);// 5. 设置监听端口server.listen(9090);// 6. 开始监听server.start_accept();// 7. 启动服务器server.run();return 0;
}2运行结果 3服务端结果 11. redis的安装与使用
11.1 redis的介绍
RedisRemote Dictionary Server是一个开源的高性能键值对key-value数据库。它通常用作数据结构服务器因为除了基本的键值存储功能外Redis 还支持多种类型的数据结构如字符串strings、哈希hashes、列表lists、集合sets、有序集合sorted sets以及范围查询、位图、超日志和地理空间索引等。以下是 Redis 的一些主要特性
内存中数据库Redis 将所有数据存储在内存中这使得读写速度非常快。持久化尽管 Redis 是内存数据库但它提供了持久化选项可以将内存中的数据保存到磁盘上以防系统故障导致数据丢失。支持多种数据结构Redis 不仅支持基本的键值对还支持列表、集合、有序集合等复杂的数据结构。原子操作Redis 支持原子操作这意味着多个操作可以作为一个单独的原子步骤执行这对于并发控制非常重要。发布/订阅功能Redis 支持发布订阅模式允许多个客户端订阅消息当消息发布时所有订阅者都会收到消息。高可用性通过 Redis 哨兵Sentinel和 Redis 集群Redis 可以提供高可用性和自动故障转移。复制Redis 支持主从复制可以提高数据的可用性和读写性能。事务Redis 提供了事务功能可以保证一系列操作的原子性执行。Lua 脚本Redis 支持使用 Lua 脚本进行复杂的数据处理可以在服务器端执行复杂的逻辑。客户端库Redis 拥有丰富的客户端库支持多种编程语言如 Python、Ruby、Java、C# 等。性能监控Redis 提供了多种监控工具和命令可以帮助开发者监控和优化性能。易于使用Redis 有一个简单的配置文件和命令行界面使得设置和使用变得容易。Redis 广泛用于缓存、会话存储、消息队列、排行榜、实时分析等领域。由于其高性能和灵活性Redis 成为了现代应用程序中非常流行的数据存储解决方案之一。
11.2 redis的安装
1使用 apt 安装
Shell
apt install redis -y2支持远程连接
修改 /etc/redis/redis.conf 修改 bind 127.0.0.1 为 bind 0.0.0.0修改 protected-mode yes 改为 protected-mode no
Plain Text
# By default, if no bind configuration directive is specified, Redis listens
# for connections from all the network interfaces available on the server.
# It is possible to listen to just one or multiple selected interfaces using
# the bind configuration directive, followed by one or more IP addresses.
#
# Examples:
#
# bind 192.168.1.100 10.0.0.1
# bind 127.0.0.1 ::1
#
# ~~~ WARNING ~~~ If the computer running Redis is directly exposed to the
# internet, binding to all the interfaces is dangerous and will expose the
# instance to everybody on the internet. So by default we uncomment the
# following bind directive, that will force Redis to listen only into
# the IPv4 loopback interface address (this means Redis will be able to
# accept connections only from clients running into the same computer it
# is running).
#
# IF YOU ARE SURE YOU WANT YOUR INSTANCE TO LISTEN TO ALL THE INTERFACES
# JUST COMMENT THE FOLLOWING LINE.
#
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# bind 127.0.0.1 # 注释掉这行
bind 0.0.0.0 # 添加这行
protected-mode no # 把 yes 改成 no3Redis的相关服务
启动Redis服务
Plain Text
service redis-server start停止Redis服务
Plain Text
service redis-server stop重启Redis服务
Plain Text
service redis-server restart4安装 hiredis
C 操作 redis 的库有很多。本项目使用 redis-plus-plus这个库的功能强大使用简单。Github 地址https://github.com/sewenew/redis-plus-plusredis-plus-plus是基于 hiredis 实现的。hiredis 是一个 C 语言实现的 redis 客户端。因此需要先安装 hiredis直接使用包管理器安装即可。
Plain Text
apt install libhiredis-dev5安装redis-plus-plus
Bash
# 下载 redis-plus-plus 源码
git clone https://github.com/sewenew/redis-plus-plus.git
# 编译/安装 redis-plus-plus
# 进入redis-plus-plus目录
cd redis-plus-plus
# 创建build并进入
mkdir build cd build
# cmake进行构建
cmake -DCMAKE_INSTALL_PREFIX/usr ..
# 编译并安装
make sudo make install构建成功后, 会在 /usr/include/ 中多出 sw 目录并且内部包含 redis-plus-plus的一系列头文件。
11.3 redis的部分接口介绍
1redis 本身支持很多数据类型的键值对但是在聊天室项目中只涉及到了字符串键值对的操作因此这里主要介绍字符串键值对的基础操作。
namespace sw
{namespace redis{struct ConnectionOptions{std::string host;int port 6379;std::string path;std::string user default;std::string password;int db 0; // 默认 0 号库bool keep_alive false;};struct ConnectionPoolOptions{std::size_t size 1; // 最大连接数量};class Redis{// uri e.g tcp://127.0.0.1:6379explicit Redis(const std::string uri);explicit Redis(const ConnectionOptions connection_opts,const ConnectionPoolOptions pool_opts {});// 删除当前库中所有数据void flushdb(bool async false);// 删除指定键值对long long del(const StringView key);// 判断指定键值对是否存在long long exists(const StringView key);// 获取一个 string 键值对OptionalString get(const StringView key);// 存放一个 string 键值对且设置过期时间-毫秒bool set(const StringView key,const StringView val,const std::chrono::milliseconds ttl std::chrono::milliseconds(0), // 0 表示不设置超时UpdateType type UpdateType::ALWAYS);void setex(const StringView key,long long ttl,const StringView val);// 向一个列表中尾插/头插 string 键值对long long rpush(const StringView key, const StringView val);long long lpush(const StringView key, const StringView val);long long rpush(const StringView key, Input first, Input last);// std::vectorstd::string elements;// redis.lrange(list, 0, -1, std::back_inserter(elements));void lrange(const StringView key, long long start, long long stop, Output output);};}
}11.4 redis的使用
1这里只进行字符串键值对的增删改查操作以及数据的生命周期设置
#include iostream
#include sw/redis/redis.h
#include gflags/gflags.h
#include threadDEFINE_string(ip, 127.0.0.1, 这是服务器的IP地址格式127.0.0.1);
DEFINE_int32(port, 6379, 这是服务器的端口, 格式: 8080);
DEFINE_int32(db, 0, 库的编号默认0号);
DEFINE_bool(keep_alive, true, 是否进行长连接保活);void print(sw::redis::Redis client)
{auto user1 client.get(会话ID1);if (user1) std::cout *user1 std::endl;auto user2 client.get(会话ID2);if (user2) std::cout *user2 std::endl;auto user3 client.get(会话ID3);if (user3) std::cout *user3 std::endl;auto user4 client.get(会话ID4);if (user4) std::cout *user4 std::endl;auto user5 client.get(会话ID5);if (user5) std::cout *user5 std::endl;
}void add_string(sw::redis::Redis client)
{client.set(会话ID1, 用户ID1);client.set(会话ID2, 用户ID2);client.set(会话ID3, 用户ID3);client.set(会话ID4, 用户ID4);client.set(会话ID5, 用户ID5);client.del(会话ID3);client.set(会话ID5, 用户ID555); //数据已存在则进行修改不存在则新增print(client);
}void expired_test(sw::redis::Redis client)
{//这次的新增数据其实已经有了因此本次是修改//不仅仅修改了val而且还给键值对新增了过期时间client.set(会话ID1, 用户ID1111, std::chrono::milliseconds(1000));print(client);std::cout ------------休眠2s-----------\n;std::this_thread::sleep_for(std::chrono::seconds(2));print(client);
}void list_test(sw::redis::Redis client)
{client.rpush(群聊1, 成员1);client.rpush(群聊1, 成员2);client.rpush(群聊1, 成员3);client.rpush(群聊1, 成员4);client.rpush(群聊1, 成员5);std::vectorstd::string users;client.lrange(群聊1, 0, -1, std::back_inserter(users));for(auto user : users) {std::cout user std::endl;}
}int main(int argc, char *argv[])
{google::ParseCommandLineFlags(argc, argv, true);//功能接口演示中//1. 构造连接选项实例化Redis对象连接服务器sw::redis::ConnectionOptions opts;opts.host FLAGS_ip;opts.port FLAGS_port;opts.db FLAGS_db;opts.keep_alive FLAGS_keep_alive;sw::redis::Redis client(opts);//2. 添加字符串键值对删除字符串键值对获取字符串键值对add_string(client);//3. 实践控制数据有效时间的操作expired_test(client);//4. 列表的操作主要实现数据的右插左获取std::cout --------------------------\n;list_test(client);return 0;
}2运行结果 3Makefile编译
main:main.cppg -stdc17 $^ -o $ -lhiredis -lredis -lgflags
.PHONY:clean
clean:rm -rf main12. ODB安装与使用
12.1 ODB的介绍
1ODB是一种面向对象的数据库管理系统它支持面向对象的编程语言允许用户通过面向对象的语言和工具来操作数据库。主要特性
面向对象ODB使用对象来存储和管理数据这些对象具有属性、方法和事件等特性使得数据的表示和管理更加直观和灵活。可扩展性ODB可以通过添加新的类和对象来实现系统的扩展以适应不断变化的需求。交互性ODB支持事件驱动机制使得应用程序可以与数据库进行实时交互。平台独立性ODB可以运行在多种平台上如Windows、Linux、Mac OS等具有广泛的平台兼容性。多语言支持ODB可以支持多种面向对象的编程语言如Java、C、Python等。
2主要作用
存储和管理数据ODB的主要作用是存储和管理面向对象数据提供数据访问、查询、更新等操作的功能。确保数据一致性ODB提供了事务管理和并发控制机制确保在多用户环境下数据的一致性和完整性。支持数据完整性ODB支持数据完整性的定义和检查如约束、触发器等以确保数据的准确性。保证数据持久性ODB将数据存储在内存和磁盘上即使在系统崩溃或电源中断的情况下也能保证数据的完整性。
3应用领域
物联网ODB可以用于存储物联网设备的实时数据如传感器数据、位置信息等实现智能决策和优化控制。大规模数据管理与传统的关系型数据库相比ODB采用面向对象的数据模型能够更好地适应复杂的数据结构和关系。在处理大规模数据时ODB可以提供更高的性能和可伸缩性。嵌入式系统ODB通过提供高效的数据存储和访问方式帮助嵌入式系统实时地处理数据并减少对存储和计算资源的需求。软件开发对于基于对象的软件开发ODB提供了更加高效和灵活的数据管理方式简化了数据访问的过程减少了开发工作的重复性和复杂度。
12.2 ODB的安装
1安装 build2
因为 build2 安装时有可能会版本更新从 16 变成 17或从 17 变 18因此注意先从 build2 官网查看安装步骤https://build2.org/install.xhtml#unix
PowerShell
[xiaomakeralibaba build2]$ curl -sSfO https://download.build2.org/0.17.0/build2-install-0.17.0.sh
[xiaomakeralibaba build2]$ sh build2-install-0.17.0.sh安装中因为网络问题超时失败解决将超时时间设置的更长一些 Shell
[xiaomakeralibaba build2]$ sh build2-install-0.17.0.sh --timeout 18002安装odb-compiler
[xiaomakeralibaba ~]$ #注意这里的 gcc-11 需要根据你自己版本而定
[xiaomakeralibaba ~]$ sudo apt-get install gcc-11-plugin-dev
[xiaomakeralibaba ~]$ mkdir odb-build cd odb-build
[xiaomakeralibaba odb-build]$ bpkg create -d odb-gcc-N cc \config.cxxg \config.cc.coptions-O3 \config.bin.rpath/usr/lib \config.install.root/usr/ \config.install.sudosudo
[xiaomakeralibaba odb-build]$ cd odb-gcc-N
[xiaomakeralibaba odb-gcc-N]$ bpkg build odbhttps://pkg.cppget.org/1/beta
[xiaomakeralibaba odb-gcc-N]$ bpkg test odb # 输出以下内容
test odb-2.5.0-b.251/tests/testscript{testscript}
tested odb/2.5.0-b.251
[xiaomakeralibaba odb-gcc-N]$ bpkg install odb
[xiaomakeralibaba odb-gcc-N]$ odb --version
bash: /usr/bin/odb: No such file or directory
#如果报错了找不到 odb那就在执行下边的命令
[xiaomakeralibaba odb-gcc-N]$ sudo echo export PATH${PATH}:/usr/local/bin ~/.bashrc
[xiaomakeralibaba odb-gcc-N]$ export PATH${PATH}:/usr/local/bin
[xiaomakeralibaba odb-gcc-N]$ odb --version
ODB object-relational mapping (ORM) compiler for C 2.5.0-b.25
Copyright (c) 2009-2023 Code Synthesis Tools CC.
This is free software; see the source for copying conditions.
There is NO
warranty; not even for MERCHANTABILITY or FITNESS FOR A PARTICULAR
PURPOSE.3安装 ODB 运行时库
[xiaomakeralibaba odb-gcc-N]$ cd ..
[xiaomakeralibaba odb-build]$ bpkg create -d libodb-gcc-N cc \config.cxxg \config.cc.coptions-O3 \config.install.root/usr/ \config.install.sudosudo
[xiaomakeralibaba odb-build]$ cd libodb-gcc-N
[xiaomakeralibaba libodb-gcc-N]$ bpkg add https://pkg.cppget.org/1/beta
[xiaomakeralibaba libodb-gcc-N]$ bpkg fetch
[xiaomakeralibaba libodb-gcc-N]$ bpkg build libodb
[xiaomakeralibaba libodb-gcc-N]$ bpkg build libodb-mysql4安装 mysql 和客户端开发包
# 安装 mysql
sudo apt install mysql-server
sudo apt install -y libmysqlclient-dev# 配置 mysql
sudo vim /etc/my.cnf 或者 /etc/mysql/my.cnf # 有哪个修改哪个就行
# 添加以下内容
[client]
default-character-setutf8
[mysql]
default-character-setutf8
[mysqld]
character-set-serverutf8
bind-address 0.0.0.0# 修改 root 用户密码
devbite:~$ sudo cat /etc/mysql/debian.cnf
# Automatically generated for Debian scripts. DO NOT TOUCH!
[client]
host localhost
user debian-sys-maint
password UWcn9vY0NkrbJMRC
socket /var/run/mysqld/mysqld.sock
[mysql_upgrade]
host localhost
user debian-sys-maint
password UWcn9vY0NkrbJMRC
socket /var/run/mysqld/mysqld.sock
devbite:~$ sudo mysql -u debian-sys-maint -p
Enter password: #这里输入上边第 6 行看到的密码
mysql ALTER USER rootlocalhost IDENTIFIED WITH
mysql_native_password BY xxxxxx;
Query OK, 0 rows affected (0.01 sec)
mysql FLUSH PRIVILEGES;
Query OK, 0 rows affected (0.01 sec)
mysql quit# 重启 mysql并设置开机启动
sudo systemctl restart mysql
sudo systemctl enable mysql5安装 boost profile 库
[xiaomakeralibaba libodb-gcc-N]$ bpkg install --all --recursive6总体卸载
[xiaomakeralibaba libodb-gcc-N]$ bpkg uninstall --all --recursive7总体升级
[xiaomakeralibaba libodb-gcc-N]$ bpkg fetch
[xiaomakeralibaba libodb-gcc-N]$ bpkg status
[xiaomakeralibaba libodb-gcc-N]$ bpkg uninstall --all --recursive
[xiaomakeralibaba libodb-gcc-N]$ bpkg build --upgrade --recursive
[xiaomakeralibaba libodb-gcc-N]$ bpkg install --all --recursive13. RabbitMq的安装和使用
13.1 RabbitMq的介绍
1核心特性
可靠性RabbitMQ基于AMQP协议提供了持久化、可靠的消息传递机制。它确保消息能够在发送和接收之间进行可靠地传输即使在出现故障的情况下也能保证消息的安全性。RabbitMQ使用消息确认机制acknowledgment确保消息被消费者正确接收并处理。此外RabbitMQ还提供了镜像队列mirrored queues功能可以在集群中的多个节点上复制队列以确保消息的持久化和可靠性。灵活性RabbitMQ支持多种消息传递模式包括点对点、发布/订阅、请求/响应等。它允许开发人员根据应用程序的需求来选择合适的消息模式实现灵活的消息传递。可扩展性RabbitMQ通过使用可扩展的消息队列和集群功能能够轻松地处理大量的消息传递。它支持水平扩展可以在需要时添加更多的节点来处理更多的消息。此外RabbitMQ还支持集群和负载均衡确保在高并发或故障场景下服务的可用性。
2核心概念
队列QueueRabbitMQ中使用队列来存储和转发消息。每个队列都是一个消息缓冲区用于在消息生产者和消费者之间进行异步通信。队列具有先进先出FIFO的特性即先进入队列的消息会先被消费者取出。此外队列还可以设置其他属性如是否独占、是否自动删除等。交换机Exchange交换机是RabbitMQ中实现消息路由的核心组件。生产者将消息发送到交换机交换机根据路由键和绑定规则将消息分发到不同的队列中。RabbitMQ提供了多种类型的交换机包括直接交换机Direct、扇出交换机Fanout、主题交换机Topic和消息头交换机Headers等。路由键Routing Key路由键是交换机在路由消息时使用的关键字。生产者发送消息时可以指定路由键交换机根据路由键和绑定规则将消息分发到相应的队列中。绑定Binding绑定是将交换机和队列按照路由规则进行关联的过程。通过绑定交换机可以将消息路由到指定的队列中。绑定关系可以是一对一、一对多或多对多的。生产者Producer生产者是消息的发送方它将消息发送到RabbitMQ的交换机中。生产者可以指定消息的路由键和交换机的名称以控制消息的路由和分发。消费者Consumer消费者是消息的接收方它从RabbitMQ的队列中获取并处理消息。消费者可以监听一个或多个队列当队列中有新消息时消费者会将其取出并处理。处理完成后消费者可以选择发送确认消息给RabbitMQ服务器以表示消息已被成功处理。
13.2 RabbitMq的安装
1安装RabbitMq
sudo apt install rabbitmq-server# 启动服务
sudo systemctl start rabbitmq-server.service
# 查看服务状态
sudo systemctl status rabbitmq-server.service
# 安装完成的时候默认有个用户 guest 但是权限不够要创建一个
administrator 用户才可以做为远程登录和发表订阅消息#添加用户
sudo rabbitmqctl add_user root 123456#设置用户 tag
sudo rabbitmqctl set_user_tags root administrator#设置用户权限
sudo rabbitmqctl set_permissions -p / root . . .*
# RabbitMQ 自带了 web 管理界面,执行下面命令开启
sudo rabbitmq-plugins enable rabbitmq_management查看 rabbitmq-server 的状态 访问 webUI 界面默认端口为 15672 取消rabbitmq-server开机自启状态如下命令
sudo systemctl disable rabbitmq-server2安装 RabbitMQ 的 C客户端库
C 语言库https://github.com/alanxz/rabbitmq-cC库: https://github.com/CopernicaMarketingSoftware/AMQP-CPP/tree/master
我们这里使用 AMQP-CPP 库来编写客户端程序。
安装 AMQP-CPP
sudo apt install libev-dev # libev 网络库组件
git clone https://github.com/CopernicaMarketingSoftware/AMQP-CPP.git
cd AMQP-CPP/
make
make install13.3 AMQP-CPP 库的简单使用
1publish.cpp
#include ev.h
#include amqpcpp.h
#include amqpcpp/libev.h
#include openssl/ssl.h
#include openssl/opensslv.hint main()
{//1. 实例化底层网络通信框架的I/O事件监控句柄auto *loop EV_DEFAULT;//2. 实例化libEvHandler句柄 --- 将AMQP框架与事件监控关联起来AMQP::LibEvHandler handler(loop);//2.5. 实例化连接对象AMQP::Address address(amqp://root:123456127.0.0.1:5672/);AMQP::TcpConnection connection(handler, address);//3. 实例化信道对象AMQP::TcpChannel channel(connection);//4. 声明交换机AMQP::Deferred deferred channel.declareExchange(test-exchange, AMQP::ExchangeType::direct); deferred.onError([](const char *message){std::cout 声明交换机失败 message std::endl;exit(0);});deferred.onSuccess([](){std::cout test-exchange 交换机创建成功 std::endl;});//5. 声明队列AMQP::DeferredQueue deferredQueue channel.declareQueue(test-queue);deferredQueue.onError([](const char *message){std::cout 声明队列失败 message std::endl;exit(0);});deferredQueue.onSuccess([](){std::cout test-queue 队列创建成功 std::endl;});//6. 针对交换机和队列进行绑定auto binding_deferred channel.bindQueue(test-exchange, test-queue, test-queue-key);binding_deferred.onError([](const char *message) {std::cout test-exchange - test-queue 绑定失败 message std::endl;exit(0);});binding_deferred.onSuccess([](){std::cout test-exchange - test-queue 绑定成功 std::endl;});//7. 向交换机发布消息for(int i 0; i 10; i) {std::string msg Hello World- std::to_string(i);bool ret channel.publish(test-exchange, test-queue-key, msg);if(ret false) {std::cout publish 失败\n;}}//启动底层网络通信框架--开启I/Oev_run(loop, 0);return 0;
}2consume.cpp
#include ev.h
#include amqpcpp.h
#include amqpcpp/libev.h
#include openssl/ssl.h
#include openssl/opensslv.h// 消息回调处理函数的实现
void MessageCb(AMQP::TcpChannel *channel, const AMQP::Message message, uint64_t deliveryTag, bool redelivered)
{std::string msg;msg.assign(message.body(), message.bodySize());std::cout msg std::endl;channel-ack(deliveryTag); // 对消息进行确认
}int main()
{// 1. 实例化底层网络通信框架的I/O事件监控句柄auto *loop EV_DEFAULT;// 2. 实例化libEvHandler句柄 --- 将AMQP框架与事件监控关联起来AMQP::LibEvHandler handler(loop);// 2.5. 实例化连接对象AMQP::Address address(amqp://root:123456127.0.0.1:5672/);AMQP::TcpConnection connection(handler, address);// 3. 实例化信道对象AMQP::TcpChannel channel(connection);// 4. 声明交换机AMQP::Deferred deferred channel.declareExchange(test-exchange, AMQP::ExchangeType::direct);deferred.onError([](const char *message){std::cout 声明交换机失败 message std::endl;exit(0); });deferred.onSuccess([](){ std::cout test-exchange 交换机创建成功 std::endl; });// 5. 声明队列AMQP::DeferredQueue deferredQueue channel.declareQueue(test-queue);deferredQueue.onError([](const char *message){std::cout 声明队列失败 message std::endl;exit(0); });deferredQueue.onSuccess([]() { std::cout test-queue 队列创建成功 std::endl; });// 6. 针对交换机和队列进行绑定auto binding_deferred channel.bindQueue(test-exchange, test-queue, test-queue-key);binding_deferred.onError([](const char *message){std::cout test-exchange - test-queue 绑定失败 message std::endl;exit(0); });binding_deferred.onSuccess([](){ std::cout test-exchange - test-queue 绑定成功 std::endl; });// 7. 订阅队列消息 -- 设置消息处理回调函数auto callback std::bind(MessageCb, channel, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);channel.consume(test-queue, consume-tag) // 返回值 DeferredConsumer.onReceived(callback).onError([](const char *message){std::cout 订阅 test-queue 队列消息失败: message std::endl;exit(0); }); // 返回值是 AMQP::Deferred// 8. 启动底层网络通信框架--开启I/Oev_run(loop, 0);return 0;
}3运行结果 13.4 RabbitMQ的二次封装
1在项目中使用 rabbitmq 的时候我们目前只需要交换机与队列的直接交换实现一台主机将消息发布给另一台主机进行处理的功能因此在这里可以对 mq 的操作进行简单的封装使 mq 的操作在项目中更加简便。封装一个 MQClient
提供声明指定交换机与队列并进行绑定的功能。提供向指定交换机发布消息的功能。提供订阅指定队列消息并设置回调函数进行消息消费处理的功能。
2具体实现
#pragma once
#include ev.h
#include amqpcpp.h
#include amqpcpp/libev.h
#include openssl/ssl.h
#include openssl/opensslv.h
#include iostream
#include functional
#include logger.hppnamespace Test
{class MQClient{public:using MessageCallback std::functionvoid(const char*, size_t);using ptr std::shared_ptrMQClient;MQClient(const std::string user, const std::string passwd, const std::string host){_loop EV_DEFAULT;_handler std::make_uniqueAMQP::LibEvHandler(_loop);std::string url amqp:// user : passwd host /;AMQP::Address address(url);_connection std::make_uniqueAMQP::TcpConnection(_handler.get(), address);_channel std::make_uniqueAMQP::TcpChannel(_connection.get());_loop_thread std::thread([this]() {ev_run(_loop, 0);});}void declareComponents(const std::string exchange,const std::string queue,const std::string routing_key routing_key,AMQP::ExchangeType echange_type AMQP::ExchangeType::direct){AMQP::Deferred deferred _channel-declareExchange(exchange, echange_type);deferred.onError([](const char *message){LOG_ERROR(声明交换机失败{}, message);exit(0); });deferred.onSuccess([exchange](){ LOG_ERROR({} 交换机创建成功, exchange);});// 5. 声明队列AMQP::DeferredQueue deferredQueue _channel-declareQueue(queue);deferredQueue.onError([](const char *message){LOG_ERROR(声明队列失败{}, message);exit(0); });deferredQueue.onSuccess([queue](){ LOG_ERROR({} 队列创建成功, queue);});// 6. 针对交换机和队列进行绑定auto binding_deferred _channel-bindQueue(exchange, queue, routing_key);binding_deferred.onError([exchange, queue](const char *message){LOG_ERROR({} - {} 绑定失败, exchange, queue);exit(0); });binding_deferred.onSuccess([exchange, queue, routing_key](){ LOG_ERROR({} - {} - {} 绑定成功, exchange, queue, routing_key);});}bool publish(const std::string exchange, const std::string msg, const std::string routing_key routing_key){LOG_DEBUG(向交换机 {}-{} 发布消息, exchange, routing_key);bool ret _channel-publish(exchange, routing_key, msg);if(ret false) {LOG_ERROR({} 发布消息失败, exchange);return false;}return true;}void consume(const std::string queue, const MessageCallback cb) {LOG_DEBUG(开始订阅 {} 队列消息, queue);auto Consume _channel-consume(queue, consume-tag); // 返回值 DeferredConsumerConsume.onReceived([this, cb](const AMQP::Message message,uint64_t deliveryTag,bool redelivered){cb(message.body(), message.bodySize());_channel-ack(deliveryTag);});Consume.onError([queue](const char *message){LOG_ERROR(订阅 {} 队列消息失败: {}, queue, message);exit(0); });}~MQClient() {ev_async_init(_async_watcher, watcher_callback);ev_async_start(_loop, _async_watcher);ev_async_send(_loop, _async_watcher);_loop_thread.join();_loop nullptr;}private:static void watcher_callback(struct ev_loop *loop, ev_async *watcher, int32_t revents) {ev_break(loop, EVBREAK_ALL);}private:struct ev_async _async_watcher;struct ev_loop *_loop;std::unique_ptrAMQP::LibEvHandler _handler;std::unique_ptrAMQP::TcpConnection _connection;std::unique_ptrAMQP::TcpChannel _channel;std::thread _loop_thread;};
}14. 短信验证码SDK
15.1 短信验证码的安装
1首先注册百度云后点击如下操作 2点击免费开通 3扫码领取 4进行一系列的信息验证 5身份验证 6获取API发送短信 7获取密钥 8安装环境
# 安装依赖
sudo apt-get install libcurl4-openssl-dev libssl-dev uuid-dev libjsoncpp-dev
# 克隆代码
git clone https://github.com/aliyun/aliyun-openapi-cpp-sdk.git
# 进入目录
cd aliyun-openapi-cpp-sdk
# 一键编译安装
sudo sh easyinstall.sh core14.2 接口的使用
1test.cpp
#include cstdlib
#include iostream
#include alibabacloud/core/AlibabaCloud.h
#include alibabacloud/core/CommonRequest.h
#include alibabacloud/core/CommonClient.h
#include alibabacloud/core/CommonResponse.husing namespace std;
using namespace AlibabaCloud;int main(int argc, char **argv)
{AlibabaCloud::InitializeSdk();AlibabaCloud::ClientConfiguration configuration(cn-shanghai);// specify timeout when create client.configuration.setConnectTimeout(1500);configuration.setReadTimeout(4000);std::string access_key ***********;std::string access_key_secret ***********;AlibabaCloud::Credentials credential(access_key, access_key_secret);/* use STS Tokencredential.setSessionToken( getenv(ALIBABA_CLOUD_SECURITY_TOKEN) );*/AlibabaCloud::CommonClient client(credential, configuration);AlibabaCloud::CommonRequest request(AlibabaCloud::CommonRequest::RequestPattern::RpcPattern);request.setHttpMethod(AlibabaCloud::HttpRequest::Method::Post);request.setDomain(dysmsapi.aliyuncs.com);request.setVersion(2017-05-25);request.setQueryParameter(Action, SendSms);request.setQueryParameter(SignName, 微服务);request.setQueryParameter(TemplateCode, SMS_475320595);request.setQueryParameter(PhoneNumbers, ***********);request.setQueryParameter(TemplateParam, {\code\:\1234\});auto response client.commonResponse(request);if (response.isSuccess()){printf(request success.\n);printf(result: %s\n, response.result().payload().c_str());}else{printf(error: %s\n, response.error().errorMessage().c_str());printf(request id: %s\n, response.error().requestId().c_str());}AlibabaCloud::ShutdownSdk();return 0;
}2运行结果
14.3 短信接口的封装
1在项目中我们实际上并不关心如何调用阿里云的 SDK我只是期望有一个接口能够设定手机号和验证码之后调用就进行发送因此我们可以将 SDK 中使用到的接口和对象进行二次封装简化项目中的使用
将 SDK 中的 client 对象管理起来。向外提供一个向指定手机号发送指定短信的接口。
2具体实现
#pragma once
#include cstdlib
#include iostream
#include memory
#include alibabacloud/core/AlibabaCloud.h
#include alibabacloud/core/CommonRequest.h
#include alibabacloud/core/CommonClient.h
#include alibabacloud/core/CommonResponse.h
#include logger.hppnamespace Test
{class DMSClient {public:using ptr std::shared_ptrDMSClient;DMSClient(const std::string access_key_id,const std::string access_key_secret){AlibabaCloud::InitializeSdk();AlibabaCloud::ClientConfiguration configuration(cn-shanghai);configuration.setConnectTimeout(1500);configuration.setReadTimeout(4000);AlibabaCloud::Credentials credential(access_key_id, access_key_secret);_client std::make_uniqueAlibabaCloud::CommonClient(credential, configuration);}bool send(const std::string phone, const std::string code){AlibabaCloud::CommonRequest request(AlibabaCloud::CommonRequest::RequestPattern::RpcPattern);request.setHttpMethod(AlibabaCloud::HttpRequest::Method::Post);request.setDomain(dysmsapi.aliyuncs.com);request.setVersion(2017-05-25);request.setQueryParameter(Action, SendSms);request.setQueryParameter(SignName, 微服务);request.setQueryParameter(TemplateCode, SMS_475320595);request.setQueryParameter(PhoneNumbers, phone);std::string param_code {\code\:\ code \};request.setQueryParameter(TemplateParam, param_code);auto response _client-commonResponse(request);if (!response.isSuccess()){LOG_ERROR(短信验证码请求失败{}, response.error().errorMessage());return false;}return true;}~DMSClient() { AlibabaCloud::ShutdownSdk(); }private:std::unique_ptrAlibabaCloud::CommonClient _client;};
}15. 语音识别SDK
15.1 语音识别的安装
1首先注册百度云后点击如下操作 2下拉点击如下操作 3点击去领取 4点击领取 5领取完成创建应用 6创建完成 7下载SDK 8安装 sdk 所需依赖
Shell
# 安装 jsoncpp
sudo apt install libjsoncpp-dev
# 安装 libcurl
sudo apt install curl
# 安装 openssl
# ubuntu 22.04 默认安装了9将上述下载的SDK下载解压即可。 15.2 语音识别的简单使用
1代码实现
#include ../../third/include/aip-cpp-sdk/speech.hvoid asr(aip::Speech client)
{std::string file_content;aip::get_file_content(16k.pcm, file_content);Json::Value result client.recognize(file_content, pcm, 16000, aip::null);if(result[err_no].asInt() ! 0) {std::cout result[err_msg].asString() std::endl;return;}std::cout result[result][0].asString() std::endl;
}int main()
{// 设置APPID/AK/SKstd::string app_id 116303252;std::string api_key l6iGtYI7zvQNXrj2wHYAeudC;std::string secret_key zpYM0mBwuSajh6bpxMcUbhIjOEh30RKw;aip::Speech client(app_id, api_key, secret_key);asr(client);return 0;
}2运行结果 测试音频下载链接 https://platform.bj.bcebos.com/sdk/asr/asr_doc/doc_download_files/public.zipC sdk 使用文档https://ai.baidu.com/ai-doc/SPEECH/dlbxfrs5o常见问题 https://ai.baidu.com/ai-doc/SPEECH/wlciqajfo音频格式及转码https://ai.baidu.com/ai-doc/SPEECH/7k38lxpwf调用示例https://github.com/Baidu-AIP/sdk-demo
15.3 语音识别的封装
1封装结果
#pragma once
#include ../third/include/aip-cpp-sdk/speech.h
#include logger.hppnamespace Test
{class ASRClient{public:using ptr std::shared_ptrASRClient;ASRClient(const std::string app_id,const std::string api_key,const std::string secret_key):_client(app_id, api_key, secret_key){}std::string recognize(const std::string speech_data, std::string err){Json::Value result _client.recognize(speech_data, pcm, 16000, aip::null);if(result[err_no].asInt() ! 0) {LOG_ERROR(语音识别失败{}, result[err_msg].asString());err result[err_msg].asString();return std::string();}return result[result][0].asString();}private:aip::Speech _client;};
}16. cmake搭建过程和介绍精简版
1设置 cmake 所需版本号
CMake
cmake_minimum_required(VERSION 3.1.3)2设置项目工程名称
CMake
project(name)3普通变量定义及内容设置
set(variable content1 content2 ...)
set(CMAKE_CXX_STANDARD 17) //设置 C特性标准4列表定义添加数据
CMake
set(variable_name )
list(APPEND variable_name content)5预定义变量
预定义变量名称备注CMAKE_CXX_STANDARDc特性标准CMAKE_CURRENT_BINARY_DIRcmake 执行命令时所在的工作路径CMAKE_CURRENT_SOURCE_DIRCMakeLists.txt 所在目录CMAKE_INSTALL_PREFIX默认安装路径
6字符串内容替换
CMake
string(REPLACE .old .new dest src)7添加头文件路径
CMake
include_directories(path)8添加链接库 CMake target_link_libraries(target lib1 lib2 …) 9添加生成目标
CMake
add_executable(target srcfiles1 srcfile2 ...)10错误或提示打印
CMake
message(FATAL_ERROR/STATUS content)11查找源码文件
CMake
aux_source_directory( dir variable )12判断文件是否存在
CMake
if (NOT EXISTS file)
endif()13循环遍历
CMake
foreach(val vals)
endforeach()14执行外部指令
CMake
add_custom_command(PRE_BUILD //表示在所有其他步骤之前执行自定义命令COMMAND //要执行的指令名称ARGS //要执行的指令运行参数选项DEPENDS //指定命令的依赖项OUTPUT //指定要生成的目标名称COMMENT //执行命令时要打印的内容
)15添加嵌套子 cmake 目录
CMake
add_subdirectory(dir)16设置安装路径
CMake
INSTALL(TARGETS ${target_name} RUNTIME DESTINATION bin)详细的cmake介绍见博客未完成。
现在已经完成服务器的环境搭建现在进行各个子模块的服务实现见博客未完成。
客户端整体代码链接未完成。
