Postgresql源码（122）Listen / Notify与事务的联动机制

2024-02-19 09:20
文章标签 源码 postgresql 122 事务 notify listen 联动机制

本文主要是介绍Postgresql源码（122）Listen / Notify与事务的联动机制，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

前言

Notify和Listen是Postgresql提供的不同会话间异步消息通信功能，例子：

LISTEN virtual;
NOTIFY virtual;
Asynchronous notification "virtual" received from server process with PID 8448.
NOTIFY virtual, 'This is the payload';
Asynchronous notification "virtual" with payload "This is the payload" received from server process with PID 8448.LISTEN foo;
SELECT pg_notify('fo' || 'o', 'pay' || 'load');
Asynchronous notification "foo" with payload "payload" received from server process with PID 14728.

功能使用PG的基础设施shm_mq + 信号机制拼装实现。

监听、通知的行为也兼容了数据库的事务的功能，事务回滚会删除监听、事务提交会触发通知。

本文对异步消息队列与事务的联动机制做一些分析。

事务提交触发

NOTIFY的功能必须等到事务提交才会触发：

postgres=# listen a1;
LISTEN
postgres=# begin;
BEGIN
postgres=*# notify a1;
NOTIFY
postgres=*# notify a1;
NOTIFY
postgres=*# commit;
COMMIT
Asynchronous notification "a1" received from server process with PID 17111.

流程比较简单，先从pendingActions中注册监听。再发信号触发异步notify。

void
AtCommit_Notify(void)
{...if (pendingActions != NULL){foreach(p, pendingActions->actions){ListenAction *actrec = (ListenAction *) lfirst(p);switch (actrec->action){case LISTEN_LISTEN:Exec_ListenCommit(actrec->channel);break;case LISTEN_UNLISTEN:Exec_UnlistenCommit(actrec->channel);break;case LISTEN_UNLISTEN_ALL:Exec_UnlistenAllCommit();break;}}}...if (pendingNotifies != NULL)SignalBackends();...
}

事务回滚清理

回滚后监听和通知都会清理：

postgres=# begin;
BEGIN
postgres=*# listen k123;
LISTEN
postgres=*# notify k123;
NOTIFY
postgres=*# abort;
ROLLBACK
postgres=# notify k123;
NOTIFY
postgres=#

事务回滚时执行清理动作：

void
AtAbort_Notify(void)
{if (amRegisteredListener && listenChannels == NIL)asyncQueueUnregister();pendingActions = NULL;pendingNotifies = NULL;
}

全部清理干净。

子事务提交不触发，交接给上一层事务

提交的子事务将notify交接给上一层事务。

postgres=# listen k000;
LISTEN
postgres=# begin;
BEGIN
postgres=*# savepoint sp1;
SAVEPOINT
postgres=*# savepoint sp2;
SAVEPOINT
postgres=*# notify k000;
NOTIFY
postgres=*# release sp2;
RELEASE
postgres=*# commit;
COMMIT
Asynchronous notification "k000" received from server process with PID 18902.

实现：

void
AtSubCommit_Notify(void)
{int			my_level = GetCurrentTransactionNestLevel();if (pendingActions != NULL &&pendingActions->nestingLevel >= my_level){if (pendingActions->upper == NULL ||pendingActions->upper->nestingLevel < my_level - 1){--pendingActions->nestingLevel;}else{ActionList *childPendingActions = pendingActions;pendingActions = pendingActions->upper;pendingActions->actions =list_concat(pendingActions->actions,childPendingActions->actions);pfree(childPendingActions);}}if (pendingNotifies != NULL &&pendingNotifies->nestingLevel >= my_level){Assert(pendingNotifies->nestingLevel == my_level);if (pendingNotifies->upper == NULL ||pendingNotifies->upper->nestingLevel < my_level - 1){--pendingNotifies->nestingLevel;}else{NotificationList *childPendingNotifies = pendingNotifies;ListCell   *l;pendingNotifies = pendingNotifies->upper;foreach(l, childPendingNotifies->events){Notification *childn = (Notification *) lfirst(l);if (!AsyncExistsPendingNotify(childn))AddEventToPendingNotifies(childn);}pfree(childPendingNotifies);}}
}

  • pendingActions：用于保存channel信息（LISTEN命令使用，Async_Listen中配置）

  • pendingNotifies：用于保存channel和payload信息（NOTIFY命令使用，Async_Notify中配置）

