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。

相关文章

Java调用C++动态库超详细步骤讲解(附源码)

Java调用C++动态库超详细步骤讲解(附源码)

《Java调用C++动态库超详细步骤讲解(附源码)》C语言因其高效和接近硬件的特性,时常会被用在性能要求较高或者需要直接操作硬件的场合,：本文主要介绍Java调用C++动态库的相关资料,文中通过代... 目录一、直接调用C++库第一步：动态库生成（vs2017+qt5.12.10）第二步：Java调用C++
阅读更多...
Win11安装PostgreSQL数据库的两种方式详细步骤

Win11安装PostgreSQL数据库的两种方式详细步骤

《Win11安装PostgreSQL数据库的两种方式详细步骤》PostgreSQL是备受业界青睐的关系型数据库,尤其是在地理空间和移动领域,：本文主要介绍Win11安装PostgreSQL数据库的... 目录一、exe文件安装 （推荐）下载安装包1. 选择操作系统2. 跳转到EDB（PostgreSQL 的
阅读更多...
SpringKafka消息发布之KafkaTemplate与事务支持功能

SpringKafka消息发布之KafkaTemplate与事务支持功能

《SpringKafka消息发布之KafkaTemplate与事务支持功能》通过本文介绍的基本用法、序列化选项、事务支持、错误处理和性能优化技术,开发者可以构建高效可靠的Kafka消息发布系统,事务支... 目录引言一、KafkaTemplate基础二、消息序列化三、事务支持机制四、错误处理与重试五、性能优
阅读更多...
Python实现无痛修改第三方库源码的方法详解

Python实现无痛修改第三方库源码的方法详解

《Python实现无痛修改第三方库源码的方法详解》很多时候,我们下载的第三方库是不会有需求不满足的情况,但也有极少的情况,第三方库没有兼顾到需求,本文将介绍几个修改源码的操作,大家可以根据需求进行选择... 目录需求不符合模拟示例 1. 修改源文件2. 继承修改3. 猴子补丁4. 追踪局部变量需求不符合很
阅读更多...
Spring事务中@Transactional注解不生效的原因分析与解决

Spring事务中@Transactional注解不生效的原因分析与解决

《Spring事务中@Transactional注解不生效的原因分析与解决》在Spring框架中,@Transactional注解是管理数据库事务的核心方式,本文将深入分析事务自调用的底层原理,解释为... 目录1. 引言2. 事务自调用问题重现2.1 示例代码2.2 问题现象3. 为什么事务自调用会失效3
阅读更多...
Spring 中 BeanFactoryPostProcessor 的作用和示例源码分析

Spring 中 BeanFactoryPostProcessor 的作用和示例源码分析

《Spring中BeanFactoryPostProcessor的作用和示例源码分析》Spring的BeanFactoryPostProcessor是容器初始化的扩展接口,允许在Bean实例化前... 目录一、概览1. 核心定位2. 核心功能详解3. 关键特性二、Spring 内置的 BeanFactory
阅读更多...
Seata之分布式事务问题及解决方案

Seata之分布式事务问题及解决方案

《Seata之分布式事务问题及解决方案》：本文主要介绍Seata之分布式事务问题及解决方案,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录Seata–分布式事务解决方案简介同类产品对比环境搭建1.微服务2.SQL3.seata-server4.微服务配置事务模式1
阅读更多...
Ubuntu 22.04 服务器安装部署(nginx+postgresql)

Ubuntu 22.04 服务器安装部署(nginx+postgresql)

《Ubuntu22.04服务器安装部署(nginx+postgresql)》Ubuntu22.04LTS是迄今为止最好的Ubuntu版本之一,很多linux的应用服务器都是选择的这个版本... 目录是什么让 Ubuntu 22.04 LTS 变得安全？更新了安全包linux 内核改进一、部署环境二、安装系统
阅读更多...
MYSQL事务死锁问题排查及解决方案

MYSQL事务死锁问题排查及解决方案

《MYSQL事务死锁问题排查及解决方案》：本文主要介绍Java服务报错日志的情况,并通过一系列排查和优化措施,最终发现并解决了服务假死的问题,文中通过代码介绍的非常详细,需要的朋友可以参考下... 目录问题现象推测 1 - 客户端无错误重试配置推测 2 - 客户端超时时间过短推测 3 - mysql 版本问
阅读更多...
对postgresql日期和时间的比较

对postgresql日期和时间的比较

《对postgresql日期和时间的比较》文章介绍了在数据库中处理日期和时间类型时的一些注意事项,包括如何将字符串转换为日期或时间类型,以及在比较时自动转换的情况,作者建议在使用数据库时,根据具体情况... 目录PostgreSQL日期和时间比较DB里保存到时分秒，需要和年月日比较db里存储date或者ti
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2