java - ActiveMQ 生产者从 PostgreSQL 表内的内容发送消息

Question

有人可以为我提供一个 ActiveMQ 生产者编码以从 PostgreSQL 中的表内的内容发送消息的引用吗，但我希望我可以将表上的所有行发送为一对一的消息... 我从某人那里得到了一份引用资料，内容与我上面想要的内容相同，但它告诉我如何从 .t​​xt 文件内的内容获取所有消息... here the referance

到目前为止我得到了这个:

package testMQDB;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
public class ProducerDB {

   public static void ProducerDB(String[] args){
       Connection c = null;
       Statement stmt = null;
       try {
       Class.forName("org.postgresql.Driver");
         c = DriverManager.getConnection("jdbc:postgresql://localhost:5432/"TestDB, "admin", "admin");
         c.setAutoCommit(false);
         System.out.println("----------------------------");

         stmt = c.createStatement();
         ResultSet rs = stmt.executeQuery( "SELECT * FROM MESSAGES;" );
         while ( rs.next() ) {
            String  message = rs.getString("MESSAGE");
            System.out.println( "Message = " + message );
            if (message.equals(true)){
                // write the ActiveMQ code here? 
            }
         }
         rs.close();
         stmt.close();
         c.close();
       } catch ( Exception e ) {
         System.err.println( e.getClass().getName()+": "+ e.getMessage() );
         System.exit(0);
       }
       System.out.println("----------------------------");
       System.out.println("Message sent successfully");
   }
}

Answer 1

为什么不使用 Camel 来简化？

http://camel.apache.org/sql-component.html

from("sql:select * from table?maxMessagesPerPoll=1&dataSource=pg") .to("activemq:queue:customers");

<bean id="pg" 
class="org.springframework.jdbc.datasource.DriverManagerDataSource"> 
        <property name="driverClassName" value="org.postgresql.Driver"/> 
        <property name="url" value="jdbc:postgresql://localhost:5432/ddd"/> 
        <property name="username" value="user"/> 
        <property name="password" value="password"/> 
    </bean> 

<route> 
      <description>select 1 row</description> 
      <from uri="sql:select * from table?maxMessagesPerPoll=1&dataSource=pg" /> 
      <to  uri="activemq:queue:customers"/>
</route> 

https://github.com/apache/camel/tree/master/examples/camel-example-sql http://camel.apache.org/jms.html

http://activemq.apache.org/sample-camel-routes.html

---does activemq have a way to pick entries from db table & insert into queue

更新

您可以优化您的代码:

public static void ProducerDB(String[] args){
    ConnectionFactory factory = null;
    javax.jms.Connection connection = null;
    Session session = null;
    Destination destination = null;
    MessageProducer producer = null;
    Connection c = null;
    Statement stmt = null;
    try {
    Class.forName("org.postgresql.Driver");
    c = DriverManager.getConnection("jdbc:postgresql://localhost:5432/"TestDB, "admin", "admin");
    c.setAutoCommit(false);
    System.out.println("----------------------------");
    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);

    stmt = c.createStatement();
    ResultSet rs = stmt.executeQuery("SELECT * FROM MESSAGES;");
    while (rs.next()) {
        String  message = rs.getString("MESSAGE");
        System.out.println("Message = " + message);
            try {
                TextMessage mssg = session.createTextMessage(message);
                System.out.println("Sent: " + mssg.getText());
                producer.send(mssg);
            }
            catch (JMSException e) {
                e.printStackTrace();
            }
    }
    }catch (Exception e) {
        System.err.println(e.getClass().getName()+": "+ e.getMessage());
    }finally{
        if (c != null) {
            c.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
    System.out.println("----------------------------");
    System.out.println("Message sent successfully");
}