子事务提交时，notify并不会真正触发，也是和其他资源一样，将自己绑定的nestingLevel转移到上一层（注意这里是绑的nestingLevel不是xid比较合理）。

整体上会有两种情况：

情况一：子事务有间隔，走这个分支pendingActions->upper->nestingLevel < my_level - 1

begin;
savepoint sp1;
notify ch123;
savepoint sp2;
savepoint sp3;
notify ch789;
release sp3;

情况二：子事务无间隔，走else分支

begin;
savepoint sp1;
notify ch123;
savepoint sp2;
notify ch456;
savepoint sp3;
notify ch789;
release sp3;

pendingActions和pendingNotifies都有自己的upper指针形成链式结构，两种数据结构在子事务提交时的行为都是将信息转移到上一层中，区别是pendingActions直接挂到上一层的actions链表；pendingNotifies调用AddEventToPendingNotifies接口完成同样的动作。

子事务回滚不触发，清理属于子事务的pendings

回滚的子事务会删除监听。

postgres=# begin;
BEGIN
postgres=*# savepoint sp1;
SAVEPOINT
postgres=*# listen k123;
LISTEN
postgres=*# savepoint sp2;
SAVEPOINT
postgres=*# listen k000;
LISTEN
postgres=*# rollback to sp2;
ROLLBACK
postgres=*# notify k123;
NOTIFY
postgres=*# notify k000;
NOTIFY
postgres=*# commit;
COMMIT
Asynchronous notification "k123" received from server process with PID 18098.
postgres=#

void
AtSubAbort_Notify(void)
{int			my_level = GetCurrentTransactionNestLevel();...while (pendingActions != NULL &&pendingActions->nestingLevel >= my_level){ActionList *childPendingActions = pendingActions;pendingActions = pendingActions->upper;pfree(childPendingActions);}while (pendingNotifies != NULL &&pendingNotifies->nestingLevel >= my_level){NotificationList *childPendingNotifies = pendingNotifies;pendingNotifies = pendingNotifies->upper;pfree(childPendingNotifies);}
}

子事务回滚的话，全部是直接删除，不在做向上归属。

Listen/Notify的实现原理

