基于jms使用ActiveMQ实现异步日志功能.消息持久到oracle 10g 数据库

importjava.io.Serializable;

importorg.apache.commons.lang.builder.EqualsBuilder;

importorg.apache.commons.lang.builder.HashCodeBuilder;

importorg.apache.commons.lang.builder.ReflectionToStringBuilder;

importorg.apache.commons.lang.builder.ToStringBuilder;

publicclassBaseEntityimplementsSerializable{

/**

publicStringtoString(){

returnReflectionToStringBuilder.toString(this);

}

publicbooleanequals(Objectother){

if((this==other))returntrue;

returnEqualsBuilder.reflectionEquals(this,other);

}

publicinthashCode(){

returnHashCodeBuilder.reflectionHashCode(this);

}

----------------------------

定义bean

packageaskyaya.entity;

importjava.util.Date;

publicclassVisitStatInfoBeanextendsBaseEntity{

/**

privatestaticfinallongserialVersionUID=1L;

Stringvisitor_ip;

Stringserver_ip;

Stringpage_url;

Stringparameter;

StringReferer_page;

intVisitor_type;

intserver_id;

intcolumn_id;

intpage_id;

intVisit_count;

intUser_id;

intProduct_id;

intSeller_id;

Dateinfo_date;

publicDategetInfo_date(){

returninfo_date;

}

publicvoidsetInfo_date(Dateinfo_date){

this.info_date=info_date;

}

publicintgetColumn_id(){

returncolumn_id;

}

publicvoidsetColumn_id(intcolumn_id){

this.column_id=column_id;

}

publicintgetPage_id(){

returnpage_id;

}

publicvoidsetPage_id(intpage_id){

this.page_id=page_id;

}

publicStringgetPage_url(){

returnpage_url;

}

publicvoidsetParameter(Stringparameter){

this.parameter=parameter;

}

publicStringgetParameter(){

returnparameter;

}

publicvoidsetPage_url(Stringparameter){

this.parameter=parameter;

}

publicintgetProduct_id(){

returnProduct_id;

}

publicvoidsetProduct_id(intproduct_id){

Product_id=product_id;

}

publicStringgetReferer_page(){

returnReferer_page;

}

publicvoidsetReferer_page(Stringreferer_page){

Referer_page=referer_page;

}

publicintgetSeller_id(){

returnSeller_id;

}

publicvoidsetSeller_id(intseller_id){

Seller_id=seller_id;

}

publicintgetServer_id(){

returnserver_id;

}

publicvoidsetServer_id(intserver_id){

this.server_id=server_id;

}

publicStringgetServer_ip(){

returnserver_ip;

}

publicvoidsetServer_ip(Stringserver_ip){

this.server_ip=server_ip;

}

publicintgetUser_id(){

returnUser_id;

}

publicvoidsetUser_id(intuser_id){

User_id=user_id;

}

publicintgetVisit_count(){

returnVisit_count;

}

publicvoidsetVisit_count(intvisit_count){

Visit_count=visit_count;

}

publicStringgetVisitor_ip(){

returnvisitor_ip;

}

publicvoidsetVisitor_ip(Stringvisitor_ip){

this.visitor_ip=visitor_ip;

}

publicintgetVisitor_type(){

returnVisitor_type;

}

publicvoidsetVisitor_type(intvisitor_type){

Visitor_type=visitor_type;

}

@Override

publicinthashCode(){

finalintPRIME=31;

intresult=super.hashCode();

result=PRIME*result+Product_id;

result=PRIME*result+((Referer_page==null)?0:Referer_page.hashCode());

result=PRIME*result+Seller_id;

result=PRIME*result+User_id;

result=PRIME*result+Visit_count;

result=PRIME*result+Visitor_type;

result=PRIME*result+column_id;

result=PRIME*result+((info_date==null)?0:info_date.hashCode());

result=PRIME*result+page_id;

result=PRIME*result+((page_url==null)?0:page_url.hashCode());

result=PRIME*result+server_id;

result=PRIME*result+((server_ip==null)?0:server_ip.hashCode());

result=PRIME*result+((visitor_ip==null)?0:visitor_ip.hashCode());

returnresult;

}

@Override

publicbooleanequals(Objectobj){

if(this==obj)

returntrue;

if(!super.equals(obj))

returnfalse;

if(getClass()!=obj.getClass())

returnfalse;

finalVisitStatInfoBeanother=(VisitStatInfoBean)obj;

if(Product_id!=other.Product_id)

returnfalse;

if(Referer_page==null){

if(other.Referer_page!=null)

returnfalse;

}elseif(!Referer_page.equals(other.Referer_page))

returnfalse;

if(Seller_id!=other.Seller_id)

returnfalse;

if(User_id!=other.User_id)

returnfalse;

if(Visit_count!=other.Visit_count)

returnfalse;

if(Visitor_type!=other.Visitor_type)

returnfalse;

if(column_id!=other.column_id)

returnfalse;

if(info_date==null){

if(other.info_date!=null)

returnfalse;

}elseif(!info_date.equals(other.info_date))

returnfalse;

if(page_id!=other.page_id)

returnfalse;

if(page_url==null){

if(other.page_url!=null)

returnfalse;

}elseif(!page_url.equals(other.page_url))

returnfalse;

if(server_id!=other.server_id)

returnfalse;

if(server_ip==null){

if(other.server_ip!=null)

returnfalse;

}elseif(!server_ip.equals(other.server_ip))

returnfalse;

if(visitor_ip==null){

if(other.visitor_ip!=null)

returnfalse;

}elseif(!visitor_ip.equals(other.visitor_ip))

returnfalse;

returntrue;

}

----------------------------

publicstaticvoidrecordVisit_JMS_Info(Stringvisitor_ip,Stringserver_ip,Stringpage_url,Stringparameter,StringReferer_page,

intVisitor_type,intserver_id,intcolumn_id,intpage_id,intVisit_count,intUser_id,

intProduct_id,intSeller_id)

{

//从页面上record_visit.jsp点击中,触发了这个方法

VisitStatInfoBeanvisitStatInfoBean=newVisitStatInfoBean();

visitStatInfoBean.setVisitor_ip(visitor_ip);

visitStatInfoBean.setServer_ip(server_ip);

visitStatInfoBean.setPage_url(page_url);

visitStatInfoBean.setReferer_page(Referer_page);

visitStatInfoBean.setVisitor_type(Visitor_type);

visitStatInfoBean.setServer_id(server_id);

visitStatInfoBean.setColumn_id(column_id);

visitStatInfoBean.setPage_id(page_id);

visitStatInfoBean.setVisit_count(Visit_count);

visitStatInfoBean.setUser_id(User_id);

visitStatInfoBean.setProduct_id(Product_id);

visitStatInfoBean.setSeller_id(Seller_id);

visitStatInfoBean.setInfo_date(newDate());

//VisitLogSenderproducer=newVisitLogSender();

try{

VisitLogSendervisitLogSender=VisitLogSender.getVisitLogSender();

visitLogSender.sendMessage(visitStatInfoBean);

//producer.close();

}catch(Exceptione){

e.printStackTrace();

//TODO:handleexception

}

..............

关键点是下面的

VisitLogSendervisitLogSender=VisitLogSender.getVisitLogSender();//单例

visitLogSender.sendMessage(visitStatInfoBean);//发送消息

..........

------------------------

packageaskyaya.jms;

importjavax.jms.Connection;

importjavax.jms.DeliveryMode;

importjavax.jms.Destination;

importjavax.jms.JMSException;

importjavax.jms.MessageProducer;

importjavax.jms.ObjectMessage;

importjavax.jms.Session;

importorg.apache.activemq.ActiveMQConnectionFactory;

importaskyaya.entity.VisitStatInfoBean;

importaskyaya.visitlog.GetJmsUrl;

/**

*@author作者Administrator:金果

*@version创建时间：

*类说明:

publicclassVisitLogSender{

privatestaticStringsubject="askyaya.visit.log";

privatestaticDestinationdestination=null;//消息的目的地

privatestaticConnectionconnection=null;//JMS客户端到JMSProvider的连接

privatestaticSessionsession=null;//一个发送或接收消息的线程

privatestaticMessageProducerproducer=null;//由Session对象创建的用来发送消息的对象

privatestaticVisitLogSendervisitLogSender=null;

publicstaticVisitLogSendergetVisitLogSender(){

if(visitLogSender==null){

visitLogSender=newVisitLogSender();

try{

visitLogSender.initialize();

}catch(Exceptione)

{e.printStackTrace();}

}

returnvisitLogSender;

}

publicvoidinitialize()throwsJMSException,Exception{

System.out.println("url----------->"+GetJmsUrl.getUrl());

ActiveMQConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("","",GetJmsUrl.getUrl());

connection=connectionFactory.createConnection();

connection.start();

session=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);//不用事务了,session以自动方式接收,通用性好

destination=session.createQueue(subject);

producer=session.createProducer(destination);

producer.setDeliveryMode(DeliveryMode.PERSISTENT);//以持久的方式到队列oracle数据库

}

//发送消息

publicvoidsendMessage(VisitStatInfoBeanmessage)throwsJMSException,Exception{

ObjectMessagemsg=session.createObjectMessage();

msg.setObject(message);

System.out.println("Producer:->Sendingaskyaya_2007message:"+message);

System.out.println("Producer:->Sendingaskyaya_2007msg:"+msg);

producer.send(msg);

System.out.println("Producer:->askyaya_2007Messagesentcomplete!");

}

//关闭连接

publicvoidclose()throwsJMSException{

System.out.println("Producer:->Closingconnection");

if(producer!=null)

producer.close();

if(session!=null)

session.close();

if(connection!=null)

connection.close();

}

----------------------------------------------

开始收了啊

我是另外建了工程

packageaskyaya.jms;

importjavax.jms.Connection;

importjavax.jms.Destination;

importjavax.jms.JMSException;

importjavax.jms.Message;

importjavax.jms.MessageConsumer;

importjavax.jms.MessageListener;

importjavax.jms.ObjectMessage;

importjavax.jms.Session;

importorg.apache.activemq.ActiveMQConnection;

importorg.apache.activemq.ActiveMQConnectionFactory;

importaskyaya.dao.VisitStatInfoDao;

importaskyaya.entity.VisitStatInfoBean;

importaskyaya.util.GetJmsUrl;

/**

*@author作者Administrator:jinguo

*@version创建时间：2008-8-30上午10:55:47

*类说明:

publicclassReceiveVisitLogimplementsMessageListener{

//privateStringuser=ActiveMQConnection.DEFAULT_USER;

//privateStringpassword=ActiveMQConnection.DEFAULT_PASSWORD;

//privateStringurl=ActiveMQConnection.DEFAULT_BROKER_URL;

privatestaticStringsubject="askyaya.visit.log";

privatestaticDestinationdestination=null;

privatestaticConnectionconnection=null;

privatestaticSessionsession=null;

privatestaticMessageConsumerconsumer=null;

//privateMessageProducerreplyProducer;//反馈信息

//连接connection，getsession

privatestaticReceiveVisitLogreceivevisitlog=null;

publicstaticReceiveVisitLoggetReceiveVisitLog(){

if(receivevisitlog==null){

receivevisitlog=newReceiveVisitLog();

try{

initialize();

}catch(Exceptione){

e.printStackTrace();

//TODO:handleexception

}

returnreceivevisitlog;

}

privatestaticvoidinitialize()throwsJMSException{

GetJmsUrlurl=newGetJmsUrl();

//ActiveMQConnectionFactoryconnectionFactory=newActiveMQConnectionFactory(user,password,url);

System.out.println("url---M:"+url.getUrl());

ActiveMQConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("","",url.getUrl());

System.out.println("connecttotcp:192.168.0.245");

Connectionconnection=connectionFactory.createConnection();

connection.start();

session=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

destination=session.createQueue(subject);

//DestinationadvisoryDestination=AdvisorySupport.getProducerAdvisoryTopic(destination)

consumer=session.createConsumer(destination);

}

//消费消息

publicvoidconsumeMessage()throwsJMSException,Exception{

//开始监听

//MessageListenerForOrgMsgmsgListener=newMessageListenerForOrgMsg();

consumer.setMessageListener(this);

System.out.println("Consumer:->afterlistenermessage...");

//如果想要去想主动的去接受消息,起用下面的

//Messagemessage=consumer.receiveNoWait();

//TextMessagemessage=(TextMessage)consumer.receive(1000);

}

//消息处理函数(onMessage是个构造函数继承,consumerTool开始执行的时候,先要初始化它,然后才initialze())

//由container选择一个实例

publicvoidonMessage(Messagemessage){

try{

System.out.println("Consumer:->Beginreceivemessage--->:"+message);

if(message!=null){

if(messageinstanceofObjectMessage){

ObjectMessageobjmsg=(ObjectMessage)message;

//从消息中提取对象,转化为bean对象

VisitStatInfoBeanvisitinfo=(VisitStatInfoBean)objmsg.getObject();

//隔离对数据库的直接访问,

VisitStatInfoDaovistdao=newVisitStatInfoDao();

//向数据库中插入消息

vistdao.insert(visitinfo);

System.out.println("Consumer:->afterinsertmessagedb");

}

//消费消息中在onMessage()方法中接收producer发送过来的消息进行处理，并可以通过replyProducer反馈信息给producer

/*if(message.getJMSReplyTo()!=null){

replyProducer.send(message.getJMSReplyTo(),session.createTextMessage("Reply:"+message.getJMSMessageID()));

}*/

}catch(Exceptione){

e.printStackTrace();

}

