java监听oracle数据库

运维百科 2025年11月21日 04:27 248 admin

Java如何监听Oracle数据库变化？

在现代软件开发中,实时监控和响应数据库的变化是一项非常重要的功能，Oracle数据库作为一款强大的关系型数据库管理系统，其稳定性和高性能使其成为许多企业的首选，如何在Java应用中监听Oracle数据库的变化，以便及时作出反应，却是一个技术难题，本文将详细介绍几种实现Java监听Oracle数据库变化的方法。

使用JDBC轮询

最基础的方法是使用JDBC（Java Database Connectivity）进行轮询，通过定时连接数据库并查询特定表或视图，可以检测到数据的变化，这种方法简单易行，但效率较低，因为每次都需要重新建立连接并执行SQL查询。

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
public class DatabasePolling {
    public static void main(String[] args) throws Exception {
        String url = "jdbc:oracle:thin:@localhost:1521:orcl";
        String user = "username";
        String password = "password";
        while (true) {
            Connection connection = DriverManager.getConnection(url, user, password);
            Statement statement = connection.createStatement();
            ResultSet resultSet = statement.executeQuery("SELECT * FROM my_table");
            while (resultSet.next()) {
                // Process the row data
            }
            resultSet.close();
            statement.close();
            connection.close();
            Thread.sleep(5000); // Sleep for 5 seconds before next poll
        }
    }
}
使用Oracle的触发器和DBMS_AQ
Oracle提供了高级的消息队列服务DBMS_AQ（Advanced Queuing），结合数据库触发器，可以实现更高效的数据变化监听，当数据库中的表发生INSERT、UPDATE或DELETE操作时，触发器会将变化的信息发送到消息队列，然后Java应用程序可以从队列中获取这些信息进行处理。


创建消息队列和监听者
CREATE TABLE my_queue (msg_id NUMBER PRIMARY KEY, message VARCHAR2(4000));
CREATE QUEUE my_queue_q LINKAGED_GLOBALLY;
CREATE QUEUE_LINK q_link FOR my_queue_q USING 'tnsname';


创建触发器
CREATE OR REPLACE TRIGGER my_trigger
AFTER INSERT OR UPDATE OR DELETE ON my_table
FOR EACH ROW
BEGIN
    DBMS_AQ.ENQUEUE(
        queue_name => 'my_queue',
        priority => 1,
        message_properties => NULL,
        message_payload => :NEW.some_column || CHR(13) || CHR(10), -- Encode newline characters
        delivery_policy => 'PERSISTENT');
END;


在Java中消费消息
import javax.jms.*;
import com.sun.messaging.ConnectionFactory;
public class MessageConsumer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new com.sun.messaging.ConnectionFactory();
        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue("javax.jms.queue.my_queue");
        MessageConsumer consumer = session.createConsumer(queue);
        while (true) {
            Message message = consumer.receive();
            if (message instanceof TextMessage) {
                String text = ((TextMessage) message).getText();
                System.out.println("Received message: " + text);
            }
        }
    }
}


使用第三方库如Apache Kafka和Debezium
除了Oracle自带的工具外,还可以利用第三方库来实现数据库变化的监听，Apache Kafka结合Debezium，可以实现跨多个数据库的数据变更捕获和实时处理。


安装和配置Kafka和Debezium

请参考Debezium官方文档进行安装和配置。



创建Debezium连接器
{
  "connector.class": "io.debezium.connector.oracle.OracleConnector",
  "database.hostname": "localhost",
  "database.port": "1521",
  "database.user": "username",
  "database.password": "password",
  "database.dbname": "orcl",
  "database.server.name": "dbserver1",
  "database.include.list": "my_table",
  "database.history.kafka.bootstrap.servers": "localhost:9092",
  "database.history.kafka.topic": "schema-changes.my_table"
}


在Java中消费Kafka消息
使用Apache Kafka客户端库来消费Kafka主题中的消息。
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import java.util.Collections;
import java.util.Properties;
public class KafkaMessageConsumer {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Collections.singletonList("schema-changes.my_table"));
        while (true) {
            ConsumerRecord<String, String> record = consumer.poll(Duration.ofMillis(100)).iterator().next();
            System.out.println("Received record: " + record.value());
        }
    }
}


通过以上方法,你可以在Java应用中实现对Oracle数据库变化的监听。