C++项目实战——基于多设计模式下的同步异步日志系统-⑩-异步缓冲区类与异步工作器类设计

2023-10-17 09:04
文章标签 实战 c++ 设计 系统 设计模式 工作 日志 项目 异步 同步 器类 缓冲区

本文主要是介绍C++项目实战——基于多设计模式下的同步异步日志系统-⑩-异步缓冲区类与异步工作器类设计，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

文章目录

  • 专栏导读
  • 异步缓冲区设计思想
  • 异步缓冲区类设计
  • 异步工作器类设计
  • 异步日志器设计
  • 异步缓冲区类整理
  • 异步工作器类整理

专栏导读

🌸作者简介：花想云 ，在读本科生一枚，C/C++领域新星创作者，新星计划导师，阿里云专家博主，CSDN内容合伙人…致力于 C/C++、Linux 学习。

🌸专栏简介：本文收录于 C++项目——基于多设计模式下的同步与异步日志系统

🌸相关专栏推荐：C语言初阶系列、C语言进阶系列 、C++系列、数据结构与算法、Linux

在这里插入图片描述
为了避免因为写日志的过程阻塞，导致业务线程在写日志的时候影响其效率（例如由于网络原因导致日志写入阻塞，进而导致业务线程阻塞），因此我们需要设计一个异步日志器。

异步的思想就是不让业务线程进行日志的实际落地操作，而是将日志消息放到缓冲区（一块指定内存）当中，接下来有一个专门的异步线程，去针对缓冲区中的数据进行处理（实际落地操作）。

所以，异步日志器的实现思想：

  • 设计一个线程安全的缓冲区；
  • 创建一个异步工作线程，专门负责缓冲区中日志消息落地操作。

异步缓冲区设计思想

在任务池的设计中，有很多备选方案，比如队列、循环队列等，但是不管哪一种都会涉及到锁冲突的情况，因为在生产者与消费者模型中，任何两个角色之间具有互斥关系，因此每一次任务的添加与取出都有可能涉及锁的冲突。

所以我们采用双缓冲区的的设计思想，优势在于：

  • 避免了空间的频繁申请与释放，且尽可能的减少了生产者与消费则之间锁冲突的概率，提高了任务处理效率。

双缓冲区的设计思想是：采用两个缓冲区，一个用来进行任务写入（push pool），一个进行任务处理（pop pool）。当异步工作线程（消费者）将缓冲区中的数据全部处理完毕之后，然后交换两个缓冲区，重新对新的缓冲区中的任务进行处理，虽然同时多线程写入也会产生冲突，但是冲突并不会像每次只处理一条的时候频繁（减少了消费者与生产者之间的锁冲突），且不涉及到空间的频繁申请释放所带来的的消耗。

在这里插入图片描述

异步缓冲区类设计

类中包含的成员：

  • 一个存放字符串数据的缓冲区（使用vector进行空间管理）；
  • 当前写入数据位置的指针（指向可写区域的起始位置，避免数据的写入覆盖）；
  • 当前读取数据位置的指针（指向可读区域的起始位置，当读取指针与写入指针指向相同的位置表示数据读取完了）；

类中提供的操作：

  • 向缓冲区中写入数据；
  • 获取可读数据起始地址的接口；
  • 获取可读数据长度的接口；
  • 移动读写位置的接口；
  • 初始化缓冲区的操作（将读写位置初始化–在一个缓冲区所有数据处理完毕之后）；
  • 提供交换缓冲区的操作（交换空间地址，并不交换空间数据）。

注意，缓冲区中直接存放格式化后的日志消息字符串，而不是LogMsg对象，这样做有两个好处：

  • 减少了LogMsg对象频繁的构造的消耗；
  • 可以针对缓冲区中的日志消息，一次性进行IO操作，减少IO次数，提高效率。
#ifndef __M_BUFFER_H__
#define __M_BUFFER_H__
#include "util.hpp"
#include <vector>
#include <cassert>namespace LOG
{#define DEFAULT_BUFFER_SIZE (1 * 1024 * 1024)#define THRESHOLD_BUFFER_SIZE (8 * 1024 * 1024)#define INCREMENT_BUFFER_SIZE (1 * 1024 * 1024)class Buffer {public:Buffer() : _buffer(DEFAULT_BUFFER_SIZE), _writer_idx(0), _reader_idx(0) {}// 向缓冲区中写入数据void push(const char* data, size_t len){// 1.考虑空间不够则扩容ensureEnoughSize(len);// 2.将数据拷贝到缓冲区std::copy(data, data + len, &_buffer[_writer_idx]);// 3.将当前写入位置向后偏移moveWriter(len);}// 返回可读数据的起始地址const char* begin(){return &_buffer[_reader_idx];}// 返回可写数据的长度size_t writeAbleSize(){// 对于扩容思路并没有用， 仅针对固定大小缓冲区return (_buffer.size() - _writer_idx);}// 返回可读数据的长度size_t readAbleSize(){ return (_writer_idx - _reader_idx);}// 重置读写位置void reset(){_writer_idx = 0;_reader_idx = 0;}// 对buffer实现交换操作void swap(Buffer & buffer){_buffer.swap(buffer._buffer);std::swap(_reader_idx, buffer._reader_idx);std::swap(_writer_idx, buffer._writer_idx);}// 判断缓冲区是否为空bool empty(){return (_reader_idx == _writer_idx);}private:void ensureEnoughSize(size_t len){if(len < writeAbleSize()) return;size_t new_size = 0;if(_buffer.size() < THRESHOLD_BUFFER_SIZE){new_size = _buffer.size() * 2 + len; // 小于阈值则翻倍增长}else{new_size = _buffer.size() + INCREMENT_BUFFER_SIZE + len;}_buffer.resize(new_size);}// 对读指针进行向后偏移操作void moveReader(size_t len){assert(len <= readAbleSize());_reader_idx += len;}// 对写指针进行向后偏移操作void moveWriter(size_t len){assert(len + _writer_idx<= _buffer.size());_writer_idx += len;}private:std::vector<char> _buffer;size_t _reader_idx; // 当前可读数据的指针size_t _writer_idx; // 当前可写数据的指针};
}
#endif

异步工作器类设计

异步工作器的主要任务是，对缓冲区中的数据进行处理，若处理缓冲区中没有数据了则交换缓冲区。

异步工作器类管理的成员有：

  • 双缓冲区（生产，消费）；
  • 互斥锁：保证线程安全；
  • 条件变量-生产&消费：生产缓冲区中没有数据，处理完消费缓冲区数据后就休眠；
  • 回调函数：针对缓冲区中数据的处理接口——外界传入一个函数，告诉异步日志器该如何处理。

异步工作器类提供的操作有：

  • 停止异步工作器；
  • 添加数据到缓冲区；

私有操作：

  • 创建线程；
  • 线程入口函数：在线程入口函数中交换缓冲区，对消费缓冲区数据使用回调函数进行处理，处理完后再次交换；
#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__
#include "buffer.hpp"
#include <iostream>
#include <thread>
#include <mutex>
#include <atomic>
#include <condition_variable>
#include <functional>
#include <memory>namespace LOG
{using Functor = std::function<void(Buffer &)>;enum class AsyncType{ASYNC_SAFE, // 安全状态， 表示缓冲区满了则阻塞，避免资源耗尽的风险ASYNC_UNSAFE // 不考虑资源耗尽的问题};class AsyncLooper{public:using ptr = std::shared_ptr<AsyncLooper>;AsyncLooper(const Functor& cb, AsyncType looper_type = AsyncType::ASYNC_SAFE):_looper_type(looper_type),_stop(false),_call_back(cb),_thread(std::thread(&AsyncLooper::threadEntry, this)){}~AsyncLooper() { stop(); }void stop() {_stop = true; // 将退出标志设置为true_con_cond.notify_all(); // 唤醒所有工作线程_thread.join(); // 等待工作线程退出}void push(const char* data, size_t len){std::unique_lock<std::mutex> lock(_mutex);// 条件变量控制，若缓冲区剩余空间大小等于数据长度，则可以添加数据if(_looper_type == AsyncType::ASYNC_SAFE)_pro_cond.wait(lock, [&](){ return _pro_buf.writeAbleSize() >= len; });// 能够走下来说明条件满足，可以向缓冲区添加数据了_pro_buf.push(data, len);// 唤醒消费者对缓冲区中的数据进行处理_con_cond.notify_one();}private:// 线程入口函数 -- 对消费者缓冲区中的数据进行处理，处理完毕后，初始化缓冲区，交换缓冲区void threadEntry(){// 为互斥锁设置一个声明周期，当缓冲区交换完毕就解锁while(1){{// 1.判断生产缓冲区有没有数据，有则交换，无则阻塞std::unique_lock<std::mutex> lock(_mutex);if(_stop && _pro_buf.empty()) break;// 若当前是退出前被唤醒或者是有数据被唤醒，则返回真，继续向下运行，否则重新进入休眠_con_cond.wait(lock, [&](){ return _stop || !_pro_buf.empty(); });_con_buf.swap(_pro_buf);// 2.唤醒生产者if(_looper_type == AsyncType::ASYNC_SAFE)_pro_cond.notify_all();}// 3.被唤醒后，对消费者缓冲区进行数据处理_call_back(_con_buf);// 4.初始化消费者缓冲区_con_buf.reset();}}private:Functor _call_back;private:AsyncType _looper_type;std::atomic<bool> _stop;Buffer _pro_buf;Buffer _con_buf;std::mutex _mutex;std::condition_variable _pro_cond;std::condition_variable _con_cond;std::thread _thread; // 异步工作器对应的工作线程};
}
#endif

异步日志器设计

异步日志器继承自日志器类，并在同步日志器类上拓展了异步工作器。当我们需要异步输出日志的时候，需要创建异步日志器和消息处理器，调用异步日志器的log、debug、error、info、fatal等函数输出不同级别日志。

  • log函数为重写Logger类的函数，主要实现将日志日志数据加入异步缓冲区；
  • realLog函数主要由异步线程调用（是为异步工作器设置的回调函数），完成日志的实际落地操作。
class AsyncLogger : public Logger
{
public:AsyncLogger(const std::string &logger_name,LogLevel::value level,LOG::Formatter::ptr &formatter,std::vector<LogSink::ptr> &sinks,AsyncType looper_type): Logger(logger_name, level, formatter, sinks),_looper(std::make_shared<AsyncLooper>(std::bind(&AsyncLogger::realLog, this, std::placeholders::_1), looper_type)){}// 将数据写入缓冲区 void log(const char *data, size_t len){_looper->push(data, len);}// 设计一个实际落地函数void realLog(Buffer &buf){if (_sinks.empty())return;for (auto &sink : _sinks){sink->log(buf.begin(), buf.readAbleSize());}}private:AsyncLooper::ptr _looper; // 异步工作器
};

异步缓冲区类整理

#ifndef __M_BUFFER_H__
#define __M_BUFFER_H__
#include "util.hpp"
#include <vector>
#include <cassert>namespace LOG
{#define DEFAULT_BUFFER_SIZE (1 * 1024 * 1024)#define THRESHOLD_BUFFER_SIZE (8 * 1024 * 1024)#define INCREMENT_BUFFER_SIZE (1 * 1024 * 1024)class Buffer {public:Buffer() : _buffer(DEFAULT_BUFFER_SIZE), _writer_idx(0), _reader_idx(0) {}// 向缓冲区中写入数据void push(const char* data, size_t len){// 1.考虑空间不够则扩容ensureEnoughSize(len);// 2.将数据拷贝到缓冲区std::copy(data, data + len, &_buffer[_writer_idx]);// 3.将当前写入位置向后偏移moveWriter(len);}// 返回可读数据的起始地址const char* begin(){return &_buffer[_reader_idx];}// 返回可写数据的长度size_t writeAbleSize(){// 对于扩容思路并没有用， 仅针对固定大小缓冲区return (_buffer.size() - _writer_idx);}// 返回可读数据的长度size_t readAbleSize(){ return (_writer_idx - _reader_idx);}// 重置读写位置void reset(){_writer_idx = 0;_reader_idx = 0;}// 对buffer实现交换操作void swap(Buffer & buffer){_buffer.swap(buffer._buffer);std::swap(_reader_idx, buffer._reader_idx);std::swap(_writer_idx, buffer._writer_idx);}// 判断缓冲区是否为空bool empty(){return (_reader_idx == _writer_idx);}private:void ensureEnoughSize(size_t len){if(len < writeAbleSize()) return;size_t new_size = 0;if(_buffer.size() < THRESHOLD_BUFFER_SIZE){new_size = _buffer.size() * 2 + len; // 小于阈值则翻倍增长}else{new_size = _buffer.size() + INCREMENT_BUFFER_SIZE + len;}_buffer.resize(new_size);}// 对读指针进行向后偏移操作void moveReader(size_t len){assert(len <= readAbleSize());_reader_idx += len;}// 对写指针进行向后偏移操作void moveWriter(size_t len){assert(len + _writer_idx<= _buffer.size());_writer_idx += len;}private:std::vector<char> _buffer;size_t _reader_idx; // 当前可读数据的指针size_t _writer_idx; // 当前可写数据的指针};
}
#endif

异步工作器类整理

#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__
#include "buffer.hpp"
#include <iostream>
#include <thread>
#include <mutex>
#include <atomic>
#include <condition_variable>
#include <functional>
#include <memory>namespace LOG
{using Functor = std::function<void(Buffer &)>;enum class AsyncType{ASYNC_SAFE, // 安全状态， 表示缓冲区满了则阻塞，避免资源耗尽的风险ASYNC_UNSAFE // 不考虑资源耗尽的问题};class AsyncLooper{public:using ptr = std::shared_ptr<AsyncLooper>;AsyncLooper(const Functor& cb, AsyncType looper_type = AsyncType::ASYNC_SAFE):_looper_type(looper_type),_stop(false),_call_back(cb),_thread(std::thread(&AsyncLooper::threadEntry, this)){}~AsyncLooper() { stop(); }void stop() {_stop = true; // 将退出标志设置为true_con_cond.notify_all(); // 唤醒所有工作线程_thread.join(); // 等待工作线程退出}void push(const char* data, size_t len){std::unique_lock<std::mutex> lock(_mutex);// 条件变量控制，若缓冲区剩余空间大小等于数据长度，则可以添加数据if(_looper_type == AsyncType::ASYNC_SAFE)_pro_cond.wait(lock, [&](){ return _pro_buf.writeAbleSize() >= len; });// 能够走下来说明条件满足，可以向缓冲区添加数据了_pro_buf.push(data, len);// 唤醒消费者对缓冲区中的数据进行处理_con_cond.notify_one();}private:// 线程入口函数 -- 对消费者缓冲区中的数据进行处理，处理完毕后，初始化缓冲区，交换缓冲区void threadEntry(){// 为互斥锁设置一个生命周期，当缓冲区交换完毕就解锁while(1){{// 1.判断生产缓冲区有没有数据，有则交换，无则阻塞std::unique_lock<std::mutex> lock(_mutex);if(_stop && _pro_buf.empty()) break;// 若当前是退出前被唤醒或者是有数据被唤醒，则返回真，继续向下运行，否则重新进入休眠_con_cond.wait(lock, [&](){ return _stop || !_pro_buf.empty(); });_con_buf.swap(_pro_buf);// 2.唤醒生产者if(_looper_type == AsyncType::ASYNC_SAFE)_pro_cond.notify_all();}// 3.被唤醒后，对消费者缓冲区进行数据处理_call_back(_con_buf);// 4.初始化消费者缓冲区_con_buf.reset();}}private:Functor _call_back;private:AsyncType _looper_type; // 选择异步工作器工作模式（安全与非安全模式）std::atomic<bool> _stop; // 工作器退出标志Buffer _pro_buf; Buffer _con_buf; std::mutex _mutex;std::condition_variable _pro_cond;std::condition_variable _con_cond;std::thread _thread; // 异步工作器对应的工作线程};
}
#endif

在这里插入图片描述

这篇关于C++项目实战——基于多设计模式下的同步异步日志系统-⑩-异步缓冲区类与异步工作器类设计的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/224358。

相关文章

Linux线程同步/互斥过程详解

Linux线程同步/互斥过程详解

《Linux线程同步/互斥过程详解》文章讲解多线程并发访问导致竞态条件,需通过互斥锁、原子操作和条件变量实现线程安全与同步,分析死锁条件及避免方法,并介绍RAII封装技术提升资源管理效率... 目录01. 资源共享问题1.1 多线程并发访问1.2 临界区与临界资源1.3 锁的引入02. 多线程案例2.1 为
阅读更多...
精选20个好玩又实用的的Python实战项目(有图文代码)

精选20个好玩又实用的的Python实战项目(有图文代码)

《精选20个好玩又实用的的Python实战项目(有图文代码)》文章介绍了20个实用Python项目,涵盖游戏开发、工具应用、图像处理、机器学习等,使用Tkinter、PIL、OpenCV、Kivy等库... 目录① 猜字游戏② 闹钟③ 骰子模拟器④ 二维码⑤ 语言检测⑥ 加密和解密⑦ URL缩短⑧ 音乐播放
阅读更多...
Spring Boot集成/输出/日志级别控制/持久化开发实践

Spring Boot集成/输出/日志级别控制/持久化开发实践

《SpringBoot集成/输出/日志级别控制/持久化开发实践》SpringBoot默认集成Logback,支持灵活日志级别配置（INFO/DEBUG等）,输出包含时间戳、级别、类名等信息,并可通过... 目录一、日志概述1.1、Spring Boot日志简介1.2、日志框架与默认配置1.3、日志的核心作用
阅读更多...
Springboot项目启动失败提示找不到dao类的解决

Springboot项目启动失败提示找不到dao类的解决

《Springboot项目启动失败提示找不到dao类的解决》SpringBoot启动失败,因ProductServiceImpl未正确注入ProductDao,原因：Dao未注册为Bean,解决：在启... 目录错误描述原因解决方法总结***************************APPLICA编
阅读更多...
SQL Server跟踪自动统计信息更新实战指南

SQL Server跟踪自动统计信息更新实战指南

《SQLServer跟踪自动统计信息更新实战指南》本文详解SQLServer自动统计信息更新的跟踪方法,推荐使用扩展事件实时捕获更新操作及详细信息,同时结合系统视图快速检查统计信息状态,重点强调修... 目录SQL Server 如何跟踪自动统计信息更新：深入解析与实战指南 核心跟踪方法1️⃣ 利用系统目录
阅读更多...
java中pdf模版填充表单踩坑实战记录(itextPdf、openPdf、pdfbox)

java中pdf模版填充表单踩坑实战记录(itextPdf、openPdf、pdfbox)

《java中pdf模版填充表单踩坑实战记录(itextPdf、openPdf、pdfbox)》：本文主要介绍java中pdf模版填充表单踩坑的相关资料,OpenPDF、iText、PDFBox是三... 目录准备Pdf模版方法1：itextpdf7填充表单（1）加入依赖（2）代码（3）遇到的问题方法2：pd
阅读更多...
Mysql中设计数据表的过程解析

Mysql中设计数据表的过程解析

《Mysql中设计数据表的过程解析》数据库约束通过NOTNULL、UNIQUE、DEFAULT、主键和外键等规则保障数据完整性,自动校验数据,减少人工错误,提升数据一致性和业务逻辑严谨性,本文介绍My... 目录1.引言2.NOT NULL——制定某列不可以存储NULL值2.UNIQUE——保证某一列的每一
阅读更多...
C++11范围for初始化列表auto decltype详解

C++11范围for初始化列表auto decltype详解

《C++11范围for初始化列表autodecltype详解》C++11引入auto类型推导、decltype类型推断、统一列表初始化、范围for循环及智能指针,提升代码简洁性、类型安全与资源管理效... 目录C++11新特性1. 自动类型推导auto1.1 基本语法2. decltype3. 列表初始化3
阅读更多...
深度解析Nginx日志分析与499状态码问题解决

深度解析Nginx日志分析与499状态码问题解决

《深度解析Nginx日志分析与499状态码问题解决》在Web服务器运维和性能优化过程中,Nginx日志是排查问题的重要依据,本文将围绕Nginx日志分析、499状态码的成因、排查方法及解决方案展开讨论... 目录前言1. Nginx日志基础1.1 Nginx日志存放位置1.2 Nginx日志格式2. 499
阅读更多...
C++11右值引用与Lambda表达式的使用

C++11右值引用与Lambda表达式的使用

《C++11右值引用与Lambda表达式的使用》C++11引入右值引用,实现移动语义提升性能,支持资源转移与完美转发；同时引入Lambda表达式,简化匿名函数定义,通过捕获列表和参数列表灵活处理变量... 目录C++11新特性右值引用和移动语义左值 / 右值常见的左值和右值移动语义移动构造函数移动复制运算符
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2