0
package testcode;
import java.sql.*;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class ProducerClear2 {
public static String vardbserver;
public static String vardbuser;
public static String vardbpassword;
public static String vardbname;
public static void main(String[] args) {
vardbserver = "TestDBtoMQ";
vardbuser = "postgresql";
vardbpassword = "admin";
ConnectionFactory factory = null;
javax.jms.Connection connection = null;
Session session = null;
Destination destination = null;
MessageProducer producer = null;
try {
factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_BROKER_URL);
connection = factory.createConnection();
connection.start();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
destination = session.createQueue("TestQueue");
producer = session.createProducer(destination);
Class.forName("org.postgresql.Driver");
System.out.println("----------------------------");
try (Connection c = DriverManager.getConnection("jdbc:postgresql://localhost:5432/" + vardbserver, vardbuser, vardbpassword);
PreparedStatement stmt = c.prepareStatement("SELECT * FROM MESSAGES where xmin::varchar::bigint > ? and xmin::varchar::bigint < ? ");
PreparedStatement max = c.prepareStatement("select max(xmin::varchar::bigint) as txid from messages")
) {
c.setAutoCommit(false);
Long previousTxId = 0L;
Long nextTxId = 0L;
while (true) {
stmt.clearParameters();
try (ResultSet rs = max.executeQuery()) {
if (rs.next()) {
nextTxId = rs.getLong(1);
}
}
stmt.setLong(1, previousTxId);
stmt.setLong(2, nextTxId + 1);
try (ResultSet rs = stmt.executeQuery()) {
while (rs.next()) {
String message = rs.getString("MESSAGE");
System.out.println("Message = " + message);
TextMessage mssg = session.createTextMessage(message);
System.out.println("Sent: " + mssg.getText());
producer.send(mssg);
}
previousTxId = nextTxId;
}
Thread.sleep(5000);
}
}
} catch (JMSException e) {
e.printStackTrace();
} catch (Exception e) {
System.err.println(e.getClass().getName() + ": " + e.getMessage());
} finally {
if (session != null) {
try {
session.close();
} catch (JMSException ex) {
// ignore
}
}
if (connection != null) {
try {
connection.close();
} catch (JMSException ex) {
// ignore
}
}
}
System.out.println("----------------------------");
System.out.println("Message sent successfully");
}
}
基本上,应用工程,以获取数据库的表内的内容,并将其发送到ActiveMQ的。当表更新时,它将发送刚刚更新的内容(不发送已发送的过去)。但是这个代码只适用于PostgreSQL
然后我打算创建一个“if”函数。所以我可以使用另一个数据库来获取数据(Oracle和MySQL)。
xmin是否仍适用于Oracle和MySQL?所以我只需要更改服务器的URL?或者我需要更改Oracle和MySQL的代码?
步骤1)旋转起来的VM。第2步)安装MySQL。步骤3)检查查询。 –
'xmin :: varchar :: bigint'不会在Oracle或MySQL中工作...... – JohnHC
xmin不是一个简单的升序编号。这将打破。 –