简单的Java中间件

在support公司项目的时候发现一个问题，需要从SQL Server的Logger Database中取出数据并作一定逻辑的判断，如果符合条件的话就将记录中的一个XML类型字段的值推送到ActiveMQ中的Queue中，所以就写了一个简单的Java中间件来推送，中间件主要分两部分：1.从SQL Server中抽取数据；2.Send Message到ActiveMQ。

使用JDBC从SQL Server中抽取数据

1.创建一个名为ActiveMQ的Java Project



2.下载连接SQL Server的Jar包

因为我需要连接的SQL Server是2012版的，所以我下载了Microsoft JDBC Driver 4.0 for SQL Server，它支持的版本有:

Microsoft®SQL Server® 2012

Microsoft®SQL Server® 2008 R2

Microsoft®SQL Server® 2008

Microsoft®SQL Server® 2005

Microsoft®SQL Azure

下载地址为：

http://www.microsoft.com/zh-cn/download/details.aspx?id=11774

将下载的压缩包解压到一个文件夹内，一直打开到最里面的文件夹，会发现有2个JAR包：



我们需要的是sqljdbc4.jar这个包

3.添加JAR包

在已经创建的ActiveMQ Project中新建一个lib文件夹，然后将刚才下载的sqljdbc4.jar包copy到这个lib，选中这个Jar包，点击右键选择Build Path->Add Build to Path

4.创建DBConnect类

代码如下

package com.lm.activemq;
import java.sql.*;

import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
public class DBConnect
{
public static void sendMessage(Session session, MessageProducer producer)
throws Exception {
String driverName="com.microsoft.sqlserver.jdbc.SQLServerDriver";
String dbURL="jdbc:sqlserver://XXXXXX:1433;DatabaseName=XXXX";
String userName="sa";
String userPwd="XXXX";

Statement ste = null;
try
{
Class.forName(driverName);
Connection dbConn=DriverManager.getConnection(dbURL,userName,userPwd);
ste = dbConn.createStatement(ResultSet.TYPE_SCROLL_SENSITIVE,  ResultSet.CONCUR_UPDATABLE);
System.out.println("连接数据库成功");

String orderNumbers[] = new String[] {
"28003"
};
for(int i = 0;i < orderNumbers.length ;i++ )
{
System.out.println("当前Order为:"+i);
String filterStr = "000"+orderNumbers[i];
ResultSet rst = ste.executeQuery("select * from FWK_TrackLog where OriginRecordId in ('"+filterStr+"') order by TrackId asc ");
int rowCount = 0;
try {
rst.last();
rowCount = rst.getRow();
} catch (Exception e) {
// TODO: handle exception
e.printStackTrace();
}
//System.out.println("行数为:"+rowCount);

if(rowCount == 1)
{
rst.beforeFirst();//remove to first
while(rst.next())
{
System.out.println("OrderNumber:"+rst.getString("OriginRecordId"));
System.out.println(rst.getString("Payload"));
TextMessage message = session.createTextMessage(rst.getString("Payload"));
producer.send(message);
}
}
}
if(ste!=null)
{
ste.close();
}
if(dbConn!=null)
{
dbConn.close();
}
}
catch(Exception e)
{
e.printStackTrace();
System.out.print("连接失败");
}
}

}

Send Message到ActiveMQ

1.添加ActiveMQ的Jar包到Project

我使用的ActiveMQ版本是apache-activemq-5.11.1-bin，需要添加的Jar包是



和刚才添加sqljdbc4.jar包一样，选中这5个Jar包，然后右击选择Build Path->Add Build to Path

4000
2.创建Sender类

这个类的主要作用是的连接ActiveMQ

package com.lm.activemq;

/**
* @Header: Sender.java
* 类描述：
*/
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Sender {

public static void main(String[] args) {

// ConnectionFactory ：连接工厂，JMS 用它创建连接
ConnectionFactory connectionFactory; // Connection ：JMS 客户端到JMS
// Provider 的连接
Connection connection = null; // Session： 一个发送或接收消息的线程
Session session; // Destination ：消息的目的地;消息发送给谁.
Destination destination; // MessageProducer：消息发送者
MessageProducer producer; // TextMessage message;
// 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");
try { // 构造从工厂得到连接对象
connection = connectionFactory.createConnection();
// 启动
connection.start();
// 获取操作连接
session = connection.createSession(Boolean.TRUE,
Session.AUTO_ACKNOWLEDGE);
// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
destination = session.createQueue("OrderRequestFeedback");
// 得到消息生成者【发送者】
producer = session.createProducer(destination);
// 设置不持久化，此处学习，实际根据项目决定
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
// 构造消息，此处写死，项目就是参数，或者方法获取
DBConnect.sendMessage(session, producer);
session.commit();
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
}
}
}
}