Reputation: 77
Is anyone have a reference for me for a ActiveMQ producer coding to sending messages from content inside a table in PostgreSQL, but i wish i can send all rows on the table become one-by-one message... I got a reference from someone that telling just same as i want above, but it tells how to get all the messages from content inside a .txt file... here the referance
So far i got this:
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");
}
}
Upvotes: 2
Views: 622
Reputation: 3913
why not simplifiying by using 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
UPDATE
you can optimize your code :
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");
}
Upvotes: 2
Reputation: 77
Got the answer...
package TestMQDB;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
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 ProducerDB {
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("----------------------------");
stmt = c.createStatement();
ResultSet rs = stmt.executeQuery("SELECT * FROM MESSAGES;");
while (rs.next()) {
String message = rs.getString("MESSAGE");
System.out.println("Message = " + message);
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);
TextMessage mssg = session.createTextMessage(message);
System.out.println("Sent: " + mssg.getText());
producer.send(mssg);
}
catch (JMSException e) {
e.printStackTrace();
}
}
rs.close();
stmt.close();
c.close();
}
catch (Exception e) {
System.err.println(e.getClass().getName()+": "+ e.getMessage());
}
System.out.println("----------------------------");
System.out.println("Message sent successfully");
}
}
Upvotes: 0