mirror of
https://gitee.com/wangkang/udf.git
synced 2026-10-28 06:13:44 +00:00
添加mq的相关配置
This commit is contained in:
udf_admin@163.com
+54
@@ -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<T> {
|
||||
|
||||
/**
|
||||
* 描述 : 消费者队列(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<T> message);
|
||||
|
||||
}
|
||||
@@ -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 <T> 任意类型
|
||||
*/
|
||||
public <T> void convertAndSend(String exchange, String routingKey, RabbitmqMessage<T> message) {
|
||||
message.setSender(springApplicationName);
|
||||
message.setSendDate(new Date());
|
||||
message.setExchange(exchange);
|
||||
message.setRoutingKey(routingKey);
|
||||
message.setTimestamp(System.currentTimeMillis());
|
||||
this.amqpTemplate.convertAndSend(exchange, routingKey, message);
|
||||
}
|
||||
|
||||
}
|
||||
+62
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
+74
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
+96
@@ -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<T> 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<String, String> header;
|
||||
|
||||
/**
|
||||
* 描述 : 消息体
|
||||
*/
|
||||
private T body; //NOSONAR
|
||||
|
||||
/**
|
||||
* 描述 : 构造函数
|
||||
*/
|
||||
public RabbitmqMessage() {
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* 描述 : 构造函数
|
||||
*
|
||||
* @param header 消息头
|
||||
* @param body 消息体
|
||||
*/
|
||||
public RabbitmqMessage(Map<String, String> header, T body) {
|
||||
this.id = UUID.randomUUID().toString();
|
||||
this.createDate = new Date();
|
||||
this.header = header;
|
||||
this.body = body;
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user