如何在Java中实现db侦听器

Question

我有一个要求,如果在db表中插入记录,则需要自动执行java进程.实现db侦听器的最简单方法是什么？

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

类似的答案:如何用java创建数据库监听器？

您可以使用支持事务的消息队列执行此操作,并在为不支持通知的数据库进行事务处理或(连接已关闭)时触发消息.在大多数情况下,您必须手动通知并跟踪要通知的内容.

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

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

ORM选项

像Hibernate JPA 这样的库有实体监听器,这使得这更容易,但这是因为他们假设他们管理所有的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程序更重要.您希望允许替代未来技术的可移植性(例如,避免专有数据库事件,外部触发器).将记录添加到表后(例如10秒后),Java进程可以稍微运行.即,调度+轮询或实时触发/消息/事件都是可接受的.

• 如果一次向表中添加多行,则需要运行一个进程,而不是多个进程.数据库触发器将为每一行启动一个java进程 - 不合适.

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

• 您希望对您的环境应用强大的安全标准(例如,避免让java或DB直接执行OS命令)

• 您希望最小化代码

  1. 核心Java标准代码,不依赖于专有数据库功能:

     • 使用ScheduledExecutorService或Quartz调度程序(或unix cron作业或Windows任务调度程序)每分钟运行一次java程序(或者可以每10秒执行一次).它既可作为调度程序,也可作为监视程序,确保程序全天候运行.Quartz也可以部署在app服务器中.
     • 让你的java程序运行1分钟(或10秒),循环,通过JDBC查询数据库并休眠几秒钟,然后最终退出.

  2. 如果您在appserver中有app:创建一个使用Timer Service的Session Bean,并再次通过JDBC Session Bean Timer Service查询该表.

  3. 有一个写入/附加到文件的数据库触发器.使用java 7 filewatcher触发文件更改Java 7文件观察器的逻辑

还有另一种选择:使用带有数据库适配器触发逻辑的开源ESB(例如Fuse或Mule或OpenAdapter),但这提供了超出您规定要求的强大功能,并且安装和学习既费时又复杂.

使用@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
    }
}

另请参阅:请参阅本答案,将CDI事件对象从EJB发送到Web容器.