From ff3edadb2dd4626e640809810c439a7044d746db Mon Sep 17 00:00:00 2001 From: "udf_admin@163.com" Date: Thu, 16 Nov 2017 23:49:39 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0mq=E7=9A=84=E7=9B=B8=E5=85=B3?= =?UTF-8?q?=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../udf/amqp/rabbitmq/IRabbitmqListener.java | 54 +++++++++++ .../org/itkk/udf/amqp/rabbitmq/Rabbitmq.java | 54 +++++++++++ .../rabbitmq/RabbitmqListenerBaseAspect.java | 62 ++++++++++++ .../rabbitmq/RabbitmqListenerLogAspect.java | 74 ++++++++++++++ .../udf/amqp/rabbitmq/RabbitmqMessage.java | 96 +++++++++++++++++++ 5 files changed, 340 insertions(+) create mode 100644 udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/IRabbitmqListener.java create mode 100644 udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/Rabbitmq.java create mode 100644 udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqListenerBaseAspect.java create mode 100644 udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqListenerLogAspect.java create mode 100644 udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqMessage.java diff --git a/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/IRabbitmqListener.java b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/IRabbitmqListener.java new file mode 100644 index 0000000..f016d3f --- /dev/null +++ b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/IRabbitmqListener.java @@ -0,0 +1,54 @@ +/** + * IRabbitmqListener.java + * Created at 2016-11-17 + * Created by wangkang + * Copyright (C) 2016 itkk.org, All rights reserved. + */ +package org.itkk.udf.amqp.rabbitmq; + +import org.springframework.amqp.core.Binding; +import org.springframework.amqp.core.Queue; + +/** + * 描述 : IRabbitmqListener + * + * @author wangkang + */ +public interface IRabbitmqListener { + + /** + * 描述 : 消费者队列(Bean) + * + * @return Queue + */ + Queue queue(); + + /** + * 描述 : 死信消费者队列(Bean) + * + * @return Queue + */ + Queue dlxQueue(); + + /** + * 描述 : 绑定(Bean) + * + * @return Binding + */ + Binding binding(); + + /** + * 描述 : 死信绑定(Bean) + * + * @return Binding + */ + Binding dlxBinding(); + + /** + * 描述 : 处理消息 + * + * @param message 消息 + */ + void process(RabbitmqMessage message); + +} 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 new file mode 100644 index 0000000..67bee88 --- /dev/null +++ b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/Rabbitmq.java @@ -0,0 +1,54 @@ +/** + * Rabbitmq.java + * Created at 2016-11-17 + * Created by wangkang + * Copyright (C) 2016 itkk.org, All rights reserved. + */ +package org.itkk.udf.amqp.rabbitmq; + +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +import java.util.Date; + +/** + * 描述 : Rabbitmq + * + * @author wangkang + */ +@Component +public class Rabbitmq { + + /** + * 描述 : 应用名称 + */ + @Value("${spring.application.name}") + private String springApplicationName; + + /** + * 描述 : amqpTemplate + */ + @Autowired + private AmqpTemplate amqpTemplate; + + /** + * 描述 : Convert a Java object to an Amqp Message and send it to a specific exchange with a + * specific routing key. + * + * @param exchange exchange + * @param routingKey routingKey + * @param message a message to send + * @param 任意类型 + */ + public void convertAndSend(String exchange, String routingKey, RabbitmqMessage message) { + message.setSender(springApplicationName); + message.setSendDate(new Date()); + message.setExchange(exchange); + message.setRoutingKey(routingKey); + message.setTimestamp(System.currentTimeMillis()); + this.amqpTemplate.convertAndSend(exchange, routingKey, message); + } + +} diff --git a/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqListenerBaseAspect.java b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqListenerBaseAspect.java new file mode 100644 index 0000000..25fb9f9 --- /dev/null +++ b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqListenerBaseAspect.java @@ -0,0 +1,62 @@ +/** + * RabbitmqListenerAspect.java + * Created at 2016-11-17 + * Created by wangkang + * Copyright (C) 2016 itkk.org, All rights reserved. + */ +package org.itkk.udf.amqp.rabbitmq; + +import lombok.extern.slf4j.Slf4j; +import org.aspectj.lang.JoinPoint; +import org.aspectj.lang.annotation.Pointcut; +import org.itkk.udf.core.exception.SystemRuntimeException; +import org.springframework.amqp.rabbit.annotation.RabbitListener; + +import java.lang.reflect.Method; +import java.util.Arrays; + +/** + * 描述 : RabbitmqListenerBaseAspect + * + * @author wangkang + */ +@Slf4j +public class RabbitmqListenerBaseAspect { + + /** + * 描述 : listenerAspect + */ + @Pointcut("@annotation(org.springframework.amqp.rabbit.annotation.RabbitListener) && this(org.itkk.udf.amqp.rabbitmq.IRabbitmqListener)") + protected void listenerAspect() { + log.debug("RabbitmqListenerBaseAspect.listenerAspect"); + } + + /** + * 描述 : 获得对列名称 + * + * @param joinPoint joinPoint + * @return 对列名称 + */ + @SuppressWarnings("rawtypes") + protected String getQueues(JoinPoint joinPoint) { + try { + String targetName = joinPoint.getTarget().getClass().getName(); + String methodName = joinPoint.getSignature().getName(); + Object[] arguments = joinPoint.getArgs(); + Class targetClass = Class.forName(targetName); + Method[] methods = targetClass.getMethods(); + for (Method method : methods) { + if (method.getName().equals(methodName)) { + Class[] clazzs = method.getParameterTypes(); + if (clazzs.length == arguments.length) { //NOSONAR + return Arrays.toString(method.getAnnotation(RabbitListener.class).queues()); + } + } + } + } catch (ClassNotFoundException e) { + throw new SystemRuntimeException(e); + } + return null; + } + +} 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 new file mode 100644 index 0000000..90f359b --- /dev/null +++ b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqListenerLogAspect.java @@ -0,0 +1,74 @@ +/** + * RabbitmqListenerLogAspect.java + * Created at 2016-11-17 + * Created by wangkang + * Copyright (C) 2016 itkk.org, All rights reserved. + */ +package org.itkk.udf.amqp.rabbitmq; + +import lombok.extern.slf4j.Slf4j; +import org.aspectj.lang.JoinPoint; +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.core.annotation.Order; +import org.springframework.stereotype.Component; + +import java.util.Date; + +/** + * 描述 : RabbitmqListenerLogAspect + * + * @author wangkang + */ +@Aspect +@Component +@Order(100) +@Slf4j +public class RabbitmqListenerLogAspect extends RabbitmqListenerBaseAspect { + + /** + * 描述 : 前置通知 + * + * @param joinPoint joinPoint + */ + @Before("listenerAspect()") + public void before(JoinPoint joinPoint) { + 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()); + } + } + + /** + * 描述 : 后置通知 + * + * @param joinPoint joinPoint + */ + @After("listenerAspect()") + public void after(JoinPoint joinPoint) { + 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()); + } + } + + /** + * 异常通知 + * + * @param joinPoint joinPoint + * @param ex ex + */ + @AfterThrowing(pointcut = "listenerAspect()", throwing = "ex") + public void fterThrowing(JoinPoint joinPoint, Exception ex) { + 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); + } + } + +} diff --git a/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqMessage.java b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqMessage.java new file mode 100644 index 0000000..7d283e5 --- /dev/null +++ b/udf-starter-amqp-rabbitmq/src/main/java/org/itkk/udf/amqp/rabbitmq/RabbitmqMessage.java @@ -0,0 +1,96 @@ +/** + * Message.java + * Created at 2016-09-19 + * Created by wangkang + * Copyright (C) 2016 itkk.org, All rights reserved. + */ +package org.itkk.udf.amqp.rabbitmq; + +import lombok.Data; +import lombok.ToString; + +import java.io.Serializable; +import java.util.Date; +import java.util.Map; +import java.util.UUID; + +/** + * 描述 : 消息 + * + * @author wangkang + */ +@Data +@ToString +public class RabbitmqMessage implements Serializable { + + /** + * 描述 : id + */ + private static final long serialVersionUID = 1L; + + /** + * 描述 : 消息ID + */ + private String id; + + /** + * 发送者 + */ + private String sender; + + /** + * 描述 : 交换机 + */ + private String exchange; + + /** + * 描述 : 路由键 + */ + private String routingKey; + + /** + * 描述 : 时间戳 + */ + private Long timestamp; + + /** + * 描述 : 消息创建时间 + */ + private Date createDate; + + /** + * 描述 : 消息发送时间 + */ + private Date sendDate; + + /** + * 描述 : 消息头 + */ + private Map header; + + /** + * 描述 : 消息体 + */ + private T body; //NOSONAR + + /** + * 描述 : 构造函数 + */ + public RabbitmqMessage() { + + } + + /** + * 描述 : 构造函数 + * + * @param header 消息头 + * @param body 消息体 + */ + public RabbitmqMessage(Map header, T body) { + this.id = UUID.randomUUID().toString(); + this.createDate = new Date(); + this.header = header; + this.body = body; + } + +}