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

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

数据库  /  管理员 发布于 2020-03-08 13:33:15   23

package askyaya.entity;
import java.io.Serializable;
import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.commons.lang.builder.ReflectionToStringBuilder;
import org.apache.commons.lang.builder.ToStringBuilder;
public class BaseEntity  implements Serializable {
/**
    
  public String toString() {
   return ReflectionToStringBuilder.toString(this);
       }

       public boolean equals(Object other) {
           if ( (this == other ) ) return true;
           return EqualsBuilder.reflectionEquals(this, other);
       }

       public int hashCode() {
        return HashCodeBuilder.reflectionHashCode(this);
       }
}

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

定义bean

package askyaya.entity;

import java.util.Date;

public class VisitStatInfoBean extends BaseEntity {
/**
  *
  */
private static final long serialVersionUID = 1L;
String visitor_ip;
String server_ip;
String page_url;
String parameter;
String Referer_page;
int Visitor_type;
int server_id;
int column_id;
int page_id;
int Visit_count;
int User_id;
int Product_id;
int Seller_id;
Date info_date;
public Date getInfo_date() {
  return info_date;
}
public void setInfo_date(Date info_date) {
  this.info_date = info_date;
}
public int getColumn_id() {
  return column_id;
}
public void setColumn_id(int column_id) {
  this.column_id = column_id;
}
public int getPage_id() {
  return page_id;
}
public void setPage_id(int page_id) {
  this.page_id = page_id;
}
public String getPage_url() {
  return page_url;
}
public void setParameter(String parameter) {
  this.parameter = parameter;
}
public String getParameter() {
  return parameter;
}
public void setPage_url(String parameter) {
  this.parameter = parameter;
}

public int getProduct_id() {
  return Product_id;
}
public void setProduct_id(int product_id) {
  Product_id = product_id;
}
public String getReferer_page() {
  return Referer_page;
}
public void setReferer_page(String referer_page) {
  Referer_page = referer_page;
}
public int getSeller_id() {
  return Seller_id;
}
public void setSeller_id(int seller_id) {
  Seller_id = seller_id;
}
public int getServer_id() {
  return server_id;
}
public void setServer_id(int server_id) {
  this.server_id = server_id;
}
public String getServer_ip() {
  return server_ip;
}
public void setServer_ip(String server_ip) {
  this.server_ip = server_ip;
}
public int getUser_id() {
  return User_id;
}
public void setUser_id(int user_id) {
  User_id = user_id;
}
public int getVisit_count() {
  return Visit_count;
}
public void setVisit_count(int visit_count) {
  Visit_count = visit_count;
}
public String getVisitor_ip() {
  return visitor_ip;
}
public void setVisitor_ip(String visitor_ip) {
  this.visitor_ip = visitor_ip;
}
public int getVisitor_type() {
  return Visitor_type;
}
public void setVisitor_type(int visitor_type) {
  Visitor_type = visitor_type;
}
@Override
public int hashCode() {
  final int PRIME = 31;
  int result = 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());
  return result;
}
@Override
public boolean equals(Object obj) {
  if (this == obj)
   return true;
  if (!super.equals(obj))
   return false;
  if (getClass() != obj.getClass())
   return false;
  final VisitStatInfoBean other = (VisitStatInfoBean) obj;
  if (Product_id != other.Product_id)
   return false;
  if (Referer_page == null) {
   if (other.Referer_page != null)
    return false;
  } else if (!Referer_page.equals(other.Referer_page))
   return false;
  if (Seller_id != other.Seller_id)
   return false;
  if (User_id != other.User_id)
   return false;
  if (Visit_count != other.Visit_count)
   return false;
  if (Visitor_type != other.Visitor_type)
   return false;
  if (column_id != other.column_id)
   return false;
  if (info_date == null) {
   if (other.info_date != null)
    return false;
  } else if (!info_date.equals(other.info_date))
   return false;
  if (page_id != other.page_id)
   return false;
  if (page_url == null) {
   if (other.page_url != null)
    return false;
  } else if (!page_url.equals(other.page_url))
   return false;
  if (server_id != other.server_id)
   return false;
  if (server_ip == null) {
   if (other.server_ip != null)
    return false;
  } else if (!server_ip.equals(other.server_ip))
   return false;
  if (visitor_ip == null) {
   if (other.visitor_ip != null)
    return false;
  } else if (!visitor_ip.equals(other.visitor_ip))
   return false;
  return true;
}
}
----------------------------



