diff --git a/pom.xml b/pom.xml index 26a5c2d..a07f0a1 100644 --- a/pom.xml +++ b/pom.xml @@ -28,6 +28,7 @@ udf-starter-core udf-starter-rms udf-starter-cors + udf-starter-amqp-rabbitmq diff --git a/udf-starter-amqp-rabbitmq/.checkstyle b/udf-starter-amqp-rabbitmq/.checkstyle new file mode 100644 index 0000000..5b35a1b --- /dev/null +++ b/udf-starter-amqp-rabbitmq/.checkstyle @@ -0,0 +1,7 @@ + + + + + + + diff --git a/udf-starter-amqp-rabbitmq/.gitignore b/udf-starter-amqp-rabbitmq/.gitignore new file mode 100644 index 0000000..c541155 --- /dev/null +++ b/udf-starter-amqp-rabbitmq/.gitignore @@ -0,0 +1,4 @@ +/target/ +/bin/ +/.classpath +/.project diff --git a/udf-starter-amqp-rabbitmq/pom.xml b/udf-starter-amqp-rabbitmq/pom.xml new file mode 100644 index 0000000..e84241f --- /dev/null +++ b/udf-starter-amqp-rabbitmq/pom.xml @@ -0,0 +1,29 @@ + + + 4.0.0 + + udf-starter-amqp-rabbitmq + udf-starter-amqp-rabbitmq + jar + http://www.egridcloud.com + + + com.egridcloud.udf + udf-starter + 1.0 + + + + + + com.egridcloud.udf + udf-starter-core + + + + org.springframework.boot + spring-boot-starter-amqp + + + diff --git a/udf-starter-amqp-rabbitmq/src/main/java/com/egridcloud/udf/amqp/rabbitmq/Amqp.java b/udf-starter-amqp-rabbitmq/src/main/java/com/egridcloud/udf/amqp/rabbitmq/Amqp.java new file mode 100644 index 0000000..0cbab03 --- /dev/null +++ b/udf-starter-amqp-rabbitmq/src/main/java/com/egridcloud/udf/amqp/rabbitmq/Amqp.java @@ -0,0 +1,62 @@ +/** + * Amqp.java + * Created at 2016-11-17 + * Created by wangkang + * Copyright (C) 2016 egridcloud.com, All rights reserved. + */ +package com.egridcloud.udf.amqp.rabbitmq; + +import java.util.Date; + +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +import com.egridcloud.udf.core.amqp.Message; + +/** + * 描述 : Amqp + * + * @author wangkang + * + */ +@Component +public class Amqp { + + /** + * 描述 : 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 任意类型 + * + * @param exchange the name of the exchange + * @param routingKey the routing key + * @param message a message to send + */ + public void convertAndSend(String exchange, String routingKey, Message message) { + this.buildMessage(message, exchange, routingKey); + this.amqpTemplate.convertAndSend(exchange, routingKey, message); + } + + /** + * 描述 : 构建消息 + * + * @param 任意类型 + * + * @param message message + * @param exchange exchange + * @param routingKey routingKey + */ + private void buildMessage(Message message, String exchange, String routingKey) { + message.setSendDate(new Date()); + message.setExchange(exchange); + message.setRoutingKey(routingKey); + } + +} diff --git a/udf-starter-amqp-rabbitmq/src/main/java/com/egridcloud/udf/amqp/rabbitmq/ListenerAspect.java b/udf-starter-amqp-rabbitmq/src/main/java/com/egridcloud/udf/amqp/rabbitmq/ListenerAspect.java new file mode 100644 index 0000000..02fcbea --- /dev/null +++ b/udf-starter-amqp-rabbitmq/src/main/java/com/egridcloud/udf/amqp/rabbitmq/ListenerAspect.java @@ -0,0 +1,126 @@ +/** + * ListenerAspect.java + * Created at 2016-11-17 + * Created by wangkang + * Copyright (C) 2016 egridcloud.com, All rights reserved. + */ +package com.egridcloud.udf.amqp.rabbitmq; + +import java.lang.reflect.Method; +import java.util.Arrays; +import java.util.Date; + +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.aspectj.lang.annotation.Pointcut; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.amqp.rabbit.annotation.RabbitListener; +import org.springframework.stereotype.Component; + +import com.egridcloud.udf.core.amqp.IListener; +import com.egridcloud.udf.core.amqp.Message; +import com.egridcloud.udf.core.exception.SystemRuntimeException; + +/** + * 描述 : ListenerAspect + * + * @author wangkang + * + */ +@Aspect +@Component +public class ListenerAspect { + + /** + * 描述 : 日志 + */ + private static final Logger LOGGER = LoggerFactory.getLogger(ListenerAspect.class); + + /** + * 描述 : listenerAspect + * + */ + @Pointcut("execution(void com.egridcloud..*.process(..)) " + + "and @annotation(org.springframework.amqp.rabbit.annotation.RabbitListener) " + + "and this(com.egridcloud.udf.core.amqp.IListener)") + public void listenerAspect() { + LOGGER.debug("ListenerAspect.listenerAspect"); + } + + /** + * 描述 : 获得对列名称 + * + * @param joinPoint joinPoint + * @return 对列名称 + */ + @SuppressWarnings("rawtypes") + private 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; + } + + /** + * 描述 : 前置通知 + * + * @param joinPoint joinPoint + * @throws ClassNotFoundException + */ + @Before("listenerAspect()") + public void before(JoinPoint joinPoint) { + if (joinPoint.getTarget() instanceof IListener) { + Message message = (Message) joinPoint.getArgs()[0]; + String queues = this.getQueues(joinPoint); + LOGGER.debug("{},{},receiveDate:{}", queues, message.getId(), new Date().getTime()); + } + } + + /** + * 描述 : 后置通知 + * + * @param joinPoint joinPoint + */ + @After("listenerAspect()") + public void after(JoinPoint joinPoint) { + if (joinPoint.getTarget() instanceof IListener) { + Message message = (Message) joinPoint.getArgs()[0]; + String queues = this.getQueues(joinPoint); + LOGGER.debug("{},{},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 IListener) { + Message message = (Message) joinPoint.getArgs()[0]; + String queues = this.getQueues(joinPoint); + LOGGER.debug("{},{},errorDate:{}", queues, message.getId(), new Date().getTime(), ex); + } + } + +} diff --git a/udf-starter-amqp-rabbitmq/src/main/java/com/egridcloud/udf/amqp/rabbitmq/RabbitMqConfig.java b/udf-starter-amqp-rabbitmq/src/main/java/com/egridcloud/udf/amqp/rabbitmq/RabbitMqConfig.java new file mode 100644 index 0000000..5c20a37 --- /dev/null +++ b/udf-starter-amqp-rabbitmq/src/main/java/com/egridcloud/udf/amqp/rabbitmq/RabbitMqConfig.java @@ -0,0 +1,54 @@ +/** + * RabbitMqConfig.java + * Created at 2016-10-16 + * Created by wangkang + * Copyright (C) 2016 egridcloud.com, All rights reserved. + */ +package com.egridcloud.udf.amqp.rabbitmq; + +import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; +import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import com.fasterxml.jackson.databind.ObjectMapper; + +/** + * 描述 : RabbitMq配置 + * + * @author wangkang + * + */ +@Configuration +public class RabbitMqConfig { + + /** + * 描述 : rabbitListenerContainerFactoryPlus + * + * @param rabbitListenerContainerFactory rabbitListenerContainerFactory + * @param jackson2JsonMessageConverter jackson2JsonMessageConverter + * @return rabbitListenerContainerFactoryPlus + */ + @Bean + public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactoryPlus( + SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory, + Jackson2JsonMessageConverter jackson2JsonMessageConverter) { + rabbitListenerContainerFactory.setMessageConverter(jackson2JsonMessageConverter); + return rabbitListenerContainerFactory; + } + + /** + * 描述 : jackson2JsonMessageConverter + * + * @param xssObjectMapper xssObjectMapper + * + * @return jackson2JsonMessageConverter + */ + @Bean + public Jackson2JsonMessageConverter jackson2JsonMessageConverter(ObjectMapper xssObjectMapper) { + Jackson2JsonMessageConverter convert = new Jackson2JsonMessageConverter(); + convert.setJsonObjectMapper(xssObjectMapper); + return convert; + } + +} diff --git a/udf-starter-core/src/main/java/com/egridcloud/udf/core/amqp/IListener.java b/udf-starter-core/src/main/java/com/egridcloud/udf/core/amqp/IListener.java new file mode 100644 index 0000000..96bb822 --- /dev/null +++ b/udf-starter-core/src/main/java/com/egridcloud/udf/core/amqp/IListener.java @@ -0,0 +1,24 @@ +/** + * IListener.java + * Created at 2016-11-17 + * Created by wangkang + * Copyright (C) 2016 egridcloud.com, All rights reserved. + */ +package com.egridcloud.udf.core.amqp; + +/** + * 描述 : IListener + * + * @author wangkang + * + */ +public interface IListener { //NOSONAR + + /** + * 描述 : 处理消息 + * + * @param message 消息 + */ + public void process(Message message); + +} diff --git a/udf-starter-core/src/main/java/com/egridcloud/udf/core/amqp/Message.java b/udf-starter-core/src/main/java/com/egridcloud/udf/core/amqp/Message.java new file mode 100644 index 0000000..9ee3f3a --- /dev/null +++ b/udf-starter-core/src/main/java/com/egridcloud/udf/core/amqp/Message.java @@ -0,0 +1,245 @@ +/** + * Message.java + * Created at 2016-09-19 + * Created by wangkang + * Copyright (C) 2016 egridcloud.com, All rights reserved. + */ +package com.egridcloud.udf.core.amqp; + +import java.io.Serializable; +import java.util.Date; +import java.util.Map; +import java.util.UUID; + +/** + * + * 描述 : 消息 + * + * + * + * @author wangkang + * + * + * + */ +public class Message implements Serializable { + + /** + * + * 描述 : id + * + */ + private static final long serialVersionUID = 1L; + + /** + * + * 描述 : 消息ID + * + */ + private String id; + + /** + * 描述 : 交换机 + */ + private String exchange; + + /** + * 描述 : 路由键 + */ + private String routingKey; + + /** + * 描述 : 时间戳 + */ + private Long timestamp; + + /** + * 描述 : 消息创建时间 + */ + private Date createDate; + + /** + * 描述 : 消息发送时间 + */ + private Date sendDate; + + /** + * 描述 : 消息头 + */ + private Map header; + + /** + * + * 描述 : 消息体 + * + */ + private T body; //NOSONAR + + /** + * 描述 : 构造函数 + * + */ + public Message() { + super(); + } + + /** + * 描述 : 构造函数 + * + * @param header 消息头 + * @param body 消息体 + */ + public Message(Map header, T body) { + super(); + this.id = UUID.randomUUID().toString(); + this.createDate = new Date(); + this.timestamp = createDate.getTime(); + this.header = header; + this.body = body; + } + + /** + * 描述 : 获取timestamp + * + * @return the timestamp + */ + public Long getTimestamp() { + return timestamp; + } + + /** + * 描述 : 设置timestamp + * + * @param timestamp the timestamp to set + */ + public void setTimestamp(Long timestamp) { + this.timestamp = timestamp; + } + + /** + * 描述 : 获取exchange + * + * @return the exchange + */ + public String getExchange() { + return exchange; + } + + /** + * 描述 : 设置exchange + * + * @param exchange the exchange to set + */ + public void setExchange(String exchange) { + this.exchange = exchange; + } + + /** + * 描述 : 获取routingKey + * + * @return the routingKey + */ + public String getRoutingKey() { + return routingKey; + } + + /** + * 描述 : 设置routingKey + * + * @param routingKey the routingKey to set + */ + public void setRoutingKey(String routingKey) { + this.routingKey = routingKey; + } + + /** + * 描述 : 获取sendDate + * + * @return the sendDate + */ + public Date getSendDate() { + return sendDate; + } + + /** + * 描述 : 设置sendDate + * + * @param sendDate the sendDate to set + */ + public void setSendDate(Date sendDate) { + this.sendDate = sendDate; + } + + /** + * 描述 : 获取createDate + * + * @return the createDate + */ + public Date getCreateDate() { + return createDate; + } + + /** + * 描述 : 设置createDate + * + * @param createDate the createDate to set + */ + public void setCreateDate(Date createDate) { + this.createDate = createDate; + } + + /** + * 描述 : 获取header + * + * @return the header + */ + public Map getHeader() { + return header; + } + + /** + * 描述 : 设置header + * + * @param header the header to set + */ + public void setHeader(Map header) { + this.header = header; + } + + /** + * 描述 : 获取id + * + * @return the id + */ + public String getId() { + return id; + } + + /** + * 描述 : 设置id + * + * @param id the id to set + */ + public void setId(String id) { + this.id = id; + } + + /** + * 描述 : 获取body + * + * @return the body + */ + public T getBody() { + return body; + } + + /** + * 描述 : 设置body + * + * @param body the body to set + */ + public void setBody(T body) { + this.body = body; + } + +}