Alawn/udf
1
0
Fork 0
mirror of https://gitee.com/wangkang/udf.git synced 2026-10-28 22:05:14 +00:00
Code Issues Releases Wiki Activity

添加消息中间件日志记录

Browse Source
This commit is contained in:
udf_admin@163.com
2017-12-10 15:23:51 +08:00
parent 879025944a
commit 54d95210a6
8 changed files with 388 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/MessageLog.java
+165
View File
@@ -0,0 +1,165 @@
package org.itkk.udf.amqp.rabbitmq;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.itkk.udf.amqp.rabbitmq.domain.CompleteLog;
import org.itkk.udf.amqp.rabbitmq.domain.ErrorLog;
import org.itkk.udf.amqp.rabbitmq.domain.ReceiveLog;
import org.itkk.udf.amqp.rabbitmq.domain.SendLog;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import java.util.Date;
import java.util.UUID;
/**
* MessageLog
*/
@Component
@Slf4j
public class MessageLog {
/**
* 描述 : 应用名称
*/
@Value("${spring.application.name}")
private String springApplicationName;
/**
* 描述 : amqpTemplate
*/
@Autowired
private AmqpTemplate amqpTemplate;
/**
* objectMapper
*/
@Autowired
private ObjectMapper objectMapper;
/**
* send
*
* @param <T> 泛型
* @param message message
*/
@Async
protected <T> void send(RabbitmqMessage<T> message) {
//将消息转换为json格式
String msgJsonString = null;
try {
msgJsonString = objectMapper.writeValueAsString(message);
} catch (JsonProcessingException e) {
log.error("MessageLog.msgJsonString convert msg to string error : ", e);
}
//构造实体
SendLog log = new SendLog();
log.setLogId(UUID.randomUUID().toString());
log.setId(message.getId());
log.setSender(message.getSender());
log.setExchange(message.getExchange());
log.setRoutingKey(message.getRoutingKey());
log.setTimestamp(message.getTimestamp());
log.setMsgCreateDate(message.getCreateDate());
log.setSendDate(message.getSendDate());
log.setMsgJsonString(msgJsonString);
//构造消息
RabbitmqMessage<SendLog> logMessage = new RabbitmqMessage<>();
logMessage.setSender(springApplicationName);
logMessage.setSendDate(new Date());
logMessage.setExchange(RabbitmqConstant.EXCHANGE_MESSAGE_LOG.class.getSimpleName());
logMessage.setRoutingKey(RabbitmqConstant.EXCHANGE_MESSAGE_LOG.SEND.name());
logMessage.setTimestamp(System.currentTimeMillis());
logMessage.setBody(log);
//发送消息
this.amqpTemplate.convertAndSend(RabbitmqConstant.EXCHANGE_MESSAGE_LOG.class.getSimpleName(), RabbitmqConstant.EXCHANGE_MESSAGE_LOG.SEND.name(), logMessage);
}
/**
* receive
*
* @param msgId msgId
* @param queues queues
* @param receiveDate receiveDate
*/
@Async
protected void receive(String msgId, String queues, Date receiveDate) {
//构造实体
ReceiveLog log = new ReceiveLog();
log.setLogId(UUID.randomUUID().toString());
log.setId(msgId);
log.setQueues(queues);
log.setReceiveDate(receiveDate);
//构造消息
RabbitmqMessage<ReceiveLog> logMessage = new RabbitmqMessage<>();
logMessage.setSender(springApplicationName);
logMessage.setSendDate(new Date());
logMessage.setExchange(RabbitmqConstant.EXCHANGE_MESSAGE_LOG.class.getSimpleName());
logMessage.setRoutingKey(RabbitmqConstant.EXCHANGE_MESSAGE_LOG.RECEIVE.name());
logMessage.setTimestamp(System.currentTimeMillis());
logMessage.setBody(log);
//发送消息
this.amqpTemplate.convertAndSend(RabbitmqConstant.EXCHANGE_MESSAGE_LOG.class.getSimpleName(), RabbitmqConstant.EXCHANGE_MESSAGE_LOG.RECEIVE.name(), logMessage);
}
/**
* complete
*
* @param msgId msgId
* @param queues queues
* @param completeDate completeDate
*/
@Async
protected void complete(String msgId, String queues, Date completeDate) {
//构造实体
CompleteLog log = new CompleteLog();
log.setLogId(UUID.randomUUID().toString());
log.setId(msgId);
log.setQueues(queues);
log.setCompleteDate(completeDate);
//构造消息
RabbitmqMessage<CompleteLog> logMessage = new RabbitmqMessage<>();
logMessage.setSender(springApplicationName);
logMessage.setSendDate(new Date());
logMessage.setExchange(RabbitmqConstant.EXCHANGE_MESSAGE_LOG.class.getSimpleName());
logMessage.setRoutingKey(RabbitmqConstant.EXCHANGE_MESSAGE_LOG.COMPLETE.name());
logMessage.setTimestamp(System.currentTimeMillis());
logMessage.setBody(log);
//发送消息
this.amqpTemplate.convertAndSend(RabbitmqConstant.EXCHANGE_MESSAGE_LOG.class.getSimpleName(), RabbitmqConstant.EXCHANGE_MESSAGE_LOG.COMPLETE.name(), logMessage);
}
/**
* error
*
* @param msgId msgId
* @param queues queues
* @param errorDate errorDate
* @param ex ex
*/
@Async
protected void error(String msgId, String queues, Date errorDate, Exception ex) {
//构造实体
ErrorLog log = new ErrorLog();
log.setLogId(UUID.randomUUID().toString());
log.setId(msgId);
log.setQueues(queues);
log.setErrorDate(errorDate);
log.setErrorMsg(ExceptionUtils.getStackTrace(ex));
//构造消息
RabbitmqMessage<ErrorLog> logMessage = new RabbitmqMessage<>();
logMessage.setSender(springApplicationName);
logMessage.setSendDate(new Date());
logMessage.setExchange(RabbitmqConstant.EXCHANGE_MESSAGE_LOG.class.getSimpleName());
logMessage.setRoutingKey(RabbitmqConstant.EXCHANGE_MESSAGE_LOG.ERROR.name());
logMessage.setTimestamp(System.currentTimeMillis());
logMessage.setBody(log);
//发送消息
this.amqpTemplate.convertAndSend(RabbitmqConstant.EXCHANGE_MESSAGE_LOG.class.getSimpleName(), RabbitmqConstant.EXCHANGE_MESSAGE_LOG.ERROR.name(), logMessage);
}
}
udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/Rabbitmq.java
+7
View File
@@ -33,6 +33,12 @@ public class Rabbitmq {
@Autowired
private AmqpTemplate amqpTemplate;
/**
* messageLog
*/
@Autowired
private MessageLog messageLog;
/**
* 描述 : Convert a Java object to an Amqp Message and send it to a specific exchange with a
* specific routing key.
@@ -49,6 +55,7 @@ public class Rabbitmq {
message.setRoutingKey(routingKey);
message.setTimestamp(System.currentTimeMillis());
this.amqpTemplate.convertAndSend(exchange, routingKey, message);
messageLog.send(message);
}
}
udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqConstant.java
+32 -2
View File
@@ -27,14 +27,44 @@ public class RabbitmqConstant {
* (EXCHANGE)消息记录
*/
public enum EXCHANGE_MESSAGE_LOG {
/**
* 发送消息
*/
SEND,
/**
* 接收消息
*/
RECEIVE,
/**
* 消息完成
*/
COMPLETE,
/**
* 消息错误
*/
ERROR
}
/**
* (EXCHANGE)消息记录(死信)
*/
public enum EXCHANGE_MESSAGE_LOG_DLX {
/**
* 发送消息
*/
SEND_DLX,
/**
* 接收消息
*/
RECEIVE_DLX,
/**
* 消息完成
*/
COMPLETE_DLX,
/**
* 消息错误
*/
ERROR_DLX
}
/**
udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqListenerLogAspect.java
+10 -3
View File
@@ -12,6 +12,7 @@ import org.aspectj.lang.annotation.After;
import org.aspectj.lang.annotation.AfterThrowing;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
@@ -28,6 +29,12 @@ import java.util.Date;
@Slf4j
public class RabbitmqListenerLogAspect extends RabbitmqListenerBaseAspect {
/**
* messageLog
*/
@Autowired
private MessageLog messageLog;
/**
* 描述 : 前置通知
*
@@ -38,7 +45,7 @@ public class RabbitmqListenerLogAspect extends RabbitmqListenerBaseAspect {
if (joinPoint.getTarget() instanceof IRabbitmqListener) {
RabbitmqMessage<?> message = (RabbitmqMessage<?>) joinPoint.getArgs()[0];
String queues = this.getQueues(joinPoint);
log.info("{},{},receiveDate:{}", queues, message.getId(), new Date().getTime());
messageLog.receive(message.getId(), queues, new Date());
}
}
@@ -52,7 +59,7 @@ public class RabbitmqListenerLogAspect extends RabbitmqListenerBaseAspect {
if (joinPoint.getTarget() instanceof IRabbitmqListener) {
RabbitmqMessage<?> message = (RabbitmqMessage<?>) joinPoint.getArgs()[0];
String queues = this.getQueues(joinPoint);
log.info("{},{},completeDate:{}", queues, message.getId(), new Date().getTime());
messageLog.complete(message.getId(), queues, new Date());
}
}
@@ -67,7 +74,7 @@ public class RabbitmqListenerLogAspect extends RabbitmqListenerBaseAspect {
if (joinPoint.getTarget() instanceof IRabbitmqListener) {
RabbitmqMessage<?> message = (RabbitmqMessage<?>) joinPoint.getArgs()[0];
String queues = this.getQueues(joinPoint);
log.info("{},{},errorDate:{}", queues, message.getId(), new Date().getTime(), ex);
messageLog.error(message.getId(), queues, new Date(), ex);
}
}
udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/CompleteLog.java
+36
View File
@@ -0,0 +1,36 @@
package org.itkk.udf.amqp.rabbitmq.domain;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.ToString;
import java.util.Date;
/**
* CompleteLog
*/
@Data
@ToString
@EqualsAndHashCode(callSuper = false)
public class CompleteLog {
/**
* 描述 : 日志ID
*/
private String logId;
/**
* 描述 : 消息ID
*/
private String id;
/**
* 描述 : 队列
*/
private String queues;
/**
* 描述 : 完成时间
*/
private Date completeDate;
}
udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/ErrorLog.java
+41
View File
@@ -0,0 +1,41 @@
package org.itkk.udf.amqp.rabbitmq.domain;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.ToString;
import java.util.Date;
/**
* ErrorLog
*/
@Data
@ToString
@EqualsAndHashCode(callSuper = false)
public class ErrorLog {
/**
* 描述 : 日志ID
*/
private String logId;
/**
* 描述 : 消息ID
*/
private String id;
/**
* 描述 : 队列
*/
private String queues;
/**
* 描述 : 错误时间
*/
private Date errorDate;
/**
* 描述 : 错误信息
*/
private String errorMsg;
}
udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/ReceiveLog.java
+36
View File
@@ -0,0 +1,36 @@
package org.itkk.udf.amqp.rabbitmq.domain;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.ToString;
import java.util.Date;
/**
* ReceiveLog
*/
@Data
@ToString
@EqualsAndHashCode(callSuper = false)
public class ReceiveLog {
/**
* 描述 : 日志ID
*/
private String logId;
/**
* 描述 : 消息ID
*/
private String id;
/**
* 描述 : 队列
*/
private String queues;
/**
* 描述 : 消息接收时间
*/
private Date receiveDate;
}
udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/SendLog.java
+61
View File
@@ -0,0 +1,61 @@
package org.itkk.udf.amqp.rabbitmq.domain;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.ToString;
import java.util.Date;
/**
* SendLog
*/
@Data
@ToString
@EqualsAndHashCode(callSuper = false)
public class SendLog {
/**
* 日志ID
*/
private String logId;
/**
* 描述 : 消息ID
*/
private String id;
/**
* 发送者
*/
private String sender;
/**
* 描述 : 交换机
*/
private String exchange;
/**
* 描述 : 路由键
*/
private String routingKey;
/**
* 描述 : 时间戳
*/
private Long timestamp;
/**
* 描述 : 消息创建时间
*/
private Date msgCreateDate;
/**
* 描述 : 消息发送时间
*/
private Date sendDate;
/**
* 描述 : 消息体(json字符串)
*/
private String msgJsonString;
}