public static void recordVisit_JMS_Info(String visitor_ip,String server_ip,String page_url, String parameter, String Referer_page,
       int Visitor_type,int server_id,int column_id,int page_id,int Visit_count,int User_id,
       int Product_id,int Seller_id)
   {   
  //从页面上record_visit.jsp 点击中,触发了这个方法    
   VisitStatInfoBean visitStatInfoBean=new VisitStatInfoBean();
   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(new Date());     
     //VisitLogSender producer=new VisitLogSender();
  
   try {
    VisitLogSender visitLogSender = VisitLogSender.getVisitLogSender();
   
    visitLogSender.sendMessage(visitStatInfoBean);
    //producer.close();
   } catch (Exception e) {
    e.printStackTrace();
    // TODO: handle exception
   }

  
     }



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

  关键点是下面的

  VisitLogSender visitLogSender = VisitLogSender.getVisitLogSender(); //单例
   
    visitLogSender.sendMessage(visitStatInfoBean); //发送消息

..........

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

package askyaya.jms;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

import askyaya.entity.VisitStatInfoBean;
import askyaya.visitlog.GetJmsUrl;


/**
* @author 作者 Administrator:金果
* @version 创建时间：
* 类说明:
*/
public class VisitLogSender {

    private static  String subject = "askyaya.visit.log"; 
   
    private static Destination destination = null;   //消息的目的地
    
    private static  Connection connection = null;   //JMS 客户端到JMS Provider 的连接
   
    private static Session session = null;    //一个发送或接收消息的线程
   
    private static  MessageProducer producer = null;  //由Session 对象创建的用来发送消息的对象
  
    private static VisitLogSender visitLogSender = null;
   
    public static VisitLogSender getVisitLogSender(){
      if(visitLogSender == null){
       visitLogSender = new VisitLogSender();      
       try{
        visitLogSender.initialize();
       }catch(Exception e)
       {e.printStackTrace();}
      }     
      return visitLogSender;
     }
    
  public  void   initialize() throws JMSException, Exception {
     
      System.out.println("url----------->"+GetJmsUrl.getUrl());
       
        ActiveMQConnectionFactory connectionFactory=new ActiveMQConnectionFactory("","",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 数据库
        
     } 
   
   
   
     //发送消息  
     public void sendMessage(VisitStatInfoBean message) throws JMSException,Exception {
           
         ObjectMessage msg=session.createObjectMessage();        
         msg.setObject(message);                   
         System.out.println("Producer:->Sending askyaya_2007 message:"+message);
         System.out.println("Producer:->Sending askyaya_2007 msg:"+msg);        
         producer.send(msg);
         System.out.println("Producer:->askyaya_2007 Message sent complete!");
     
     }   
   
     // 关闭连接 
    public void close() throws JMSException {   
         System.out.println("Producer:->Closing connection");   
         if (producer != null)   
             producer.close();   
         if (session != null)   
             session.close();   
         if (connection != null)  
             connection.close();   
     }  

}

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

开始收了啊

我是另外建了工程



package askyaya.jms;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session;

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

import askyaya.dao.VisitStatInfoDao;
import askyaya.entity.VisitStatInfoBean;
import askyaya.util.GetJmsUrl;

/**
* @author 作者 Administrator:jinguo
* @version 创建时间：2008-8-30 上午10:55:47
* 类说明:
*/
public class ReceiveVisitLog implements MessageListener{

  // private String user =ActiveMQConnection.DEFAULT_USER;   
   
  // private String password=ActiveMQConnection.DEFAULT_PASSWORD;   
  
   // private String url=ActiveMQConnection.DEFAULT_BROKER_URL;   
  
    private static String subject = "askyaya.visit.log";   
  
    private static Destination destination = null;   
  
    private static Connection connection = null;   
  
    private static  Session session = null;   
  
    private static MessageConsumer consumer = null;
   
  
   
  //  private MessageProducer replyProducer; //反馈信息
   
    //连接connection，get session
     private static ReceiveVisitLog receivevisitlog=null;
    
     public static ReceiveVisitLog getReceiveVisitLog(){
      if(receivevisitlog==null){
       receivevisitlog=new  ReceiveVisitLog();
       try {
       initialize();
   } catch (Exception e) {
    e.printStackTrace();
    // TODO: handle exception
   }      
      }
      return receivevisitlog;     
     }
   
   
    private static  void initialize() throws JMSException{
    
      GetJmsUrl url=new GetJmsUrl();
      //ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,password,url);
     System.out.println("url---M:"+url.getUrl());
    
      
      ActiveMQConnectionFactory connectionFactory=new ActiveMQConnectionFactory("","",url.getUrl());
         System.out.println("connect to  tcp:192.168.0.245 ");
         Connection connection=connectionFactory.createConnection();
         connection.start();
        
         session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);        
         destination=session.createQueue(subject);
     //  Destination advisoryDestination = AdvisorySupport.getProducerAdvisoryTopic(destination)
         consumer=session.createConsumer(destination);
    
    }
   
