如何在 Java 中实现一个数据库监听器

Question

我有一个要求，如果在数据库表中插入一条记录，则需要自动执行一个 java 进程。实现数据库监听器的最简单方法是什么？

Answer 1

我有一个适用于 Oracle 的解决方案。你不需要创建你自己的，因为现在 Oracle 购买了 Java，它为它发布了一个监听器。据我所知，这在内部不使用轮询，而是将通知推送到 Java 端（可能基于某些触发器）：

public interface oracle.jdbc.dcn.DatabaseChangeListener 
extends java.util.EventListener {
    void onDatabaseChangeNotification(oracle.jdbc.dcn.DatabaseChangeEvent arg0);
}

你可以像这样实现它（这只是一个示例）：

public class DBListener implements DatabaseChangeListener {
    private DbChangeNotification toNotify;

    public BNSDBListener(DbChangeNotification toNotify) {
        this.toNotify = toNotify;
    }

    @Override
    public void onDatabaseChangeNotification(oracle.jdbc.dcn.DatabaseChangeEvent e) {
        synchronized( toNotify ) {
            try {
                toNotify.notifyDBChangeEvent(e); //do sth
            } catch (Exception ex) {
                Util.logMessage(CLASSNAME, "onDatabaseChangeNotification", 
                    "Errors on the notifying object.", true);
                Util.printStackTrace(ex);
                Util.systemExit();                                       
            }
        }       
    }
}

编辑：
可以使用以下类注册：oracle.jdbc.OracleConnectionWrapper

public class oracle.jdbc.OracleConnectionWrapper implements oracle.jdbc.OracleConnection {...}

假设你在某处创建了一个方法：

public void registerPushNotification(String sql) {
    oracle.jdbc.driver.OracleConnection oracleConnection = ...;//connect to db

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

    //this is what does the actual registering on the db end
    oracle.jdbc.dcn.DatabaseChangeRegistration dbChangeRegistration= oracleConnection.registerDatabaseChangeNotification(dbProperties);

    //now you can add the listener created before my EDIT
    listener = new DBListener(this);
    dbChangeRegistration.addListener(listener);

    //now you need to add whatever tables you want to monitor
    Statement stmt = oracleConnection.createStatement();
    //associate the statement with the registration:
    ((OracleStatement) stmt).setDatabaseChangeRegistration(dbChangeRegistration); //look up the documentation to this method [http://docs.oracle.com/cd/E11882_01/appdev.112/e13995/oracle/jdbc/OracleStatement.html#setDatabaseChangeRegistration_oracle_jdbc_dcn_DatabaseChangeRegistration_]

    ResultSet rs = stmt.executeQuery(sql); //you have to execute the query to link it to the statement for it to be monitored
    while (rs.next()) { ...do sth with the results if interested... }

    //see what tables are being monitored
    String[] tableNames = dbChangeRegistration.getTables();
    for (int i = 0; i < tableNames.length; i++) {
        System.out.println(tableNames[i]    + " has been registered.");
    }
    rs.close();
    stmt.close();
}

此示例不包括 try-catch 子句或任何异常处理。

Answer 2

这里有一个类似的答案：How to make a database listener with java?

您可以使用支持事务的消息队列来执行此操作，并在事务被提交或（连接关闭）不支持通知的数据库时触发消息。在大多数情况下，您必须手动通知并跟踪要通知的内容。

Spring 为 AMQP 和 JMS 提供了一些自动事务支持。您可以使用的更简单的替代方法是Guava's AsyncEventBus，但这仅适用于一个 JVM。 对于以下所有选项，我建议您使用消息队列通知平台的其余部分。

选项 - 非轮询非数据库特定

ORM 选项

像 Hibernate JPA have entity listeners 这样的一些库使这变得更容易，但那是因为他们假设他们管理所有的 CRUDing。

对于常规 JDBC，您必须自己记账。也就是说，在连接被提交或关闭之后，您将向 MQ 发送消息，表明某些内容已更新。

JDBC 解析

簿记的一个复杂选项是将您的java.sql.DataSource 和/或java.sql.Connection 包装/装饰成自定义的，这样您就可以在commit()（并关闭）上发送一条消息。我相信一些联合缓存系统可以做到这一点。您可以捕获已执行的 SQL 并解析以查看它是 INSERT 还是 UPDATE，但如果没有非常复杂的解析和元数据，您将无法获得行级监听。遗憾的是，我不得不承认这是 ORM 提供的优势之一，因为它知道您的更新内容。

道选项

如果您的不使用 ORM，最好的选择是在事务关闭后手动在您的 DAO 中发送一条消息，说明一行已更新。 请确保在发送消息之前交易已关闭。

选项 - 轮询非数据库特定

有点遵循@GlenBest 的建议。

