前提条件：
我这边有一个三台pg数据库，1个master和2个standby，standby_s1是sync状态，standby_s2是potential状态，虽然配置的同步复制关系，但是standby_s1依然是可以有同步延时的情况，如果恰好这时master数据库挂了，而我又不想有数据，事务丢失，那我这时候不能完全确定是不是需要把主库切到standby_s1，需要查看两个standby最后一次事务哪个是最新的，那我该怎么来判断两个standby的最后状态呢？大神@德哥提供了一个很好的文章，解决了这个问题，那就是写一个lib文件（我是纯属copy，做一下记录，具体原理需要看德哥的博客）：
vi get_upstream_conninfo.c
#include "postgres.h"
#include "fmgr.h"
#include "access/xlog.h"
#include "replication/walreceiver.h"
#include "utils/elog.h"
#include "utils/builtins.h"
#include "utils/timestamp.h"
#include "funcapi.h"
#include "access/htup_details.h"
#include "catalog/pg_type.h"
#include "utils/pg_lsn.h"
#ifdef PG_MODULE_MAGIC
PG_MODULE_MAGIC;
#endif
PG_FUNCTION_INFO_V1(get_rcv_replication_stat);
Datum
get_rcv_replication_stat(PG_FUNCTION_ARGS)
{
Assert(PG_NARGS() == 0); // 表示没有输入参数
if (!RecoveryInProgress()) // 在数据库处于恢复状态下时运行,否则不允许
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("recovery is not in progress"),
errhint("This functions can only be executed during recovery.")));
/* use volatile pointer to prevent code rearrangement */
volatile WalRcvData *walrcv = WalRcv; // 共享内存中用于管理流复制的数据结构
TupleDesc tupdesc; // 创建一个行描述变量
Datum values[8]; // 创建一个存储值的Datum数组, 需要返回几个字段, 创建相应长度的数组
bool nulls[8]; // 创建一个数组, 表示对应的每个值是否为空
/* Initialise values and NULL flags arrays 初始化 */
MemSet(values, 0, sizeof(values));
MemSet(nulls, 0, sizeof(nulls));
/* Initialise attributes information in the tuple descriptor 定义字段类型和字段名, 到相应的头文件src/include/catalog/pg_type.h找到对应的类型 */
tupdesc = CreateTemplateTupleDesc(8, false);
TupleDescInitEntry(tupdesc, (AttrNumber) 1, "last_walend_time",
TIMESTAMPTZOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber) 2, "last_recv_lsn",
LSNOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber) 3, "last_apply_lsn",
LSNOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber) 4, "last_apply_delay_ms",
INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber) 5, "receiver_pid",
INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber) 6, "receiver_state",
INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber) 7, "receiver_start_time",
INT8OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber) 8, "receiver_conninfo",
TEXTOID, -1, 0);
BlessTupleDesc(tupdesc); // 完成对返回类型的构造, 参考src/include/funcapi.h
// 接下来将每个值转换为对应的Datum存储到values数组, 对应的nulls数组仅当值为空时设置为true.
TimestampTz receipttime;
receipttime = walrcv->latestWalEndTime;
values[0] = TimestampTzGetDatum(receipttime);
XLogRecPtr recvPtr;
recvPtr = GetWalRcvWriteRecPtr(NULL, NULL);
if (recvPtr == 0)
nulls[1] = true;
else
values[1] = LSNGetDatum(recvPtr);
XLogRecPtr applyPtr;
applyPtr = GetXLogReplayRecPtr(NULL);
if (recvPtr == 0)
nulls[2] = true;
else
values[2] = LSNGetDatum(applyPtr);
int apply_delay_ms;
apply_delay_ms = GetReplicationApplyDelay();
if (apply_delay_ms == -1)
nulls[3] = true;
else
values[3] = Int32GetDatum(apply_delay_ms);
values[4] = Int32GetDatum(walrcv->pid);
values[5] = Int32GetDatum(walrcv->walRcvState);
values[6] = Int64GetDatum(walrcv->startTime);
values[7] = PointerGetDatum(cstring_to_text((char *)walrcv->conninfo));
// 返回
/* Returns the record as Datum */
PG_RETURN_DATUM(HeapTupleGetDatum(
heap_form_tuple(tupdesc, values, nulls)));
}
[root@digoal ~]# gcc -O3 -Wall -Wextra -Werror -I /opt/soft_bak/postgresql-9.4.4/src/include -g -fPIC -c ./get_upstream_conninfo.c -o digoal.o
[root@digoal ~]# gcc -O3 -Wall -Wextra -Werror -I /opt/soft_bak/postgresql-9.4.4/src/include -g -shared digoal.o -o libdigoal.so
[root@digoal ~]# cp libdigoal.so /opt/pgsql/lib
创建函数和视图
postgres=# create or replace function get_rcv_replication_stat() returns record as '$libdir/libget_upstream_conninfo.so', 'get_rcv_replication_stat' language C STRICT;
postgres=# create or replace view get_rcv_replication_stat as
select now(),
extract(epoch from now()) as now_epoch,
last_walend_time,
last_recv_lsn,
last_apply_lsn,
last_apply_delay_ms,
receiver_pid,
case receiver_state
when 0 then 'stopped'
when 1 then 'starting'
when 2 then 'streaming'
when 3 then 'waiting'
when 4 then 'restarting'
when 5 then 'stopping'
else null end as receiver_state,
receiver_start_epoch,
conninfo
from get_rcv_replication_stat() as
t (last_walend_time timestamptz,
last_recv_lsn pg_lsn,
last_apply_lsn pg_lsn,
last_apply_delay_ms int,
receiver_pid int,
receiver_state int,
receiver_start_epoch int8,
conninfo text );
在主节点查询
postgres=# select * from get_rcv_replication_stat ;
ERROR: recovery is not in progress
HINT: This functions can only be executed during recovery.
在standby节点查询
postgres=# select * from get_rcv_replication_stat ;
-[ RECORD 1 ]--------+---------------------------------------------------------------
now | 2015-08-04 17:00:49.520785+08
now_epoch | 1438678849.52079
last_walend_time | 2015-08-04 17:00:48.841503+08
last_recv_lsn | 5/4B568518
last_apply_lsn | 5/4B568518
last_apply_delay_ms | 0
receiver_pid | 6667
receiver_state | streaming
receiver_start_epoch | 1438675316
conninfo | host=192.168.150.128 port=1921 user=replica keepalives_idle=60