    // 消费消息   
    public void consumeMessage() throws JMSException,Exception {
  
       //开始监听       
       // MessageListenerForOrgMsg msgListener=new MessageListenerForOrgMsg();  
       consumer.setMessageListener(this);
      
       System.out.println("Consumer:->after listener message...");
      
      //如果想要去想主动的去接受消息,起用下面的       
      //Message message=consumer.receiveNoWait();       
      //TextMessage message=(TextMessage)consumer.receive(1000);
      
  }   
     
 
    // 消息处理函数(onMessage是个构造函数继承,consumerTool开始执行的时候,先要初始化它,然后才initialze())
    //由 container 选择一个实例
public void onMessage(Message message){
  try {  
 
     System.out.println("Consumer:->Begin receive message--->:"+message);
       if(message!=null){
      if(message instanceof ObjectMessage){
      ObjectMessage  objmsg=(ObjectMessage)message;  
      //从消息中提取对象,转化为bean对象
       VisitStatInfoBean visitinfo=(VisitStatInfoBean)objmsg.getObject();
    
      //隔离对数据库的直接访问,
      VisitStatInfoDao vistdao=new VisitStatInfoDao();
      //向数据库中插入消息
      vistdao.insert(visitinfo);
      System.out.println("Consumer:->after insert messagedb");
     }
       }
   
     
   //消费消息中在onMessage()方法中接收producer发送过来的消息进行处理，并可以通过replyProducer反馈信息给producer
   /*if(message.getJMSReplyTo()!=null){   
     replyProducer.send(message.getJMSReplyTo(),session.createTextMessage("Reply: "+message.getJMSMessageID()));  
   }*/
  
  } catch (Exception e) {
   e.printStackTrace();
  
  }
 
}

  // 关闭连接   
    public void close() throws JMSException {   
        System.out.println("Consumer:->Closing connection");   
        if (consumer != null)   
            consumer.close();   
        if (session != null)   
            session.close();   
        if (connection != null)  
            connection.close();   
    }   
   
   
    public static void main(String[] args) {

  //取得接收信息
  ReceiveVisitLog receiveVisitLog = ReceiveVisitLog.getReceiveVisitLog();
   GetJmsUrl url=new GetJmsUrl();
  
  //接收信息时间间隔 
  //int sleepTime = 30*1000;
  int sleepTime=Integer.parseInt(url.getSleepTime());
 
  while (true){                       
         try {
          System.out.println("睡眠" + sleepTime/1000 + "秒之后, 开始接收消息......");
          receiveVisitLog.consumeMessage();

             Thread.sleep(sleepTime);
             System.out.println("接收消息完毕.");
            
   } catch (Exception e) {
    e.printStackTrace();
   
   } 
  }

}
   
   
}

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

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

package askyaya.dao;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

import askyaya.entity.VisitStatInfoBean;

public class VisitStatInfoDao {
  
public  void insert(VisitStatInfoBean vfBean)throws SQLException{
 
   Connection ApplicationConn=askyaya.util.DBManager.getconn();
    String strsql="insert into tbl_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'),?)";
 
    PreparedStatement pstmt = null;
    ApplicationConn.setAutoCommit(false);
    try {  
     pstmt = ApplicationConn.prepareStatement(strsql);          
   //VisitStatInfoBean vfBean=new VisitStatInfoBean();
    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("insert the tbl_visit_stat_info is end ");
         } catch (SQLException sqle) {
         int errcode = 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 (Exception ex) {
      ex.printStackTrace();
     }
    }

}
}
---------------------------------

得到地址的通用方法

package askyaya.util;

import java.io.InputStream;
import java.util.Properties;



public   class  GetJmsUrl {
   
private final static  String strUrl=null;
private final static String sleeptime=null;
 
private final    Properties props=new Properties();

private final String url="askyayaJmsConstant.properties";
 
public GetJmsUrl(){
   
} 

public  String getUrl(){ 
  InputStream is=getClass().getResourceAsStream(url); 
     try {
      if(strUrl==null){
      props.load(is);
      }
  } catch (Exception e) {
   System.err.println("Can not read the Properties file " + url);  
  }       
    String strUrl= props.getProperty("hostname");        
  return strUrl;
 
}

public   String  getSleepTime(){ 
  InputStream is=getClass().getResourceAsStream(url);
  try {
   if(sleeptime==null){
        props.load(is);       
   } 
  } catch (Exception e) {
   e.printStackTrace();
   // TODO: handle exception
  }
  String   strSleepTime= props.getProperty("sleepTime");
 
  return strSleepTime;
 
}


}
------------------------

askyayaJmsConstant.properties

设置如下:



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

sleepTime=30000

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

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



<!--
    Licensed to the Apache Software Foundation (ASF) under one or more
    contributor license agreements.  See the NOTICE file distributed with
    this work for additional information regarding copyright ownership.
    The ASF licenses this file to You under the Apache License, Version 2.0
    (the "License"); you may not use this file except in compliance with
    the License.  You may obtain a copy of the License at
  
    http://www.apache.org/licenses/LICENSE-2.0
  
    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS O