From 54d95210a6309cfeaed12871539ccb9560e57f28 Mon Sep 17 00:00:00 2001 From: "udf_admin@163.com" Date: Sun, 10 Dec 2017 15:23:51 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E6=B6=88=E6=81=AF=E4=B8=AD?= =?UTF-8?q?=E9=97=B4=E4=BB=B6=E6=97=A5=E5=BF=97=E8=AE=B0=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../itkk/udf/amqp/rabbitmq/MessageLog.java | 165 ++++++++++++++++++ .../org/itkk/udf/amqp/rabbitmq/Rabbitmq.java | 7 + .../udf/amqp/rabbitmq/RabbitmqConstant.java | 34 +++- .../rabbitmq/RabbitmqListenerLogAspect.java | 13 +- .../udf/amqp/rabbitmq/domain/CompleteLog.java | 36 ++++ .../udf/amqp/rabbitmq/domain/ErrorLog.java | 41 +++++ .../udf/amqp/rabbitmq/domain/ReceiveLog.java | 36 ++++ .../udf/amqp/rabbitmq/domain/SendLog.java | 61 +++++++ 8 files changed, 388 insertions(+), 5 deletions(-) create mode 100644 udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/MessageLog.java create mode 100644 udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/CompleteLog.java create mode 100644 udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/ErrorLog.java create mode 100644 udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/ReceiveLog.java create mode 100644 udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/SendLog.java diff --git a/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/MessageLog.java b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/MessageLog.java new file mode 100644 index 0000000..6c9af96 --- /dev/null +++ b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/MessageLog.java @@ -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 泛型 + * @param message message + */ + @Async + protected void send(RabbitmqMessage 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 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 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 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 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); + } +} diff --git a/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/Rabbitmq.java b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/Rabbitmq.java index 67bee88..614aa06 100644 --- a/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/Rabbitmq.java +++ b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/Rabbitmq.java @@ -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); } } diff --git a/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqConstant.java b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqConstant.java index 4f7eefa..5e6ddd7 100644 --- a/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqConstant.java +++ b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqConstant.java @@ -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 } /** diff --git a/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqListenerLogAspect.java b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqListenerLogAspect.java index 90f359b..e9b120d 100644 --- a/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqListenerLogAspect.java +++ b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqListenerLogAspect.java @@ -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); } } diff --git a/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/CompleteLog.java b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/CompleteLog.java new file mode 100644 index 0000000..4a613f7 --- /dev/null +++ b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/CompleteLog.java @@ -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; +} diff --git a/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/ErrorLog.java b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/ErrorLog.java new file mode 100644 index 0000000..4414e99 --- /dev/null +++ b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/ErrorLog.java @@ -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; +} diff --git a/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/ReceiveLog.java b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/ReceiveLog.java new file mode 100644 index 0000000..94a0f0e --- /dev/null +++ b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/ReceiveLog.java @@ -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; +} diff --git a/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/SendLog.java b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/SendLog.java new file mode 100644 index 0000000..73ccc0f --- /dev/null +++ b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/domain/SendLog.java @@ -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; +}