如何不使用SQL查询，通过Streams从数据库获取数据

之前面试遇到的这个问题，核心是要绕开SQL查询的方式，直接从数据库的变更流/日志层获取数据——面试官说的Streams其实指的是**变更数据捕获（CDC）**或者数据库原生的日志流机制，这类方案完全不依赖SQL，和ORM有本质区别（ORM只是封装了SQL，底层还是会发查询请求）。

下面是不同主流数据库的具体实现方式：

MySQL：解析Binlog流 #

MySQL的Binlog是记录所有数据变更的二进制日志，我们可以直接读取并解析这个流来获取数据变化，不需要执行任何SQL。常用的工具包括Java的Canal、Python的pymysqlreplication。

举个Python的简单示例：

from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_event import WriteRowsEvent, UpdateRowsEvent, DeleteRowsEvent

# 初始化Binlog流读取器
stream = BinLogStreamReader(
    connection_settings={"host": "localhost", "port": 3306, "user": "root", "passwd": "your_pwd"},
    server_id=100,  # 自定义唯一ID，不能和数据库集群内其他节点重复
    blocking=True,  # 持续阻塞等待新的Binlog事件
    only_events=[WriteRowsEvent, UpdateRowsEvent, DeleteRowsEvent]  # 只监听增删改事件
)

# 循环消费变更事件
for event in stream:
    for row in event.rows:
        if isinstance(event, WriteRowsEvent):
            print(f"新增数据: {row['values']}")
        elif isinstance(event, UpdateRowsEvent):
            print(f"更新数据: 旧值{row['before_values']}, 新值{row['after_values']}")
        elif isinstance(event, DeleteRowsEvent):
            print(f"删除数据: {row['values']}")

stream.close()

PostgreSQL：逻辑复制流 #

PostgreSQL支持逻辑复制，通过将WAL（预写日志）解析为可读的变更数据来实现流读取。需要先开启逻辑复制模式，再创建发布和订阅，或者直接用插件读取流。

步骤+示例：

1. 修改postgresql.conf，设置wal_level = logical，重启数据库
2. 创建发布（指定要监听的表）：

   CREATE PUBLICATION my_data_pub FOR TABLE public.my_table;
   

3. 用Python的psycopg2读取流：

import psycopg2
from psycopg2.extras import LogicalReplicationConnection

# 建立逻辑复制连接
conn = psycopg2.connect(
    dbname="my_db", user="postgres", password="your_pwd", host="localhost",
    connection_factory=LogicalReplicationConnection
)
cur = conn.cursor()

# 定义变更回调函数
def handle_change(msg):
    print(f"收到变更记录: {msg.payload}")
    # 发送反馈确认已处理，避免重复消费
    msg.cursor.send_feedback(flush_lsn=msg.data_start)

# 启动复制，指定slot名称（需提前创建）
cur.start_replication(slot_name='my_repl_slot', decode=True, callback=handle_change)

# 持续监听
while True:
    cur.poll()

Oracle：Streams CDC或GoldenGate #

Oracle有原生的Streams CDC功能，也可以用GoldenGate实现更复杂的变更捕获。通过Oracle Streams API，你可以直接订阅数据库的变更流，获取实时的增删改数据，全程不需要执行SQL查询。

比如用Oracle的DBMS_CDC_PUBLISH和DBMS_CDC_SUBSCRIBE包来配置捕获和订阅，之后就能从CDC视图中读取变更流（注意这里的视图是CDC机制维护的，不是通过SQL查询原始表）。

关键区别说明 #

• ORM：本质是生成并执行SQL，只是帮你封装了SQL语法，最终还是会向数据库发送查询请求，不符合要求。
• Streams/CDC：直接读取数据库的底层变更日志，相当于从数据库的"操作记录"里拿数据，完全不依赖SQL查询，这才是面试官要的方案。

内容的提问来源于stack exchange，提问作者Aravindh_P