PostgreSQL 如何让心跳永远不死，支持半同步自动同步、异步升降级 - udf 心跳

背景

在心跳时，通过自定义UDF，实现心跳永远不被堵塞，并且支持更加当前的配置自动的进行同步、异步模式的升降级。实现半同步的功能。

UDF输入

1、优先模式（同步、异步）

2、同步等待超时时间

当优先为同步模式时，假设当前为同步配置，如果备库异常导致事务提交等待超过指定时间，则自动降级为异步。

当优先为异步模式时，假设当前为同步配置，自动降级为异步。

当优先为同步模式时，假设当前为异步配置，如果备库恢复到streaming模式，自动升级为同步。

使用技术点：

1、alter system

2、reload conf

3、cancle backend

4、dblink 异步调用

心跳UDF逻辑

判断当前实例状态  
  
  只读  
  
    退出  
  
  读写  
  
    判断当前事务模式   
  
      异步  
  
        发心跳  
  
        优先模式是什么  
  
          异步  
  
            退出  
  
          同步  
  
            判断是否需要升级  
  
              升级  
  
              退出  
  
  
      同步  
  
        消耗异步消息  
  
        发远程心跳  
  
        查询是否超时  
  
          降级  
  
        否则  
  
          消耗异步消息  
  
        优先模式是什么  
  
        异步  
  
          降级  
  
          退出  
  
        同步  
  
          退出  

设计

1、当前postgresql.conf配置

synchronous_commit='remote_write';  
synchronous_standby_names='*';  

表示同步模式。

2、心跳表设计

create table t_keepalive(id int primary key, ts timestamp, pos pg_lsn);  

3、心跳写入方法

insert into t_keepalive values (1,now(),pg_current_wal_lsn()) on conflict (id) do update set ts=excluded.ts,pos=excluded.pos returning id,ts,pos;  

4、创建一个建立连接函数，不报错

create or replace function conn(        
  name,   -- dblink名字        
  text    -- 连接串,URL        
) returns void as $$          
declare          
begin          
  perform dblink_connect($1, $2);         
  return;          
exception when others then          
  return;          
end;          
$$ language plpgsql strict;      

5、更加以上逻辑创建心跳UDF。

create or replace function keepalive (  
  prio_commit_mode text,    
  tmout interval  
) returns t_keepalive as $$  
declare  
  res1 int;  
  res2 timestamp;  
  res3 pg_lsn;  
  commit_mode text;  
  conn text := format('hostaddr=%s port=%s user=%s dbname=%s application_name=', '127.0.0.1', current_setting('port'), current_user, current_database());  
  conn_altersys text := format('hostaddr=%s port=%s user=%s dbname=%s', '127.0.0.1', current_setting('port'), current_user, current_database());  
  app_prefix_stat text := 'keepalive_dblink';  
begin  
  if prio_commit_mode not in ('sync','async') then  
    raise notice 'prio_commit_mode must be [sync|async]';  
    return null;  
  end if;  
  
  show synchronous_commit into commit_mode;  
  
  create extension IF NOT EXISTS dblink;  
  
  -- 判断当前实例状态  
  if pg_is_in_recovery()   
  
  -- 只读  
  then  
    raise notice 'Current instance in recovery mode.';  
    return null;  
      
  -- 读写  
  else  
  
    -- 判断当前事务模式   
    if commit_mode in ('local','off')  
  
    -- 异步  
    then  
  
      -- 发心跳  
      insert into t_keepalive values (1,now(),pg_current_wal_lsn()) on conflict (id) do update set ts=excluded.ts,pos=excluded.pos returning id,ts,pos into res1,res2,res3;  
  
      -- 优先模式是什么  
      if prio_commit_mode='async'   
  
      -- 异步  
      then  
  
        -- 退出  
        return row(res1,res2,res3)::t_keepalive;  
  
      -- 同步  
      else  
  
        -- 判断是否需要升级  
        perform 1 from pg_stat_replication where state='streaming' limit 1;  
        if found  
  
        -- 升级  
        then  
          perform dblink_exec(conn_altersys, 'alter system set synchronous_commit=remote_write', true);   
          perform pg_reload_conf();   
  
          -- 退出  
          return row(res1,res2,res3)::t_keepalive;  
        end if;  
  
        return row(res1,res2,res3)::t_keepalive;  
      end if;  
  
  
    -- 同步  
    else  
  
      -- 消耗异步消息  
      perform conn(app_prefix_stat,  conn||app_prefix_stat);     
      perform t from dblink_get_result(app_prefix_stat, false) as t(id int, ts timestamp, pos pg_lsn);  
  
      -- 发远程心跳  
      perform dblink_send_query(app_prefix_stat, $_$ insert into t_keepalive values (1,now(),pg_current_wal_lsn()) on conflict (id) do update set ts=excluded.ts,pos=excluded.pos returning id,ts,pos $_$);    
  
      -- 查询是否超时  
      <<ablock>>  
      loop  
        perform pg_sleep(0.2);  
  
        perform 1 from pg_stat_activity where application_name=app_prefix_stat and state='idle' limit 1;  
        -- 未超时  
        if found then  
          select id,ts,pos into res1,res2,res3 from dblink_get_result(app_prefix_stat, false) as t(id int, ts timestamp, pos pg_lsn);  
          raise notice 'no timeout';  
          exit ablock;  
        end if;  
            
        perform 1 from pg_stat_activity where wait_event='SyncRep' and application_name=app_prefix_stat and clock_timestamp()-query_start > tmout limit 1;  
        -- 降级  
        if found then  
          perform dblink_exec(conn_altersys, 'alter system set synchronous_commit=local', true);   
          perform pg_reload_conf();  
          perform pg_cancel_backend(pid) from pg_stat_activity where wait_event='SyncRep';  
          select id,ts,pos into res1,res2,res3 from dblink_get_result(app_prefix_stat, false) as t(id int, ts timestamp, pos pg_lsn);  
          raise notice 'timeout';  
          exit ablock;  
        end if;  
            
        perform pg_sleep(0.2);  
      end loop;  
  
      -- 优先模式是什么  
      if prio_commit_mode='async'   
  
      -- 异步  
      then  
        show synchronous_commit into commit_mode;  
        -- 降级  
        if commit_mode in ('on','remote_write','remote_apply')   
        then  
          perform dblink_exec(conn_altersys, 'alter system set synchronous_commit=local', true);   
          perform pg_reload_conf();  
          perform pg_cancel_backend(pid) from pg_stat_activity where wait_event='SyncRep';  
        end if;  
              
        -- 退出  
        return row(res1,res2,res3)::t_keepalive;  
  
      -- 同步  
      else  
  
        -- 退出  
        return row(res1,res2,res3)::t_keepalive;  
      end if;  
          
    end if;  
  
  end if;  
