Python oracledb连续查询通知（CQN）无法接收数据库变更消息

尝试在Python脚本中使用连续查询通知（CQN）功能，获取数据库中特定表的变更通知。数据库连接成功且可查询表数据，但回调函数无法接收消息。

相关代码 #

def cqn_callback(message):
    print("Notification:")
    for query in message.queries:
        for tab in query.tables:
            print("Table:", tab.name)
            print("Operation:", tab.operation)
            for row in tab.rows:
                if row.operation & oracledb.OPCODE_INSERT:
                    print("INSERT of rowid:", row.rowid)
                if row.operation & oracledb.OPCODE_DELETE:
                    print("DELETE of rowid:", row.rowid)

subscr = connection.subscribe(callback=cqn_callback,
                              operations=oracledb.OPCODE_INSERT | oracledb.OPCODE_DELETE,
                              qos=oracledb.SUBSCR_QOS_QUERY | oracledb.SUBSCR_QOS_ROWIDS)
subscr.registerquery("select * from regions")
input("Hit enter to stop CQN demo\n")

现象与环境 #

运行脚本后可看到数据库中已创建注册，但通过SQL*Plus或SQL Developer执行插入、删除操作后，仍无法收到通知。

• 数据库：Docker部署的Oracle 12C
• Python版本：3.10.7
• 使用oracledb厚模式，通过oracledb.init_oracle_client(lib_dir = ".../instantclient_21_3")加载客户端库

• 检查事务提交状态：Oracle CQN通知在事务提交后才会触发。确保在SQL*Plus或SQL Developer中执行插入/删除操作后，执行了COMMIT命令——如果操作在未提交的事务中，通知不会发送。
• 验证用户权限：确保数据库用户拥有CHANGE NOTIFICATION系统权限，同时对regions表拥有SELECT权限。执行以下SQL授予权限：

  GRANT CHANGE NOTIFICATION TO your_username;
  GRANT SELECT ON regions TO your_username;
  

• 调整QOS参数测试：Oracle 12C中，SUBSCR_QOS_QUERY对查询的确定性要求严格。可以尝试去掉该参数，改用表级通知测试：

  subscr = connection.subscribe(callback=cqn_callback,
                                operations=oracledb.OPCODE_INSERT | oracledb.OPCODE_DELETE,
                                qos=oracledb.SUBSCR_QOS_ROWIDS)
  subscr.registerquery("select * from regions")
  

• 检查数据库队列配置：确认数据库的JOB_QUEUE_PROCESSES参数值大于0，CQN依赖该队列发送通知。执行以下SQL查看并修改：

  SHOW PARAMETER JOB_QUEUE_PROCESSES;
  ALTER SYSTEM SET JOB_QUEUE_PROCESSES=10 SCOPE=BOTH;
  

• 验证版本兼容性：当前使用21.3版本的Instant Client连接12C数据库，虽然支持向下兼容，但可以尝试使用与数据库版本匹配的Instant Client（如12C版本）排除适配问题。
• 添加回调异常捕获：当前回调函数未处理异常，若执行时出错可能导致通知静默丢失。修改回调函数添加异常处理：

  def cqn_callback(message):
      try:
          print("Notification:")
          for query in message.queries:
              for tab in query.tables:
                  print("Table:", tab.name)
                  print("Operation:", tab.operation)
                  for row in tab.rows:
                      if row.operation & oracledb.OPCODE_INSERT:
                          print("INSERT of rowid:", row.rowid)
                      if row.operation & oracledb.OPCODE_DELETE:
                          print("DELETE of rowid:", row.rowid)
      except Exception as e:
          print(f"Callback error: {e}")
  

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