基于Oracle Streams + Oracle AQ 捕获变更，发布变更（二）

要求：使用Oracle Streams捕获某个用户下部分表的DML操作变更，并通过Oracle的AQ（高级队列）对外发布，然后Java端通过JMS来获取变更，并执行后续同步操作。

 

Java部分：

 

1、jms配置信息

jms.local.username=strmadmin

jms.local.password=strmadmin

jms.local.jdbcUrl=jdbc:oracle:thin:@127.0.0.1:1521:orcl

jms.local.queueName=jms_queue

jms.local.agentName=jms_agent

jms.local.batchSize=10

jms.local.receiveTimeout=3000

 

对应的java class：

public final class JmsConfig {

    public String username;

    public String password;

    public String jdbcUrl;

    public String queueName;

    public String agentName;

   

    public int batchSize = 10;

    public long receiveTimeout = 5000L;

}

 

2、获取TopicConnection

private TopicConnection getTopicConnection(String jdbcUrl, String username, String password) throws JMSException {

        Properties info = new Properties();

        info.put(username, password);

 

        TopicConnectionFactory topicConnectionFactory =

            AQjmsFactory.getTopicConnectionFactory(jdbcUrl,info);

       

        return topicConnectionFactory.createTopicConnection(username, password);

    }

 

3、建立连接，并获取session

JmsConfig config = ......;

TopicConnection conn = getTopicConnection(config.jdbcUrl, config.username, config.password);

AQjmsSession session = (AQjmsSession)conn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);

       

conn.start();

 

4、获取Topic，创建subscriber，并接受消息

AQjmsDestination topic = (AQjmsDestination)session.getTopic(config.username, config.queueName);

       

AQjmsConsumer subscriber = (AQjmsConsumer)session.createDurableSubscriber(topic,  config.agentName); 

       

topic.start(session, false, true);

 

Message m = subscriber.receive(config.receiveTimeout);

AQjmsBytesMessage payload = (AQjmsBytesMessage)msg;

Long id = payload.getLongProperty("id");

String tablename =  payload.getStringProperty("tablename");

......

说明：这里接受信息会阻塞，在超过给定的超时时间后，如果没有获取到消息，会返回null。

JMS还支持另外一种监听方式：MessageListener

subscriber.setMessageListener(new MessageListener() {

    @Override

    public void onMessage(Message message) {

    }

});

    

5、在获取到Message以后，就可以后去其中的数据，然后进行后续的操作，比如通过这些信息，去更新另一个库中对应的数据。

 

需要注意的问题：

1、需要的oracle相关的jar包：aqapi13_g，jmscommon，orai18n以及ojdbc5-11.2.0.3.0

2、由于数据库的编码为ZHS16GBK，而该编码Java本身不支持，导致在获取非数字的信息时，比如表名等会出现乱码。好找oracle自身有国际化支持，只需要将对应的jar包orai18n加入classpath即可。该jar包可以在oracle客户端或者服务端的jlib目录下面找到。

3、JMS消息接收在采用监听的方式时，是在独立的线程中进行的，如果当前线程没有阻塞，则该线程结束也会导致监听线程结束。可以通过调用subscriber.getListenerWorkerThread().join() 让当前线程等待监听线程结束。

4、在创建Topic时，createTopicSession(false, Session.AUTO_ACKNOWLEDGE) 需要设置Session.AUTO_ACKNOWLEDGE，这样当我们接受消息以后，JMS会自动回复确认消息，然后Oracle数据库消息队列jms_queue中的数据才会被移除，否则jms_queue队列的数据会一直存在。