//关闭连接

publicvoidclose()throwsJMSException{

System.out.println("Consumer:->Closingconnection");

if(consumer!=null)

consumer.close();

if(session!=null)

session.close();

if(connection!=null)

connection.close();

}

publicstaticvoidmain(String[]args){

//取得接收信息

ReceiveVisitLogreceiveVisitLog=ReceiveVisitLog.getReceiveVisitLog();

GetJmsUrlurl=newGetJmsUrl();

//接收信息时间间隔

//intsleepTime=30*1000;

intsleepTime=Integer.parseInt(url.getSleepTime());

while(true){

try{

System.out.println("睡眠"+sleepTime/1000+"秒之后,开始接收消息......");

receiveVisitLog.consumeMessage();

Thread.sleep(sleepTime);

System.out.println("接收消息完毕.");

}catch(Exceptione){

e.printStackTrace();

}

--------------------------------

为了方便大家我把dao方法也贴了出来

packageaskyaya.dao;

importjava.sql.Connection;

importjava.sql.PreparedStatement;

importjava.sql.ResultSet;

importjava.sql.SQLException;

importjava.sql.Statement;

importaskyaya.entity.VisitStatInfoBean;

publicclassVisitStatInfoDao{

publicvoidinsert(VisitStatInfoBeanvfBean)throwsSQLException{

ConnectionApplicationConn=askyaya.util.DBManager.getconn();

Stringstrsql="insertintotbl_visit_stat_info(visitor_ip,server_id,server_ip,column_id,"

+"page_id,page_url,parameter,Visit_count,User_id,Visitor_type,Product_id,Seller_id,"

+"Info_date,visit_date,Referer_page)values(?,?,?,?,?,?,?,?,?,?,?,?,Sysdate,to_char(Sysdate,'yyyy-mm-dd'),?)";

PreparedStatementpstmt=null;

ApplicationConn.setAutoCommit(false);

try{

pstmt=ApplicationConn.prepareStatement(strsql);

//VisitStatInfoBeanvfBean=newVisitStatInfoBean();

pstmt.setString(1,vfBean.getVisitor_ip());

pstmt.setInt(2,vfBean.getServer_id());

pstmt.setString(3,vfBean.getServer_ip());

pstmt.setInt(4,vfBean.getColumn_id());

pstmt.setInt(5,vfBean.getPage_id());

pstmt.setString(6,vfBean.getPage_url());

pstmt.setString(7,vfBean.getParameter());

pstmt.setInt(8,vfBean.getVisit_count());

pstmt.setInt(9,vfBean.getUser_id());

pstmt.setInt(10,vfBean.getVisitor_type());

pstmt.setInt(11,vfBean.getProduct_id());

pstmt.setInt(12,vfBean.getSeller_id());

pstmt.setString(13,vfBean.getReferer_page());

pstmt.executeUpdate();

ApplicationConn.commit();

//System.out.println("insertthetbl_visit_stat_infoisend");

}catch(SQLExceptionsqle){

interrcode=sqle.getErrorCode();

//System.err.println("aq.executeQuery:"+sqle.getMessage()+":;");

//违反唯一约束

if(errcode==1){}

else{

//System.err.println("aq.executeQuery:"+sqle.getMessage()+":;");

}

ApplicationConn.rollback();

}finally{

try{

pstmt.close();

ApplicationConn.close();

}catch(Exceptionex){

ex.printStackTrace();

}

---------------------------------

得到地址的通用方法

packageaskyaya.util;

importjava.io.InputStream;

importjava.util.Properties;

publicclassGetJmsUrl{

privatefinalstaticStringstrUrl=null;

privatefinalstaticStringsleeptime=null;

privatefinalPropertiesprops=newProperties();

privatefinalStringurl="askyayaJmsConstant.properties";

publicGetJmsUrl(){

}

publicStringgetUrl(){

InputStreamis=getClass().getResourceAsStream(url);

try{

if(strUrl==null){

props.load(is);

}

}catch(Exceptione){

System.err.println("CannotreadthePropertiesfile"+url);

}

StringstrUrl=props.getProperty("hostname");

returnstrUrl;

}

publicStringgetSleepTime(){

InputStreamis=getClass().getResourceAsStream(url);

try{

if(sleeptime==null){

props.load(is);

}

}catch(Exceptione){

e.printStackTrace();

//TODO:handleexception

}

StringstrSleepTime=props.getProperty("sleepTime");

returnstrSleepTime;

}

------------------------

askyayaJmsConstant.properties

设置如下:

hostname=tcp://192.168.0.245:61616?wireFormat.maxInactivityDuration=0

sleepTime=30000

---------------

我把服务器的activemq.xml配置也贴出来把,很关键的哟,不懂可以问我.

<!--

LicensedtotheApacheSoftwareFoundation(ASF)underoneormore

contributorlicenseagreements.SeetheNOTICEfiledistributedwith

thisworkforadditionalinformationregardingcopyrightownership.

TheASFlicensesthisfiletoYouundertheApacheLicense,Version2.0

(the"License");youmaynotusethisfileexceptincompliancewith

theLicense.YoumayobtainacopyoftheLicenseat

http://www.apache.org/licenses/LICENSE-2.0

Unlessrequiredbyapplicablelaworagreedtoinwriting,software

distributedundertheLicenseisdistributedonan"ASIS"BASIS,

WITHOUTWARRANTIESORCONDITIONSOFANYKIND,eitherexpressorimplied.

SeetheLicenseforthespecificlanguagegoverningpermissionsand

limitationsundertheLicense.

-->

<beans

xmlns="http://www.springframework.org/schema/beans"

xmlns:amq="http://activemq.apache.org/schema/core"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans-2.0.xsd

http://activemq.apache.org/schema/corehttp://activemq.apache.org/schema/core/activemq-core.xsd

http://activemq.apache.org/camel/schema/springhttp://activemq.apache.org/camel/schema/spring/camel-spring.xsd">

<beanclass="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>

<brokerxmlns="http://activemq.apache.org/schema/core"brokername="localhost"dataDirectory="${activemq.base}/data">

<policyEntryqueue=">"memoryLimit="5mb"/>

<policyEntrytopic=">"memoryLimit="5mb">

</dispatchPolicy>

</subscriptionRecoveryPolicy>

</policyEntry>

</policyEntries>

</policyMap>

</destinationPolicy>

<managementContextcreateConnector="false"/>

</managementContext>

<networkConnectorname="default-nc"uri="multicast://default"/>

<!--Exampleofastaticconfiguration:

<networkConnectorname="host1andhost2"uri="static://(tcp://host1:61616,tcp://host2:61616)"/>

-->

</networkConnectors>

<!--

<amqPersistenceAdaptersyncOnWrite="false"directory="${activemq.base}/data"maxFileLength="20mb"/>

</persistenceAdapter>

-->

<!--

<journaledJDBCdataDirectory="${activemq.base}/data"dataSource="#postgres-ds"/>

</persistenceAdapter>

-->

<jdbcPersistenceAdapterdataSource="#oracle-ds"/>

</persistenceAdapter>

<memoryUsagelimit="20mb"/>

</memoryUsage>

<storeUsagelimit="1gb"name="foo"/>

</storeUsage>

<tempUsagelimit="100mb"/>

</tempUsage>

</systemUsage>

<transportConnectorname="openwire"uri="tcp://192.168.0.245:61616?wireFormat.maxInactivityDuration=0"discoveryUri="multicast://default"/>

<transportConnectorname="ssl"uri="ssl://192.168.0.245:61617"/>

<transportConnectorname="stomp"uri="stomp://192.168.0.245:61613"/>

<transportConnectorname="xmpp"uri="xmpp://192.168.0.245:61222"/>

</transportConnectors>

</broker>

<!--

**LetsdeploysomeEnterpriseIntegrationPatternsinsidetheActiveMQMessageBroker

**Formoredetailssee

**http://activemq.apache.org/enterprise-integration-patterns.html

-->

<camelContextid="camel"xmlns="http://activemq.apache.org/camel/schema/spring">

<route>

<fromuri="activemq:example.A"/>

<touri="activemq:example.B"/>

</route>

</camelContext>

<!--

<commandAgentxmlns="http://activemq.apache.org/schema/core"brokerUrl="vm://localhost"/>

-->

<jettyxmlns="http://mortbay.com/schemas/jetty/1.0">

<nioConnectorport="8161"/>

</connectors>

<webAppContextcontextPath="/admin"resourceBase="${activemq.base}/webapps/admin"logUrlOnStart="true"/>

<webAppContextcontextPath="/demo"resourceBase="${activemq.base}/webapps/demo"logUrlOnStart="true"/>

<webAppContextcontextPath="/fileserver"resourceBase="${activemq.base}/webapps/fileserver"logUrlOnStart="true"/>

</handlers>

</jetty>

<!--

<beanid="postgres-ds"class="org.postgresql.ds.PGPoolingDataSource">

<propertyname="serverName"value="localhost"/>

<propertyname="databaseName"value="activemq"/>

<propertyname="portNumber"value="0"/>

<propertyname="user"value="activemq"/>

<propertyname="password"value="activemq"/>

<propertyname="dataSourceName"value="postgres"/>

<propertyname="initialConnections"value="1"/>

<propertyname="maxConnections"value="10"/>

</bean>

-->

<!--

<beanid="mysql-ds"class="org.apache.commons.dbcp.BasicDataSource"destroy-method="close">

<propertyname="driverClassName"value="com.mysql.jdbc.Driver"/>

<propertyname="url"value="jdbc:mysql://localhost/activemq?relaxAutoCommit=true"/>

<propertyname="username"value="activemq"/>

<propertyname="password"value="activemq"/>

<propertyname="maxActive"value="200"/>

<propertyname="poolPreparedStatements"value="true"/>

</bean>

-->

<beanid="oracle-ds"class="org.apache.commons.dbcp.BasicDataSource"destroy-method="close">

<propertyname="driverClassName"value="oracle.jdbc.driver.OracleDriver"/>

<propertyname="url"value="jdbc:oracle:thin:@192.168.0.245:1521:askyaya"/>

<propertyname="username"value="activemq"/>

<propertyname="password"value="activemq"/>

<propertyname="maxActive"value="200"/>

<propertyname="poolPreparedStatements"value="true"/>

</bean>

<!--

<beanid="derby-ds"class="org.apache.derby.jdbc.EmbeddedDataSource">

<propertyname="databaseName"value="derbydb"/>

<propertyname="createDatabase"value="create"/>

</bean>

-->

</beans>

----------------------

以上我用jdbc来写的,速度非常快,也可以hibernate实现如果有大虾知道的话贴出来,

最后吗,有些人问我,什么样的人适合做程序员,告诉大家

就是突破能力一定要强,想锻炼这能力,多看看书哟.以上实现我只用一个星期就基本就解决了jms大部分难点.

基于jms使用ActiveMQ实现异步日志功能.消息持久到oracle 10g 数据库

相关推荐