(This content is a summary derived from code comments.)

  1. 同一台机器上有多个后端进程。多个后端进程监听多个通道。（在代码的其他部分，通道也被称为“conditions”。）
  2. 在基于磁盘的存储中有一个中央队列（目录 pg_notify/），通过 slru.c 模块将活跃使用的页面映射到共享内存中。所有的通知消息都被放置在队列中，稍后由监听的后端进程读取。没有集中的信息知道哪个后端进程监听哪个通道；每个后端进程都有自己感兴趣的通道列表。虽然只有一个队列，但通知被视为数据库本地的；这是通过在每个通知消息中包含发送者的数据库 OID 来实现的。监听的后端进程会忽略不匹配其数据库 OID 的消息。这一点很重要，因为它确保了发送者和接收者有相同的数据库编码，不会错误解释通道名称或有效载荷字符串中的非 ASCII 文本。由于通知不期望在数据库崩溃后存活，我们可以在任何重启时简单地清除 pg_notify 数据，并且不需要 WAL 支持或 fsync。
  3. 每个至少监听一个频道的后端进程都会通过将其进程ID注册到AsyncQueueControl的数组中来进行注册。然后，它会扫描中央队列中的所有传入通知，首先将通知的数据库OID与自身的数据库OID进行比较，然后将通知的频道与其监听的频道列表进行比较。如果匹配成功，它会将通知事件传递给前端。不匹配的事件将被简单地跳过。
  4. NOTIFY语句（Async_Notify例程）将通知存储在后端本地列表中，直到事务结束才会处理。来自同一事务的重复通知只发送一次通知。这样做是为了节省工作量，例如，当触发器在一个200万行的表上触发时，会为每一行的更改发送一个通知。如果应用程序需要接收每个已发送的单个通知，可以在额外的有效负载参数中轻松添加一些唯一的字符串。当事务准备提交时，PreCommit_Notify()将待处理的通知添加到队列的头部。队列的头指针始终指向下一个空闲位置，而位置只是一个页号和该页中的偏移量。这是在将事务标记为已提交之前完成的。如果在写入通知时遇到问题，我们仍然可以调用elog(ERROR, …)，事务将回滚。一旦我们将所有通知放入队列中，我们将返回到CommitTransaction()，然后执行实际的事务提交。在提交后，我们会再次被调用（AtCommit_Notify()）。在这里，我们对有效的监听状态（listenChannels）进行任何实际的更新。然后，我们向可能对我们的消息感兴趣的后端进程发送信号（包括我们自己的后端进程，如果正在监听）。这是通过SignalBackends()完成的，它会扫描监听后端进程的列表，并向每个监听后端进程发送一个PROCSIG_NOTIFY_INTERRUPT信号（我们不知道哪个后端进程在监听哪个频道，因此必须向它们全部发送信号）。但是，我们可以排除那些已经是最新状态的后端进程，并且还可以排除其他数据库中的后端进程（除非它们远远落后，应该被踢出以使其前进指针）。最后，在完全退出事务并即将进入空闲状态时，我们会扫描队列中需要发送到前端的消息（可能是来自其他后端的通知，或者是自己发送的通知）。这一步骤不是CommitTransaction序列的一部分，有两个重要原因。首先，我们在向前端发送数据时可能会出现错误，而在事务提交后进行清理时出现错误是非常糟糕的。其次，在某些情况下，一个过程在单个前端命令中发出多个提交，我们不希望在命令完成之前向前端发送通知；但是对于其他后端来说，每次提交后的通知应该立即发送出去。
  5. 收到PROCSIG_NOTIFY_INTERRUPT信号后，信号处理程序会设置进程的latch，如果该后端处于空闲状态（即等待前端命令并且不在事务块内），则会立即触发事件处理（参见ProcessClientReadInterrupt()）。否则，处理程序可能只设置一个标志，在下次进入空闲状态之前进行处理。入站通知处理包括读取自上次扫描以来到达的所有通知。我们读取每个通知，直到达到未提交事务的通知或者头指针的位置。
  6. 为了限制磁盘空间的消耗，需要推进尾指针，以便可以截断旧的页面。这是相对昂贵的操作（特别是，它需要一个独占锁），因此我们不希望经常执行。如果发送后端将队列头推进到新页面，则会执行此操作，但每QUEUE_CLEANUP_DELAY页只执行一次。

一个在相同频道上监听的应用程序将会收到自己发送的NOTIFY消息。如果这些消息对应用程序没有用处，可以通过将NOTIFY消息中的be_pid与应用程序自身后端的PID进行比较来忽略它们。（从FE/BE协议2.0开始，在启动期间，后端的PID会提供给前端。）上述设计确保通过忽略自我通知，不会错过来自其他后端的通知。用于通知管理的共享内存使用量（NUM_NOTIFY_BUFFERS）可以根据需要进行调整，而不会影响除性能之外的任何内容。可以同时排队的通知数据的最大量由max_notify_queue_pages GUC确定。

这篇关于Postgresql源码（122）Listen / Notify与事务的联动机制的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

Spring的基础事务注解@Transactional作用解读

Spring的基础事务注解@Transactional作用解读

《Spring的基础事务注解@Transactional作用解读》文章介绍了Spring框架中的事务管理,核心注解@Transactional用于声明事务,支持传播机制、隔离级别等配置,结合@Tran... 目录一、事务管理基础1.1 Spring事务的核心注解1.2 注解属性详解1.3 实现原理二、事务事
阅读更多...
MyBatis/MyBatis-Plus同事务循环调用存储过程获取主键重复问题分析及解决

MyBatis/MyBatis-Plus同事务循环调用存储过程获取主键重复问题分析及解决

