amqp rabbitmq

pom.xml

@@ -28,6 +28,7 @@
     <module>udf-starter-core</module>
     <module>udf-starter-rms</module>
     <module>udf-starter-cors</module>
+    <module>udf-starter-amqp-rabbitmq</module>
   </modules>
   <!-- 依赖管理 -->
   <dependencyManagement>

udf-starter-amqp-rabbitmq/.checkstyle

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<fileset-config file-format-version="1.2.0" simple-config="true" sync-formatter="true">
+  <fileset name="all" enabled="true" check-config-name="checkstyle.egridcloud.com" local="false">
+    <file-match-pattern match-pattern="." include-pattern="true"/>
+  </fileset>
+</fileset-config>

udf-starter-amqp-rabbitmq/.gitignore

@@ -0,0 +1,4 @@
+/target/
+/bin/
+/.classpath
+/.project

udf-starter-amqp-rabbitmq/pom.xml

@@ -0,0 +1,29 @@
+<?xml version="1.0"?>
+<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
+  xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+  <modelVersion>4.0.0</modelVersion>
+  <!-- maven坐标 -->
+  <artifactId>udf-starter-amqp-rabbitmq</artifactId>
+  <name>udf-starter-amqp-rabbitmq</name>
+  <packaging>jar</packaging>
+  <url>http://www.egridcloud.com</url>
+  <!-- 父依赖 -->
+  <parent>
+    <groupId>com.egridcloud.udf</groupId>
+    <artifactId>udf-starter</artifactId>
+    <version>1.0</version>
+  </parent>
+  <!-- 依赖 -->
+  <dependencies>
+    <!-- udf-core-starter -->
+    <dependency>
+      <groupId>com.egridcloud.udf</groupId>
+      <artifactId>udf-starter-core</artifactId>
+    </dependency>
+    <!-- spring-boot-starter-amqp -->
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-amqp</artifactId>
+    </dependency>
+  </dependencies>
+</project>

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 <T> 任意类型
+   *
+   * @param exchange the name of the exchange
+   * @param routingKey the routing key
+   * @param message a message to send
+   */
+  public <T> void convertAndSend(String exchange, String routingKey, Message<T> message) {
+    this.buildMessage(message, exchange, routingKey);
+    this.amqpTemplate.convertAndSend(exchange, routingKey, message);
+  }
+
+  /**
+   * 描述 : 构建消息
+   *
+   * @param <T> 任意类型
+   *
+   * @param message message
+   * @param exchange exchange
+   * @param routingKey routingKey
+   */
+  private <T> void buildMessage(Message<T> message, String exchange, String routingKey) {
+    message.setSendDate(new Date());
+    message.setExchange(exchange);
+    message.setRoutingKey(routingKey);
+  }
+
+}

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);
+    }
+  }
+
+}

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;
+  }
+
+}

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<T> { //NOSONAR
+
+  /**
+   * 描述 : 处理消息
+   *
+   * @param message 消息
+   */
+  public void process(Message<T> message);
+
+}

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<T> 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<String, String> header;
+
+  /**
+   * 
+   * 描述 : 消息体
+   * 
+   */
+  private T body; //NOSONAR
+
+  /**
+   * 描述 : 构造函数
+   *
+   */
+  public Message() {
+    super();
+  }
+
+  /**
+   * 描述 : 构造函数
+   *
+   * @param header 消息头
+   * @param body 消息体
+   */
+  public Message(Map<String, String> 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<String, String> getHeader() {
+    return header;
+  }
+
+  /**
+   * 描述 : 设置header
+   *
+   * @param header the header to set
+   */
+  public void setHeader(Map<String, String> 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;
+  }
+
+}