Oracle DCN的DatabaseChangeListener未触发事件通知求助

问题概述 #

实现Oracle Database Change Notification（DCN）监听表的插入操作，已通过SELECT * FROM USER_CHANGE_NOTIFICATION_REGS确认注册成功，但无论是应用内部还是外部脚本插入记录，控制台都无法收到事件通知。

可能的原因及解决方案 #

1. 注册用的连接被过早关闭 #

在initialize方法的finally块中，你关闭了创建DCN注册的连接conn.close()。Oracle JDBC的DCN实现依赖该连接接收数据库的推送通知，关闭连接会导致客户端无法接收后续事件。

解决方法：移除finally块中的conn.close()代码，保持该连接活跃。如果需要管理连接生命周期，可使用连接池并确保该连接在应用运行期间不被关闭。

2. 用户缺少必要权限 #

数据库用户需要拥有CHANGE NOTIFICATION权限才能接收DCN通知，即使注册成功，权限不足也会导致无法收到事件。

解决方法：执行以下SQL授予权限：

GRANT CHANGE NOTIFICATION TO CONTRATOS;

3. DCN属性设置不符合需求 #

你设置了OracleConnection.DCN_QUERY_CHANGE_NOTIFICATION = "true"，该属性针对查询结果集的变化通知（仅当查询返回的记录发生变化时触发）。如果需要监听整个表的所有插入操作，无需启用该属性。

解决方法：移除该属性设置，或改为false：

// 移除这一行
// prop.setProperty(OracleConnection.DCN_QUERY_CHANGE_NOTIFICATION, "true");

4. Spring初始化线程阻塞问题 #

@PostConstruct方法是Spring初始化Bean时的主线程，你在方法末尾调用this.wait()会导致Spring初始化卡住，且一旦线程被中断，监听器将停止工作。

解决方法：将DCN监听逻辑移至独立后台线程，避免阻塞Spring初始化：

@PostConstruct
public void initialize() {
    new Thread(() -> {
        try {
            var conn = connect();
            var prop = new Properties();
            prop.setProperty(OracleConnection.DCN_NOTIFY_ROWIDS, "true");
            
            DatabaseChangeRegistration dcr = conn.registerDatabaseChangeNotification(prop);
            DCNListener list = new DCNListener(this);
            dcr.addListener(list);
            
            Statement stmt = conn.createStatement();
            ((OracleStatement) stmt).setDatabaseChangeRegistration(dcr);
            ResultSet rs = stmt.executeQuery("SELECT * FROM CONTRATOS.CON_ESTADOS_PAGOS_CONTRATOS_NOTIFY");
            
            for (String table : dcr.getTables())
                System.out.println(table + " is part of the registration.");
            rs.close();
            stmt.close();
            
            // 保持线程运行以维持监听
            synchronized (this) {
                this.wait();
            }
        } catch (SQLException | InterruptedException e) {
            e.printStackTrace();
        }
    }).start();
}

5. 监听器线程未持续运行 #

DCN监听器依赖后台线程接收事件，如果初始化线程结束，监听器可能被销毁。需确保有持续运行的线程维持监听状态。

附用户实现代码 #

OracleDCNListener类 #

package co.app.dcn;

import oracle.jdbc.OracleConnection;
import oracle.jdbc.OracleDriver;
import oracle.jdbc.OracleStatement;
import oracle.jdbc.dcn.DatabaseChangeRegistration;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Properties;

@Component
public class OracleDCNListener {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Value("${spring.datasource.username}")
    private String DCN_USER;

    @Value("${spring.datasource.password}")
    private String DCN_PASSWORD;

    @Value("${spring.datasource.url}")
    private String DCN_URL;