end;  
$$ language plpgsql strict;  

测试

1、当前为同步模式

postgres=# show synchronous_commit ;  
 synchronous_commit   
--------------------  
 remote_write  
(1 row)  

2、人为关闭从库，心跳自动将数据库改成异步模式，并通知所有等待中会话。

postgres=# select * from keepalive ('sync','5 second');  
NOTICE:  extension "dblink" already exists, skipping  
NOTICE:  timeout  
 id |             ts             |     pos       
----+----------------------------+-------------  
  1 | 2019-01-30 00:48:39.800829 | 23/9501D5F8  
(1 row)  
  
postgres=# show synchronous_commit ;  
 synchronous_commit   
--------------------  
 local  
(1 row)  

3、恢复从库，心跳自动将数据库升级为优先sync模式。

postgres=# select * from keepalive ('sync','5 second');  
NOTICE:  extension "dblink" already exists, skipping  
 id |             ts             |     pos       
----+----------------------------+-------------  
  1 | 2019-01-30 00:48:47.329119 | 23/9501D6E8  
(1 row)  
  
postgres=# select * from keepalive ('sync','5 second');  
NOTICE:  extension "dblink" already exists, skipping  
NOTICE:  no timeout  
 id |             ts             |     pos       
----+----------------------------+-------------  
  1 | 2019-01-30 00:49:11.991855 | 23/9501E0C8  
(1 row)  
  
postgres=# show synchronous_commit ;  
 synchronous_commit   
--------------------  
 remote_write  
(1 row)  

小结

在心跳时，通过自定义UDF，实现心跳永远不被堵塞，并且支持更加当前的配置自动的进行同步、异步模式的升降级。实现半同步的功能。

UDF输入

1、优先模式（同步、异步）

2、同步等待超时时间

当优先为同步模式时，假设当前为同步配置，如果备库异常导致事务提交等待超过指定时间，则自动降级为异步。

当优先为异步模式时，假设当前为同步配置，自动降级为异步。

当优先为同步模式时，假设当前为异步配置，如果备库恢复到streaming模式，自动升级为同步。

使用技术点：

1、alter system

2、reload conf

3、cancle backend

4、dblink 异步调用

使用心跳实现半同步，大大简化了整个同步、异步模式切换的流程。当然如果内核层面可以实现，配置几个参数，会更加完美。

参考

dblin 异步

《PostgreSQL 数据库心跳（SLA（RPO）指标的时间、WAL SIZE维度计算）》

《PostgreSQL 双节点流复制如何同时保证可用性、可靠性(rpo,rto) - （半同步，自动降级方法实践）》

PostgreSQL 许愿链接

您的愿望将传达给PG kernel hacker、数据库厂商等, 帮助提高数据库产品质量和功能, 说不定下一个PG版本就有您提出的功能点. 针对非常好的提议，奖励限量版PG文化衫、纪念品、贴纸、PG热门书籍等，奖品丰富，快来许愿。开不开森.

9.9元购买3个月阿里云RDS PostgreSQL实例

PostgreSQL 解决方案集合