1
背景及概述
为了应对这一情况，MySQL 8.0.20版本推出了基于zstd压缩算法的binlog压缩功能。
本文将从源码角度对binlog压缩功能进行浅析，帮助读者了解其使用方式。同时，通过测试结果展示该功能的效果和性能影响，体现binlog压缩功能为用户带来了较好的应用价值。
2
原理浅析
MySQL 8.0.20版本增加的binlog日志事务压缩功能，通过使用zstd算法对事务信息进行压缩，然后再写入binlog文件。压缩事务由Transaction_payload_log_event对象处理，该对象继承自Transaction_payload_event和Log_event。压缩后的事务内容保存在Transaction_payload_event中，核心成员如下所示。
class Transaction_payload_event : public Binary_log_event {protected://event所包含的数据信息const char *m_payload{nullptr};//压缩后的数据信息大小uint64_t m_payload_size{0};//压缩方式，定义见后，当前仅包含代表非压缩的NONE和ZSTD方式transaction::compression::type m_compression_type{transaction::compression::type::NONE};//压缩前的数据信息大小uint64_t m_uncompressed_size{0};......}//定义压缩方式enum type {/* ZSTD compression. */ZSTD = 0,/* No compression. */NONE = 255,};
压缩binlog的时机为写入缓存并提交事务时，通过调用this->compress(thd)来完成压缩动作，相关调用堆栈及函数如下所示。
调用堆栈：#0 in binlog_cache_data::finalize#1 in MYSQL_BIN_LOG::commit#2 in ha_commit_trans#3 in trans_commit#4 in mysql_execute_command#5 in dispatch_sql_command#6 in dispatch_command#7 in do_commandint binlog_cache_data::finalize(THD *thd, Log_event *end_event) {……if (int error = flush_pending_event(thd)) return error;if (int error = write_event(end_event)) return error;if (int error = this->compress(thd)) return error;……}
binlog压缩处理函数即为binlog_cache_data::compress，其流程图如下。
该函数首先会判断binlog_transaction_compression参数对应的变量，如果为true，则继续进行压缩处理。接着，函数会判断是否存在Incident events、非事务性更改等非压缩场景，若不存在，则进行压缩。通过这些场景校验后，函数获取压缩器，并根据需要申请更大的新内存空间。随后，基于cache（缓存）中压缩的内容生成Transaction_payload_log_event，并设置压缩前后的相关元信息，函数的主要流程如下所示。
bool binlog_cache_data::compress(THD *thd) {...// binlog_transaction_compression参数值判断if (thd->variables.binlog_trx_compression == false) goto end;...//校验是否存在非压缩场景if (has_incident()) goto end;if (thd->get_transaction()->has_modified_non_trans_table(Transaction_ctx::STMT) ||thd->get_transaction()->has_modified_non_trans_table(Transaction_ctx::SESSION))goto end;if (may_have_sbr_stmts()) goto end;...//获取压缩器，并检查是否需要申请更大内存空间if ((compressor = cctx.get_compressor(thd)) == nullptr) goto end;...Transaction_payload_log_event tple{thd};ctype = compressor->compression_type_code();compressor->open();stream.set_compressor(compressor);//将cache中的数据拷贝压缩if ((error = m_cache.copy_to(&stream))) goto compression_end;compressor->close();//基于压缩后的数据生成Transaction_payload_log_event的内容std::tie(buffer, size, std::ignore) = compressor->get_buffer();tple.set_payload((const char *)buffer);//设置压缩后数据的大小tple.set_payload_size(size);//设置压缩方式tple.set_compression_type(ctype);//设置压缩前的数据大小tple.set_uncompressed_size(uncompressed_size);//将event写入cache中error = write_event(&tple);//释放新申请buffer空间，压缩器中重置为旧buffer空间if (old_buffer) {std::tie(buffer, std::ignore, std::ignore) = compressor->get_buffer();compressor->set_buffer(old_buffer, old_capacity);free(buffer);}}
备机接收到Transaction_payload_log_event后，会将其写入RelayLog中，并由worker或者applier线程负责解析回放binlog中的各个log event（日志事件），核心处理逻辑如下所示。
int Transaction_payload_log_event::do_apply_event(Relay_log_info const *rli) {...//依次解压缩并处理事务中包含的各个log eventfor (auto ptr : it) {...if ((res = apply_payload_event(rli, (const uchar *)ptr))) break;...}...}bool Transaction_payload_log_event::apply_payload_event(Relay_log_info const *rli, const uchar *event_buf) {...if (binlog_event_deserialize(ptr, event_len, &fdle, true, &ev)) {res = true;goto end;}...if (is_mts_worker(thd)) {...//并行方式，由worker线程回放res = ev->do_apply_event_worker(worker);} else {...//非并行方式，由applier线程回放res = ev->apply_event(coord);}...}
void Transaction_payload_log_event::print(FILE *,PRINT_EVENT_INFO *info) const {...//打印压缩事务的起始格式if (!info->short_form) {std::ostringstream oss;oss << "\tTransaction_Payload\t" << to_string() << std::endl;oss << "# Start of compressed events!" << std::endl;print_header(head, info, false);my_b_printf(head, "%s", oss.str().c_str());}//遍历并依次打印事务中各个log eventbinary_log::transaction::compression::Iterable_buffer it(m_payload, m_payload_size, m_uncompressed_size, m_compression_type);for (auto ptr : it) {...if (binlog_event_deserialize((const unsigned char *)buffer, event_len,&glob_description_event, true, &ev))...process_event(info, ev, header()->log_pos, "", true);}...//打印压缩事务的结束格式if (!info->short_form) my_b_printf(head, "# End of compressed events!\n");...}
3
使用方式 &效果性能验证
binlog压缩相关控制变量具体描述见表1。
同时，可通过如下所示的show variables命令查看当前参数值，通过set命令进行参数值的更新。
mysql> show global variables like '%binlog_transaction_compression%';+-------------------------------------------+-------+| Variable_name | Value |+-------------------------------------------+-------+| binlog_transaction_compression | OFF || binlog_transaction_compression_level_zstd | 3 |+-------------------------------------------+-------+2 rows in set (0.02 sec)
//开启binlog压缩mysql> set global binlog_transaction_compression = on;Query OK, 0 rows affected (0.00 sec)//根据需要，调整zstd压缩等级mysql> set global binlog_transaction_compression_level_zstd = 7;Query OK, 0 rows affected (0.00 sec)
压缩比验证：
创建8U32G的实例，使用sysbench导入250张表，每张表包含25000行数据。开启32个压测线程，在混合读写模型（oltp_read_write）下多次开启压缩/非开启压缩，进行长达60s的压测，对比压测结束后binlog文件大小。
结果显示，压缩比（压缩前/压缩后）约为2，binlog压缩功能表现出较明显的压缩效果。
binlog_transaction_compression_level_zstd = 3//未开启压缩binlog文件大小，354.21MB//开启压缩后binlog文件大小，171.50MB
性能验证：
创建8U32G的实例，使用sysbench导入250张表，每张表包含25000行数据。开启32个压测线程，设置压缩级别binlog_transaction_compression_level_zstd为3，在混合读写模型（oltp_read_write）下多次开启压缩/非开启压缩，完成压测后取平均值数据。
4
总结
基于zstd压缩算法的binlog压缩功能，以较小的性能劣化，实现了存储开销减半，并节省了网络带宽。该功能在绝大多数实际使用场景中表现出较好的应用效果。
参考资料
[1] https://dev.mysql.com/doc/relnotes/mysql/8.0/en/news-8-0-20.html
[2] MySQL binlog 压缩功能对性能的影响，https://blog.csdn.net/weixin_39629075/article/details/113557347
[3] 在降本增效背景下谈MySQL压缩功能，https://zhuanlan.zhihu.com/p/629972777
END
华为云数据库 新用户特惠
Flexus云数据库RDS
6个月366元（原价1200）/
一年722.4元（原价2400）
RDS for MySQL单机版 1年729.12元
扫码抢购
活动时间：2025/4/15-2025/5/31
戳“阅读原文”，了解更多