《MyBatis/MyBatis-Plus同事务循环调用存储过程获取主键重复问题分析及解决》MyBatis默认开启一级缓存,同一事务中循环调用查询方法时会重复使用缓存数据,导致获取的序列主键值均为1,... 目录问题原因解决办法如果是存储过程总结问题myBATis有如下代码获取序列作为主键IdMappe
阅读更多...
使用Node.js和PostgreSQL构建数据库应用

使用Node.js和PostgreSQL构建数据库应用

《使用Node.js和PostgreSQL构建数据库应用》PostgreSQL是一个功能强大的开源关系型数据库,而Node.js是构建高效网络应用的理想平台,结合这两个技术,我们可以创建出色的数据驱动... 目录初始化项目与安装依赖建立数据库连接执行CRUD操作查询数据插入数据更新数据删除数据完整示例与最佳
阅读更多...
详解Spring中REQUIRED事务的回滚机制详解

详解Spring中REQUIRED事务的回滚机制详解

《详解Spring中REQUIRED事务的回滚机制详解》在Spring的事务管理中,REQUIRED是最常用也是默认的事务传播属性,本文就来详细的介绍一下Spring中REQUIRED事务的回滚机制,... 目录1. REQUIRED 的定义2. REQUIRED 下的回滚机制2.1 异常触发回滚2.2 回
阅读更多...
java 恺撒加密/解密实现原理(附带源码)

java 恺撒加密/解密实现原理(附带源码)

《java恺撒加密/解密实现原理(附带源码)》本文介绍Java实现恺撒加密与解密,通过固定位移量对字母进行循环替换,保留大小写及非字母字符,由于其实现简单、易于理解,恺撒加密常被用作学习加密算法的入... 目录Java 恺撒加密/解密实现1. 项目背景与介绍2. 相关知识2.1 恺撒加密算法原理2.2 Ja
阅读更多...
Nginx屏蔽服务器名称与版本信息方式(源码级修改)

Nginx屏蔽服务器名称与版本信息方式(源码级修改)

《Nginx屏蔽服务器名称与版本信息方式(源码级修改)》本文详解如何通过源码修改Nginx1.25.4,移除Server响应头中的服务类型和版本信息,以增强安全性,需重新配置、编译、安装,升级时需重复... 目录一、背景与目的二、适用版本三、操作步骤修改源码文件四、后续操作提示五、注意事项六、总结一、背景与
阅读更多...
Android实现图片浏览功能的示例详解(附带源码)

Android实现图片浏览功能的示例详解(附带源码)

《Android实现图片浏览功能的示例详解(附带源码)》在许多应用中,都需要展示图片并支持用户进行浏览,本文主要为大家介绍了如何通过Android实现图片浏览功能,感兴趣的小伙伴可以跟随小编一起学习一... 目录一、项目背景详细介绍二、项目需求详细介绍三、相关技术详细介绍四、实现思路详细介绍五、完整实现代码
阅读更多...
Spring 中的切面与事务结合使用完整示例

Spring 中的切面与事务结合使用完整示例

《Spring中的切面与事务结合使用完整示例》本文给大家介绍Spring中的切面与事务结合使用完整示例,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考... 目录 一、前置知识：Spring AOP 与 事务的关系 事务本质上就是一个“切面”二、核心组件三、完
阅读更多...
PostgreSQL简介及实战应用

PostgreSQL简介及实战应用

《PostgreSQL简介及实战应用》PostgreSQL是一种功能强大的开源关系型数据库管理系统,以其稳定性、高性能、扩展性和复杂查询能力在众多项目中得到广泛应用,本文将从基础概念讲起,逐步深入到高... 目录前言1. PostgreSQL基础1.1 PostgreSQL简介1.2 基础语法1.3 数据库
阅读更多...
Oracle迁移PostgreSQL隐式类型转换配置指南

Oracle迁移PostgreSQL隐式类型转换配置指南

《Oracle迁移PostgreSQL隐式类型转换配置指南》Oracle迁移PostgreSQL时因类型差异易引发错误,需通过显式/隐式类型转换、转换关系管理及冲突处理解决,并配合验证测试确保数据一致... 目录一、问题背景二、解决方案1. 显式类型转换2. 隐式转换配置三、维护操作1. 转换关系管理2.
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2