当前位置：首页 > article >正文

PG逻辑订阅功能

article 2024/11/19 4:30:54

目录标题

一、复制标识、存储过程
- - 排查处理过程
  - - 步骤1：确认表的复制标识设置
    - 步骤2：设置复制标识
    - 步骤3：检查存储过程的逻辑
    - 步骤4：重新尝试执行存储过程
    - 步骤5：验证表结构
  - 总结
二、相关视图
- - 分析查询结果
  - - 查询`pg_stat_subscription`
    - 查询`pg_stat_replication`
    - 查询`pg_replication_slot`
    - 查询`pg_subscription_rel`
  - 总结
  - 建议
三、逻辑订阅功能
- - - 1. 确保逻辑复制是启用的
    - 2. 创建一个复制插槽
    - 3. 创建一个订阅
    - 4. 查看订阅状态
    - 5. 查看复制插槽状态
    - 6. 管理订阅的同步状态
    - 7. 处理订阅数据
    - 8. 取消订阅
    - 注意事项：

一、复制标识、存储过程

排查处理过程

步骤1：确认表的复制标识设置

查询当前表的复制标识设置：
```
SHOW replica_identity ON t_my_table;
```
如果返回结果不是FULL或DEFAULT：
- 说明表没有正确的复制标识，需要设置复制标识。

步骤2：设置复制标识

设置复制标识：

ALTER TABLE t_my_table REPLICA IDENTITY FULL;

验证设置是否成功：
```
SHOW replica_identity ON t_my_table;
```

步骤3：检查存储过程的逻辑

查看存储过程p_altercolumn的定义：

CREATE OR REPLACE FUNCTION p_altercolumn(p_table character varying, p_column character varying, p_datatype character varying, p_isnotnull character varying, p_defaultvalue character varying, p_others character varying)
RETURNS void AS $$
BEGIN
    -- 逻辑处理，例如：
    EXECUTE format('ALTER TABLE %I ALTER COLUMN %I TYPE %s', p_table, p_column, p_datatype);
    IF p_isnotnull = 'NOT NULL' THEN
        EXECUTE format('ALTER TABLE %I ALTER COLUMN %I SET NOT NULL', p_table, p_column);
    END IF;
    IF p_defaultvalue <> 'NULL' THEN
        EXECUTE format('ALTER TABLE %I ALTER COLUMN %I SET DEFAULT %s', p_table, p_column, p_defaultvalue);
    END IF;
    -- 其他逻辑
END;
$$ LANGUAGE plpgsql;

确认存储过程能够处理各种数据类型和约束。

步骤4：重新尝试执行存储过程

调用存储过程：

CALL p_altercolumn('t_my_table', 'fid', 'BIGINT', 'NOT NULL', '1111', '0');

检查执行结果：
- 确认存储过程执行没有错误。

步骤5：验证表结构

查询表结构：
```
\d t_my_table
```
确认列fid的数据类型和约束：
- 数据类型应该是BIGINT。
- 应该设置为NOT NULL。

总结

通过以上步骤，我们可以确保：

表的复制标识设置正确。
存储过程逻辑正确。
存储过程能够成功执行。

如果在执行过程中遇到任何错误，请记录错误信息并进一步排查。例如：

检查日志：
- 分析数据库日志，查看是否有更多详细信息。
检查权限：
- 确保执行存储过程的用户具有足够的权限。
检查依赖：
- 确保所有依赖的服务和组件都正常运行。

通过这些步骤，应该能够解决更新表时遇到的问题。如果还有其他问题，请提供更多信息。

二、相关视图

分析查询结果

查询`pg_stat_subscription`

目的：获取数据库订阅的状态。
结果：(0 rows)，表明数据库中没有活跃的订阅。

查询`pg_stat_replication`

目的：获取流复制的状态信息。
结果：返回了多个流复制进程的信息。
示例信息：
- pid：16394，表示复制进程的进程ID。
- usename：未提供具体值，表示执行复制的数据库角色用户名。
- application_name：PostgreSQL JDBC，表明复制客户端使用的是PostgreSQL JDBC驱动连接。
- client_addr：245.0.0.9，表示复制客户端的IP地址。

查询`pg_replication_slot`

目的：获取复制插槽的状态。
结果：返回了复制插槽的详细信息。
示例信息：
- slot_name：dbz，表示复制插槽的名称。
- plugin：decoderbufs，表示使用的解码插件，用于逻辑解码。

查询`pg_subscription_rel`

目的：获取订阅关联的关系状态。
结果：(0 rows)，表明没有订阅关系信息。

总结

订阅信息：数据库中没有配置订阅，pg_stat_subscription和pg_subscription_rel查询均返回空结果。
流复制状态：pg_stat_replication查询显示存在活跃的流复制进程，表明数据库复制正在运行。
复制插槽：pg_replication_slot查询显示了复制插槽的详细信息，表明复制插槽已正确配置。

建议

检查订阅配置：如果预期数据库中应该有订阅配置，需要检查为什么没有订阅信息。可能需要配置订阅或检查订阅是否被删除。
监控复制状态：继续监控pg_stat_replication以确保流复制的健康状态。
检查复制插槽使用情况：如果需要使用逻辑解码，检查pg_replication_slot以确保插槽被正确使用。

三、逻辑订阅功能

PostgreSQL的逻辑订阅功能允许你订阅一个或多个表的变化，这样当这些表发生变化时，你可以得到通知。订阅可以用于多种用途，比如数据同步、审计日志、缓存更新等。

以下是设置和使用PostgreSQL订阅的一般步骤：

1. 确保逻辑复制是启用的

在PostgreSQL的配置文件（通常是postgresql.conf）中，确保以下参数被设置：

wal_level = logical
max_replication_slots = 1 # 或更多，取决于你的需要
max_wal_senders = 1 # 或更多

2. 创建一个复制插槽

复制插槽是用于逻辑复制的WAL（Write-Ahead Logging）流的持久化存储区。可以使用以下命令创建一个：

SELECT * FROM pg_create_logical_replication_slot('my_slot', 'pgoutput');

这里my_slot是新插槽的名字，pgoutput是输出插件的名字，用于逻辑解码。

3. 创建一个订阅

你可以使用以下命令创建一个订阅：

CREATE SUBSCRIPTION my_subscription
    CONNECTION 'host=myhost user=myuser password=mypass dbname=mydb'
    PUBLICATION my_publication
    WITH (copy_data = true);

这里：