    @PostConstruct
    public void initialize() throws SQLException {


        var conn = connect();

        var prop = new Properties();

        prop.setProperty(OracleConnection.DCN_NOTIFY_ROWIDS, "true");
        prop.setProperty(OracleConnection.DCN_QUERY_CHANGE_NOTIFICATION, "true");

        DatabaseChangeRegistration dcr = conn.registerDatabaseChangeNotification(prop);

        try {
            DCNListener list = new DCNListener(this);
            dcr.addListener(list);

            // second step: add objects in the registration:
            Statement stmt = conn.createStatement();
            // associate the statement with the registration:
            ((OracleStatement) stmt).setDatabaseChangeRegistration(dcr);
            ResultSet rs = stmt.executeQuery("SELECT * FROM CONTRATOS.CON_ESTADOS_PAGOS_CONTRATOS_NOTIFY");

            String[] tableNames = dcr.getTables();
            for (int i = 0; i < tableNames.length; i++)
                System.out.println(tableNames[i] + " is part of the registration.");
            rs.close();
            stmt.close();
        } catch (SQLException ex) {
            if (conn != null)
                conn.unregisterDatabaseChangeNotification(dcr);
            throw ex;
        } finally {
            try {
                // Note that we close the connection!
                conn.close();
            } catch (Exception innerex) {
                innerex.printStackTrace();
            }
        }

        synchronized (this) {
            // The following code modifies the dept table and commits:
            try {
                OracleConnection conn2 = connect();
                conn2.setAutoCommit(false);
                Statement stmt2 = conn2.createStatement();
                stmt2.executeUpdate("insert into contratos.CON_ESTADOS_PAGOS_CONTRATOS_NOTIFY (ID, ID_CLASE_DOCUMENTO, NUMERO_DOCUMENTO, OBSERVACIONES, ID_ESTADO, DESCRIPCION_ESTADO, FECHA_HORA_CREACION) values (99999920, 0, 0, ' ', 0, ' ', ' ')",
                        Statement.RETURN_GENERATED_KEYS);
                ResultSet autoGeneratedKey = stmt2.getGeneratedKeys();
                if (autoGeneratedKey.next())
                    System.out.println("inserted one row with ROWID=" + autoGeneratedKey.getString(1));
                stmt2.executeUpdate("insert into contratos.CON_ESTADOS_PAGOS_CONTRATOS_NOTIFY (ID, ID_CLASE_DOCUMENTO, NUMERO_DOCUMENTO, OBSERVACIONES, ID_ESTADO, DESCRIPCION_ESTADO, FECHA_HORA_CREACION) values (99999922, 0, 0, ' ', 0, ' ', ' ')",
                        Statement.RETURN_GENERATED_KEYS);
                autoGeneratedKey = stmt2.getGeneratedKeys();
                if (autoGeneratedKey.next())
                    System.out.println("inserted one row with ROWID=" + autoGeneratedKey.getString(1));
                stmt2.close();
                conn2.commit();
                conn2.close();
            } catch (SQLException ex) {
                ex.printStackTrace();
            }

            // wait until we get the event
            try {
                this.wait();
            } catch (InterruptedException ie) {
            }
        }

        // At the end: close the registration (comment out these 3 lines in order
        // to leave the registration open).
        //OracleConnection conn3 = connect();
        //conn3.unregisterDatabaseChangeNotification(dcr);
        //conn3.close();
    }

    OracleConnection connect() throws SQLException {
        OracleDriver dr = new OracleDriver();
        Properties prop = new Properties();
        prop.setProperty("user", DCN_USER);
        prop.setProperty("password", DCN_PASSWORD);

        return (OracleConnection) dr.connect(DCN_URL, prop);
    }

}

DCNListener类 #

package co.app.dcn;

import oracle.jdbc.dcn.DatabaseChangeEvent;
import oracle.jdbc.dcn.DatabaseChangeListener;

public class DCNListener implements DatabaseChangeListener {

    OracleDCNListener oracleDCNListener;

    public DCNListener(OracleDCNListener oracleDCNListener) {
        this.oracleDCNListener = oracleDCNListener;
    }

    public void onDatabaseChangeNotification(DatabaseChangeEvent e)
    {
        Thread t = Thread.currentThread();
        System.out.println("DCNDemoListener: got an event ("+this+" running on thread "+t+")");
        System.out.println(e.toString());
        synchronized( oracleDCNListener ){ oracleDCNListener.notify();}
    }
}

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