我会做一些不同的事情。我会将定时器外部化或使其只有一台服务器运行定时器（即调度程序）。我只会使用 ScheduledExecutorService（最好将其包裹在 Guava 的 ListenerScheduledExecutorService 中）而不是 Quartz（恕我直言，使用石英轮询超级杀伤力）。

您想要观看的所有表格都应该添加一个“通知”列。

然后你做这样的事情：

// BEGIN Transaction
List<String> ids = execute("SELECT id FROM table where notified = 'f'");
//If db not transactional either insert ids in a tmp table or use IN clause
execute("update table set notified = 't' where notified = 'f'")
// COMMIT Transaction
for (String id : ids) { mq.sendMessage(table, id); }

选项 - 数据库特定

使用 Postgres NOTIFY，您仍然需要在一定程度上进行轮询，以便您完成上述大部分操作，然后将消息发送到总线。

Answer 3

一般的解决方案可能包括在感兴趣的表上创建一个触发器，通知任何听众关于INSERT 事件。一些数据库已经为这种进程间通知提供了正式的方法。例如：

甲骨文：

• DBMS_ALERT 是一种简单的通知方式
• Oracle AQ / Oracle Streams 提供更复杂的队列机制

Postgres：

• NOTIFY 语句是此类通知的简单方法

其他：

• 其他数据库中可能有类似的通知机制，我不知道。
• 您始终可以通过在事件表中插入一个事件来实现自己的事件通知队列表，该事件表由 Java 进程使用/轮询。不过，要做到这一点并提高性能可能会非常棘手。

Answer 4

假设：

• 拥有标准的可移植代码比即时实时执行 java 程序更重要。您希望允许对未来替代技术的可移植性（例如，避免专有数据库事件、外部触发器）。 Java 进程可以在记录添加到表后稍微运行（例如 10 秒后）。即调度+轮询或实时触发/消息/事件都是可以接受的。

• 如果一次将多行添加到表中，您希望运行一个进程，而不是多个。 DB 触发器会为每一行启动一个 java 进程 - 不合适。

• 服务质量很重要。即使出现硬件或软件致命错误，您也希望 java 程序再次运行并处理不完整的数据。

• 您希望将强大的安全标准应用于您的环境（例如，避免让 java 或 DB 直接执行 OS 命令）

• 你想最小化代码

  1. 不依赖专有数据库功能的核心 Java 标准代码：

     • 使用 ScheduledExecutorService 或 Quartz 调度程序（或 unix cron 作业或 Windows 任务调度程序）每分钟运行一次 java 程序（或每 10 秒运行一次）。它既充当调度程序又充当看门狗，确保程序全天候运行。 Quartz 也可以部署在应用服务器中。
     • 让您的 java 程序运行 1 分钟（或 10 秒），循环，通过 JDBC 查询 DB 并休眠几秒钟，然后最终退出。

  2. 如果您在应用服务器中有应用： 创建一个使用 Timer Service 的 Session Bean 并再次通过 JDBC Session Bean Timer Service 查询表。

  3. 有一个可以写入/追加到文件的数据库触发器。使用 java 7 filewatcher 在文件更改时触发逻辑Java 7 File Watcher

还有另一种选择：使用带有 DB 适配器触发逻辑（例如 Fuse 或 Mule 或 OpenAdapter）的开源 ESB，但这会提供超出您声明要求的强大功能，并且安装和学习既费时又复杂。

使用@Schedule 的EJB 定时器示例：

public class ABCRequest {
   // normal java bean with data from DB
}

@Singleton
public class ABCProcessor {    
    @Resource DataSource myDataSource;   
    @EJB ABCProcessor abcProcessor;
    // runs every 3 minutes 
    @Schedule(minute="*/3", hour="*")
    public void processNewDBData() {
        // run a JDBC prepared statement to see if any new data in table, put data into RequestData
        try
        {
           Connection con = dataSource.getConnection();
           PreparedStatement ps = con.prepareStatement("SELECT * FROM ABC_FEED;");
           ...
           ResultSet rs = ps.executeQuery();
           ABCRequest abcRequest
           while (rs.hasNext()) {
               // population abcRequest
           }
           abcProcessor.processABCRequest(abcRequst);
        } ...
    }    
}

@Stateless
public class class ABCProcessor {
    public void processABCRequest(ABCRequest abcRequest) {
    // processing job logic
    }
}

另请参阅：See This Answer，用于将 CDI 事件对象从 EJB 发送到 Web 容器。

Answer 5

我不确定此解决方案在多大程度上满足您的需求，但可以考虑作为一种选择。如果您使用的是 oracle，那么您可以编写一个 java 程序并将其编译为 oracle 函数。您可以从 post insert 触发器调用您的 java 程序。

Java program in oracle DB