Skip to content

B
bailuntec-cost
Commit 4eb9f0a4 by liyanlin
Options

fix

parent fcb801a0
Showing with 121 additions and 1447 deletions
cost-service/src/main/java/com/blt/other/common/rabbitmq/config/RabbitMQConfig.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.config;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 14:20 2021-04-27
* @Modified by:
*/
@Configuration
public class RabbitMQConfig {
@Value("${spring.rabbitmq.host}")
private String host;
@Value("${spring.rabbitmq.port}")
private int port;
@Value("${spring.rabbitmq.username}")
private String username;
@Value("${spring.rabbitmq.password}")
private String password;
@Value("${spring.rabbitmq.virtualHost:/}")
private String virtualHost;
/**
* 消息发送失败，是否回调给发送者
*/
@Value("${spring.rabbitmq.template.mandatory:false}")
private Boolean mandatory;
/**
* 是否确认
*/
@Value("${spring.rabbitmq.publisher-confirms:false}")
private Boolean publisherConfirms;
/**
* 如果mandatorys设置成true，该值也设置 成true
*/
@Value("${spring.rabbitmq.publisher-returns:false}")
private Boolean publisherReturns;
// RabbitMQConfig中定义connectionFactory中设置属性
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
cachingConnectionFactory.setAddresses(this.host);
cachingConnectionFactory.setPort(this.port);
cachingConnectionFactory.setUsername(this.username);
cachingConnectionFactory.setPassword(this.password);
cachingConnectionFactory.setVirtualHost(this.virtualHost);
// 如果消息要设置成回调，则以下的配置必须要设置成true
cachingConnectionFactory.setPublisherConfirms(this.publisherConfirms);
cachingConnectionFactory.setPublisherReturns(this.publisherReturns);
return cachingConnectionFactory;
}
// 同时为了调用SpringBoot集成rabbitMQ提供的发送的方法，我们需要注入rabbitTemplate
/**
* 因为要设置回调类，所以应是prototype类型，如果是singleton类型，则回调类为最后一次设置
* 主要是为了设置回调类
*
* @return
*/
@Bean(name = "rabbitTemplate")
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(this.connectionFactory());
template.setMessageConverter(new Jackson2JsonMessageConverter());
return template;
}
@Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
return new RabbitAdmin(connectionFactory);
}
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/declare/AbstractDeclare.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.declare;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 2021-04-28
* @Modified by:
*/
public abstract class AbstractDeclare {
@Autowired
RabbitAdmin rabbitAdmin;
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/declare/AmBindDeclare.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.declare;
import com.alibaba.fastjson.JSONObject;
import com.blt.other.common.rabbitmq.model.MqExchange;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Map;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 2021-04-28
* @Modified by:
*/
@Component
@Slf4j
public class AmBindDeclare extends AbstractDeclare {
/**
* 队列与交换机进行绑定
*
* @param queueName 队列名称
* @param exchangeName 交换机名称
* @param routingKey 路由键
* @return
*/
public boolean queueBind(String queueName, String exchangeName, String routingKey) {
return this.bind(queueName, Binding.DestinationType.QUEUE, exchangeName, routingKey, null);
}
/**
* 绑定队列
* 该绑定的默认交换机是 amq.direct 交换机, direct类型的交换机且开启持久化
* 路由键也是该队列
*
* @param queueName
* @return
*/
public boolean queueBind(String queueName) {
return this.queueBind(queueName, MqExchange.DEFAULT_DIRECT_EXCHANGE, queueName);
}
/**
* 交换机和交换机进行绑定
*
* @param destExchangeName 目标交换机名称
* @param exchangeName 交换机名称
* @param routingKey 路由键
* @return
*/
public boolean exchangeBind(String destExchangeName, String exchangeName, String routingKey) {
return this.bind(destExchangeName, Binding.DestinationType.EXCHANGE, exchangeName, routingKey, null);
}
/**
* bind绑定
*
* @param destName 目标名称（可以是队列 也可以是交换机）
* @param type 绑定的类型 交换机 / 队列
* @param exchangeName 交换机的名称
* @param routingKey 路由键
* @param map 结构参数
* @return
*/
public boolean bind(String destName, Binding.DestinationType type, String exchangeName, String routingKey, Map<String, Object> map) {
log.info("bind parameter is destName: " + destName + ", type: " + type.name()
+ ", exchangeName: " + exchangeName + ", routingKey: " + routingKey + ", map: " + JSONObject.toJSONString(map));
Binding binding = new Binding(destName, Binding.DestinationType.QUEUE, exchangeName, routingKey, map);
try {
this.rabbitAdmin.declareBinding(binding);
} catch (Exception e) {
if (log.isDebugEnabled()) {
throw new AmqpException(e);
}
return Boolean.FALSE;
}
return Boolean.TRUE;
}
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/declare/AmExchangeDeclare.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.declare;
import com.blt.other.common.rabbitmq.model.MqExchange;
import com.blt.other.common.rabbitmq.type.ExchangeTypeEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 2021-04-28
* @Modified by:
*/
@Component
@Slf4j
public class AmExchangeDeclare extends AbstractDeclare {
/**
* 向rabbitMQ服务器注册指定的交换机以及交换机的类型
*
* @param mqExchage
* @return
*/
public Exchange declareExchange(MqExchange mqExchage) {
log.info("declare exchange is :" + mqExchage.toString());
Exchange exchange = null;
exchange = this.initExchange(mqExchage);
this.rabbitAdmin.declareExchange(exchange);
log.info("declare exchange success");
return exchange;
}
/**
* 从RabbitMQ服务端上删除指定的交换机
*
* @param exchangeName
* @return
*/
public boolean deleteExchange(String exchangeName) {
log.info("delete exchange is : " + exchangeName);
if (StringUtils.isEmpty(exchangeName)) {
throw new AmqpException("the parameter exchangeName couldn't not be null");
}
return this.rabbitAdmin.deleteExchange(exchangeName);
}
/**
* 根据不同类型初始化不同类型的交换机
*
* @param mqExchage
* @return
*/
private Exchange initExchange(MqExchange mqExchage) {
ExchangeTypeEnum exchangeTypeEnum = mqExchage.getType();
switch (exchangeTypeEnum) {
case DIRECT:
return new DirectExchange(mqExchage.getName(), mqExchage.isDurable(), mqExchage.isAutoDelete(), mqExchage.getArguments());
case TOPIC:
return new TopicExchange(mqExchage.getName(), mqExchage.isDurable(), mqExchage.isAutoDelete(), mqExchage.getArguments());
case FANOUT:
return new FanoutExchange(mqExchage.getName(), mqExchage.isDurable(), mqExchage.isAutoDelete(), mqExchage.getArguments());
case HEADERS:
return new HeadersExchange(mqExchage.getName(), mqExchage.isDurable(), mqExchage.isAutoDelete(), mqExchage.getArguments());
default:
return null;
}
}
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/declare/AmQueueDeclare.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.declare;
import com.bailuntec.common.BeanUtils;
import com.blt.other.common.rabbitmq.model.MqQueue;
import com.rabbitmq.client.AMQP;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 2021-04-28
* @Modified by:
*/
@Component
@Slf4j
public class AmQueueDeclare extends AbstractDeclare{
/**
* 声明队列
* 向rabbitMQ服务器声明一个队列
*
* @param mqQueue
* @return
*/
public Queue declareQueue(MqQueue mqQueue) {
log.info("the parameter queue is : " + mqQueue.toString());
Queue queue = new Queue(mqQueue.getName());
BeanUtils.copyProperties(mqQueue, queue);
log.info("declare queue is : " + queue.toString());
super.rabbitAdmin.declareQueue(queue);
log.info("declare queue success");
return queue;
}
/**
* 清空队列中的消息
*
* @param queueName
* @return 清楚队列中的消息的个数
*/
public int purgeQueue(String queueName) {
if (StringUtils.isEmpty(queueName)) {
throw new AmqpException("队列名不能为空");
}
log.info("purge queue is : " + queueName);
return super.rabbitAdmin.purgeQueue(queueName);
}
/**
* 判断指定的队列是否存在
* 1. 如果存在则返回该队列
* 2. 如果不存在则返回null
*
* @param queueName
* @return true 存在， false 不存在
*/
public boolean isQueueExist(String queueName) {
if (StringUtils.isEmpty(queueName)) {
throw new AmqpException("队列名不能为空");
}
log.info("isQueueExist queue is : " + queueName);
String isExist = super.rabbitAdmin.getRabbitTemplate().execute((channel -> {
try {
AMQP.Queue.DeclareOk declareOk = channel.queueDeclarePassive(queueName);
return declareOk.getQueue();
} catch (Exception e) {
if (log.isDebugEnabled()) {
throw new AmqpException(e);
}
return null;
}
}));
log.info("the queue " + queueName + " is exist : " + isExist);
return StringUtils.isEmpty(isExist) ? Boolean.FALSE : Boolean.TRUE;
}
/**
* 从rabbitMQ服务器中删除指定的队列
*
* @param queueName
* @return
*/
public boolean deleteQueue(String queueName) {
log.info("delete queue is :" + queueName);
if (StringUtils.isEmpty(queueName)) {
throw new AmqpException("队列名不能为空");
}
return super.rabbitAdmin.deleteQueue(queueName);
}
/**
* 从rabbitMQ服务器中删除指定的队列
*
* @param queueName 队列名称
* @param unused 队列是否在使用，如果设置为true则该队列只能在没有被使用的情况下才能删除
* @param empty 队列是否为空，如果设置为true则该队列只能在该队列没有消息时才会被删除
*/
public void deleteQueue(String queueName, boolean unused, boolean empty) {
log.info("delete queue is : { queueName : '" + queueName
+ "' , unused: '" + unused + "' , empty:'" + empty + "'}");
if (StringUtils.isEmpty(queueName)) {
throw new AmqpException("队列名不能为空");
}
super.rabbitAdmin.deleteQueue(queueName, unused, empty);
}
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/listen/AbstractMessageHandler.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.listen;
import com.alibaba.fastjson.JSONObject;
import com.blt.other.common.rabbitmq.model.MqMessage;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;
import java.util.concurrent.ConcurrentHashMap;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 17:38 2021-04-27
* @Modified by:
*/
@Component
@Slf4j
public abstract class AbstractMessageHandler implements ChannelAwareMessageListener {
@Value("${spring.message.queue.retryTimes:5}")
private Integer retryTimes;
/**
* 用户自定义消息处理
*
* @param message 消息
*/
public abstract void handleMessage(String message, Channel channel);
private ConcurrentHashMap<String, AcknowledgeMode> ackMap = new ConcurrentHashMap<>(8);
/**
* 消息处理
*
* @param message 消息体
* @param channel channel通道
* @throws Exception
*/
@Override
public void onMessage(Message message, Channel channel) throws Exception {
log.info("接收到发送的消息.......");
// 业务处理是否成功
boolean handleResult = false;
// 消息处理标识
long deliveryTag = message.getMessageProperties().getDeliveryTag();
// 获取消费的队列名
String queue = message.getMessageProperties().getConsumerQueue();
MqMessage mqMessage = null;
// TODO 进行自己的业务处理 比如记录日志
try {
String msg = new String(message.getBody());
mqMessage = JSONObject.parseObject(msg, MqMessage.class);
// 自定义业务处理
this.handleMessage(JSONObject.toJSONString(mqMessage.getMessageBody()), channel);
} catch (Exception e) {
if (log.isDebugEnabled()) {
e.printStackTrace();
}
}
// TODO 如果消息处理失败，处理失败的采取措施， 确保消息不丢失
this.onMessageCompleted(mqMessage, queue, channel, deliveryTag, handleResult);
}
/**
* 消息处理结束后进行复处理
*
* @param mqMessage 消息实体
* @param queue
* @param channel
* @param deliveryTag
* @param handleResult 业务处理是否成功
*/
private void onMessageCompleted(MqMessage mqMessage, String queue, Channel channel, long deliveryTag, boolean handleResult) {
log.info("消息：" + mqMessage.toString() + "处理完成，等待事务提交和状态更新");
if (!handleResult) {
// TODO 业务处理失败，需要更新状态
return;
}
AcknowledgeMode ack = this.ackMap.get(queue);
if (ack.isManual()) {
//重试5次
int retryTimes = 5;
//进行消息
RetryTemplate oRetryTemplate = new RetryTemplate();
SimpleRetryPolicy oRetryPolicy = new SimpleRetryPolicy();
oRetryPolicy.setMaxAttempts(retryTimes);
oRetryTemplate.setRetryPolicy(oRetryPolicy);
try {
// obj为doWithRetry的返回结果，可以为任意类型
Integer result = oRetryTemplate.execute(new RetryCallback<Integer, Exception>() {
int count = 0;
@Override
public Integer doWithRetry(RetryContext context) throws Exception {//开始重试
channel.basicAck(deliveryTag, false);
AbstractMessageHandler.log.info("消息" + mqMessage.toString() + "已签收");
return ++this.count;
}
}, new RecoveryCallback<Integer>() {
@Override
public Integer recover(RetryContext context) throws Exception { //重试多次后都失败了
AbstractMessageHandler.log.info("消息" + mqMessage.toString() + "签收失败");
return Integer.MAX_VALUE;
}
});
if (result.intValue() <= retryTimes) {
//消息签收成功 更改状态
} else {
//MQ服务器或网络出现问题，签收失败 更改状态
}
} catch (Exception e) {
log.error("消息" + mqMessage.toString() + "签收出现异常：" + e.getMessage());
}
} else {
log.info("消息自动签收");
}
}
/**
* @param ack
* @Title: setAck
* @date: 2018年9月14日 上午11:17:41
* @Description: 注入消息签收模式
*/
public final void setAck(String queue, AcknowledgeMode ack) {
this.ackMap.put(queue, ack);
log.info("注入队列 " + queue + " 消息签收模式: " + ack.name());
}
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/listen/MessageListen.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.listen;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 17:43 2021-04-27
* @Modified by:
*/
@Component
@Slf4j
public class MessageListen {
@Autowired
private ConnectionFactory connectionFactory;
/**
* 在容器中加入消息监听
*
* @param queue
* @param messageHandler
* @param isAck
* @throws Exception
*/
public void addMessageLister(String queue, AbstractMessageHandler messageHandler, boolean isAck) throws Exception {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(this.connectionFactory);
container.setQueueNames(queue);
AcknowledgeMode ack = AcknowledgeMode.NONE;
if (isAck) {
ack = AcknowledgeMode.MANUAL;
}
messageHandler.setAck(queue, ack);
container.setAcknowledgeMode(ack);
MessageListenerAdapter adapter = new MessageListenerAdapter(messageHandler);
container.setMessageListener(adapter);
container.start();
log.info("------ 已成功监听异步消息触发通知队列：" + queue + " ------");
}
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/model/HandleResult.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.model;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 10:26 2021-04-28
* @Modified by:
*/
public class HandleResult {
/**
* 是否回调
*/
private boolean callback;
/**
* 类型
*/
private Integer type;
/**
* 内容
*/
private String content;
/**
* 消息
*/
private String msg;
/**
* 是否处理成功
*/
private boolean result;
public HandleResult(CallBack callBack) {
this.callback = callBack.callback;
this.content = callBack.content;
this.type = callBack.type;
this.msg = callBack.msg;
this.result = callBack.callback;
}
public boolean isCallback() {
return this.callback;
}
public void setCallback(boolean callback) {
this.callback = callback;
}
/*public Integer getType() {
if (StringUtils.isEmpty(this.type)) {
this.type = HandleSignalEnum.SINGAL_CALLBACK.getCode();
}
return this.type;
}*/
public void setType(Integer type) {
this.type = type;
}
public String getContent() {
return this.content;
}
public void setContent(String content) {
this.content = content;
}
public String getMsg() {
return this.msg;
}
public void setMsg(String msg) {
this.msg = msg;
}
public boolean getResult() {
return this.result;
}
public void setResult(boolean result) {
this.result = result;
}
public static class CallBack {
/**
* 是否回调
*/
private boolean callback = true;
/**
* 回调请求类型
*/
private Integer type;
/**
* 回调内容
*/
private String content;
private String msg;
/**
* 是否处理成功
*/
private boolean result;
public CallBack(boolean result) {
this.result = result;
}
public CallBack type(Integer type) {
this.type = type;
return this;
}
public CallBack content(String content) {
this.content = content;
return this;
}
public CallBack callback(boolean callback) {
this.callback = callback;
return this;
}
public CallBack msg(String msg) {
this.msg = msg;
return this;
}
public HandleResult builder() {
return new HandleResult(this);
}
}
}
\ No newline at end of file
cost-service/src/main/java/com/blt/other/common/rabbitmq/model/MessageData.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.model;
import lombok.Data;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 10:29 2021-04-28
* @Modified by:
*/
@Data
public class MessageData {
/**
* 消息的内容
*/
private Object content;
/**
* 发送方
*/
private String sender;
/**
* 接收方
*/
private String receiver;
/**
* 消息的类型（发布的内容/指令）
*/
private String msgType;
/**
* 唯一标志
*/
private String tag;
/**
* 处理结果
*/
private boolean handleResult;
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/model/MqExchange.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.model;
import com.blt.other.common.rabbitmq.type.ExchangeTypeEnum;
import com.sun.istack.NotNull;
import lombok.Data;
import java.util.Map;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 2021-04-28
* @Modified by:
*/
@Data
public class MqExchange {
/**
* 默认的DIRECT类型的交换机
*/
public static final String DEFAULT_DIRECT_EXCHANGE = "amq.direct";
/**
* 默认的TOPIC类型的交换机
*/
public static final String DEFAULT_TOPIC_EXCHANGE = "amq.topic";
/**
* 默认的HEADERS类型的交换机
*/
public static final String DEFAULT_HEADERS_EXCHANGE = "amq.headers";
/**
* 默认的FANOUT类型的交换机
*/
public static final String DEFAULT_FANOUT_EXCHANGE = "amq.fanout";
/**
* 交换机的名称
*/
@NotNull
private String name;
/**
* 交换机的类型
*/
@NotNull
private ExchangeTypeEnum type;
/**
* 是否持久化
* 持久化可以将交换机存盘，在服务器重启的时候不会丢失相关的信息
* 默认是开启持久化
*/
private boolean durable = Boolean.TRUE;
/**
* 是否自动删除
* 自动删除的前提是至少有一个队列或者交换机与这个交互机绑定，之后所有与这个交换机绑定的队列或者交换机都与此解绑
*/
private boolean autoDelete;
public MqExchange name(String name) {
this.name = name;
return this;
}
public MqExchange type(ExchangeTypeEnum type) {
this.type = type;
return this;
}
public MqExchange durable(boolean durable) {
this.durable = durable;
return this;
}
public MqExchange autoDelete(boolean autoDelete) {
this.autoDelete = autoDelete;
return this;
}
public MqExchange arguments(Map<String, Object> arguments) {
this.arguments = arguments;
return this;
}
/**
* 自定义属性参数
* 比如：alternate-exchange
*/
private Map<String, Object> arguments;
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/model/MqMessage.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.model;
import lombok.Data;
import java.io.Serializable;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 17:25 2021-04-27
* @Modified by:
*/
@Data
public class MqMessage implements Serializable {
private Object messageBody;
private String messageId;
private String exchangeName;
private String queueName;
private String routingKey;
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/model/MqQueue.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.model;
import com.sun.istack.NotNull;
import lombok.Builder;
import lombok.Data;
import java.io.Serializable;
import java.util.Map;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 09:37 2021-04-28
* @Modified by:
*/
@Data
public class MqQueue implements Serializable {
/**
* 队列名称
*/
@NotNull
private String name;
/**
* 是否持久化
* 持久化会存盘，服务器重启时不会丢失相关信息
*/
private boolean durable;
/**
* 是否排他
* 如果是排他，则该队列对首次声明他的连接有效，并在连接断开时自动删除
* 注意：
* 1. 同一个连接的其他的Channel是可以连接该排他队列的
* 2. 首次是说其他连接就不同创建同名的排他队列
* 适用于一个客户端同时发送和读取消息
*/
private boolean exclusive;
/**
* 是否自动删除
* 自动删除的前提是至少有一个队列或者交换机与这个交互机绑定，之后所有与这个交换机绑定的队列或者交换机都与此解绑
*/
private boolean autoDelete;
/**
* 结构化参数
* x-message-ttl、x-expires等
*/
private Map<String, Object> arguments;
public MqQueue name(String name) {
this.name = name;
return this;
}
public MqQueue durable(boolean durable) {
this.durable = durable;
return this;
}
public MqQueue exclusive(boolean exclusive) {
this.exclusive = exclusive;
return this;
}
public MqQueue autoDelete(boolean autoDelete) {
this.autoDelete = autoDelete;
return this;
}
public MqQueue arguments(Map<String, Object> arguments) {
this.arguments = arguments;
return this;
}
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/register/AbstractRegisterQueue.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.register;
import com.blt.other.common.rabbitmq.declare.AmBindDeclare;
import com.blt.other.common.rabbitmq.declare.AmQueueDeclare;
import com.blt.other.common.rabbitmq.listen.AbstractMessageHandler;
import com.blt.other.common.rabbitmq.listen.MessageListen;
import com.blt.other.common.rabbitmq.model.MqExchange;
import com.blt.other.common.rabbitmq.model.MqQueue;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Binding;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.util.StringUtils;
import java.util.Map;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 2021-04-28
* @Modified by:
*/
@Data
@Slf4j
public abstract class AbstractRegisterQueue {
@Autowired
AmBindDeclare amBindDeclare;
@Autowired
AmQueueDeclare amQueueDeclare;
@Autowired
MessageListen messageListen;
@Value("${spring.rabbitmq.queue.isAck:false}")
private Boolean isAck;
/**
* 子类提供自定义的消息监听
*
* @return
*/
public abstract AbstractMessageHandler messageHandler();
/**
* 实例化队列名
*
* @param queue
* @return
*/
public AbstractRegisterQueue queue(String queue) {
this.queue = queue;
return this;
}
/**
* 实例化交换机
*
* @param exchange
* @return
*/
public AbstractRegisterQueue exchange(String exchange) {
this.exchange = exchange;
return this;
}
/**
* 实例化路由键
*
* @param routingKey
* @return
*/
public AbstractRegisterQueue routingKey(String routingKey) {
this.routingKey = routingKey;
return this;
}
/**
* 实例化结构化属性
*
* @param properties
* @return
*/
public AbstractRegisterQueue properties(Map<String, Object> properties) {
this.properties = properties;
return this;
}
/**
* 队列名
*/
private String queue;
/**
* 交换机 默认是 amq.direct 交换机
*/
private String exchange = MqExchange.DEFAULT_DIRECT_EXCHANGE;
/**
* 路由键 默认是队列名
*/
private String routingKey = this.getQueue();
/**
* 结构化属性
*/
private Map<String, Object> properties;
public String getRoutingKey() {
if (StringUtils.isEmpty(this.routingKey)) {
return this.getQueue();
}
return this.routingKey;
}
/**
* 注册队列，并且监听队列
*
* @return
*/
public boolean registerQueue() {
MqQueue mqQueue = new MqQueue().name(this.queue);
this.amQueueDeclare.declareQueue(mqQueue);
boolean tag = this.amBindDeclare.bind(this.queue, Binding.DestinationType.QUEUE, this.exchange, this.getRoutingKey(), this.properties);
if (tag) {
try {
this.messageListen.addMessageLister(this.queue, this.messageHandler(), this.isAck);
return Boolean.TRUE;
} catch (Exception e) {
if (log.isDebugEnabled()) {
e.printStackTrace();
}
return Boolean.FALSE;
}
}
return tag;
}
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/service/receiver/AbstractReceiver.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.service.receiver;
import com.blt.other.common.rabbitmq.model.HandleResult;
import com.blt.other.common.rabbitmq.model.MessageData;
import lombok.extern.slf4j.Slf4j;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 2021-04-28
* @Modified by:
*/
@Slf4j
public abstract class AbstractReceiver implements Receiver{
/**
* 用户自定义消息处理
*
* @param messageData
* @return
*/
public abstract HandleResult exec(MessageData messageData) throws Exception;
/**
* 成功处理
*
* @param messageData
* @return
*/
public abstract HandleResult handleSuccess(MessageData messageData);
/**
* 失败处理
*
* @param messageData
* @return
*/
public abstract HandleResult handleFail(MessageData messageData);
/**
* 处理
*
* @param messageData
* @return
*/
@Override
public final HandleResult handleMessage(MessageData messageData) {
log.info(this.getClass().getSimpleName() + "-->handleMessage()参数 unicomData:{}", messageData.toString());
HandleResult handleResult = null;
try {
// 根据自行处理的返回结果
handleResult = this.exec(messageData);
// 执行成功处理的逻辑
handleResult = this.handleSuccess(messageData);
} catch (Exception e) {
e.printStackTrace();
messageData.setContent(e.getMessage());
return this.handleFail(messageData);
}
return handleResult;
}
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/service/receiver/DefaultRegisterQueueService.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.service.receiver;
import com.blt.other.common.rabbitmq.listen.AbstractMessageHandler;
import com.blt.other.common.rabbitmq.register.AbstractRegisterQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 2021-04-28
* @Modified by:
*/
@Service
public class DefaultRegisterQueueService extends AbstractRegisterQueue {
@Autowired
MessageHandle messageHandle;
@Override
public AbstractMessageHandler messageHandler() {
return this.messageHandle;
}
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/service/receiver/MessageHandle.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.service.receiver;
import com.blt.other.common.rabbitmq.listen.AbstractMessageHandler;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 2021-04-28
* @Modified by:
*/
@Component
@Slf4j
public class MessageHandle extends AbstractMessageHandler {
@Override
public void handleMessage(String message, Channel channel) {
System.out.println("-----------------------------------------------");
log.info("进入自定义处理方法中......\n消息为：" + message);
System.out.println("进入自定义处理方法中......\n消息为：" + message);
System.out.println("-----------------------------------------------");
}
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/service/receiver/Receiver.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.service.receiver;
import com.blt.other.common.rabbitmq.model.HandleResult;
import com.blt.other.common.rabbitmq.model.MessageData;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 10:25 2021-04-28
* @Modified by:
*/
public interface Receiver {
HandleResult handleMessage(MessageData messageData);
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/service/sender/AbstractSendService.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.service.sender;
import com.blt.other.common.rabbitmq.model.MqMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpConnectException;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import javax.annotation.PostConstruct;
import java.util.UUID;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 17:04 2021-04-27
* @Modified by:
*/
@Component
@Slf4j
public abstract class AbstractSendService implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{
public final static String DEFAULT_EXCHANGE = "amq.direct";
@Autowired
RabbitTemplate rabbitTemplate;
/**
* 简单的发送消息
* 发送的交换机是默认的 amq.direct交换机，该交换机的类型是DIRECT类型，开启持久化机制
* 发送的队列即为RoutingKey,需要绑定队列时
*
* @param queue 队列，默认是跟路由键是相同的
* @param content 发送的内容
*/
public void send(String queue, String content) {
if (StringUtils.isEmpty(queue)) {
throw new AmqpConnectException(new Exception("发送的队列不能为空"));
}
if (StringUtils.isEmpty(content)) {
throw new AmqpConnectException(new Exception("内容不能为空"));
}
this.send(DEFAULT_EXCHANGE, queue, content, null, UUID.randomUUID().toString());
}
/**
* 发送一条有过期时间的消息
*
* @param queue 队列，默认是跟路由键相同的
* @param content 发送的内容
* @param expireTime 过期时间 时间毫秒
*/
public void send(String queue, String content, int expireTime) {
if (StringUtils.isEmpty(queue)) {
throw new AmqpConnectException(new Exception("发送的队列不能为空"));
}
if (StringUtils.isEmpty(content)) {
throw new AmqpConnectException(new Exception("内容不能为空"));
}
MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
// 设置消息的过期时间
message.getMessageProperties().setExpiration(expireTime + "");
return message;
}
};
this.send(DEFAULT_EXCHANGE, queue, content, messagePostProcessor, UUID.randomUUID().toString());
}
/**
* 按照给定的交换机、路由键、发送内容、发送的自定义属性来发送消息
* TODO 待完善交互方式
*
* @param exchange 交换机名称
* @param routingKey 路由键
* @param object 发送的内容
* @param messagePostProcessor 发送消息自定义处理
* @param messageId 消息ID
*/
public void send(String exchange, String routingKey, Object object, MessagePostProcessor messagePostProcessor, String messageId) {
if (StringUtils.isEmpty(exchange)) {
throw new AmqpConnectException(new Exception("交换机不能为空"));
}
if (StringUtils.isEmpty(routingKey)) {
throw new AmqpConnectException(new Exception("路由键不能为空"));
}
if (StringUtils.isEmpty(object)) {
throw new AmqpConnectException(new Exception("发送的内容不能为空"));
}
CorrelationData correlationData = new CorrelationData();
correlationData.setId(StringUtils.isEmpty(messageId) ? UUID.randomUUID().toString() : messageId);
MqMessage mqMessage = new MqMessage();
mqMessage.setMessageBody(object);
mqMessage.setMessageId(correlationData.getId());
mqMessage.setExchangeName(exchange);
mqMessage.setQueueName(routingKey);
mqMessage.setRoutingKey(routingKey);
if (StringUtils.isEmpty(messagePostProcessor)) {
this.rabbitTemplate.convertAndSend(exchange, routingKey, mqMessage, correlationData);
} else {
// 发送对应的消息
this.rabbitTemplate.convertAndSend(exchange, routingKey, mqMessage, messagePostProcessor, correlationData);
}
}
/**
* 默认实现发送确认的处理方法
* 子类需要重写该方法，实现自己的业务处理逻辑
*
* @param messageId 消息
* @param ack
* @param cause
*/
public abstract void handleConfirmCallback(String messageId, boolean ack, String cause);
/**
* 默认实现发送匹配不上队列时 回调函数的处理
*
* @param message
* @param replyCode
* @param replyText
* @param routingKey
*/
public abstract void handleReturnCallback(Message message, int replyCode, String replyText,
String routingKey);
/**
* 交换机如果根据自身的类型和路由键匹配上对应的队列时，是否调用returnCallback回调函数
* true: 调用returnCallback回调函数
* false： 不调用returnCallback回调函数 这样在匹配不上对应的队列时，会导致消息丢失
*/
@Value("${spring.message.mandatory:false} ")
private Boolean mandatory;
/**
* 默认队列的优先级
*/
public static final int MESSAGE_PRIORITY = 1;
@PostConstruct
public final void init() {
log.info("sendservice 初始化...... ");
this.rabbitTemplate.setConfirmCallback(this);
this.rabbitTemplate.setReturnCallback(this);
}
/**
* 确认后回调方法
*
* @param correlationData
* @param ack
* @param cause
*/
@Override
public final void confirm(CorrelationData correlationData, boolean ack, String cause) {
log.info("confirm-----correlationData:" + correlationData.toString() + "---ack:" + ack + "----cause:" + cause);
// TODO 记录日志（数据库或者es）
this.handleConfirmCallback(correlationData.getId(), ack, cause);
}
/**
* 失败后回调方法
*
* @param message
* @param replyCode
* @param replyText
* @param exchange
* @param routingKey
*/
@Override
public final void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
log.info("return-----message:" + message.toString() + "---replyCode:" + replyCode + "----replyText:" + replyText + "----exchange:" + exchange + "----routingKey:" + routingKey);
// TODO 记录日志（数据库或者es）
this.handleReturnCallback(message, replyCode, replyText, routingKey);
}
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/service/sender/DefaultSendService.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.service.sender;
import org.springframework.amqp.core.Message;
import org.springframework.stereotype.Service;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 09:20 2021-04-28
* @Modified by:
*/
@Service
public class DefaultSendService extends AbstractSendService {
@Override
public void handleConfirmCallback(String messageId, boolean ack, String cause) {
}
@Override
public void handleReturnCallback(Message message, int replyCode, String replyText, String routingKey) {
}
}
cost-service/src/main/java/com/blt/other/common/rabbitmq/type/ExchangeTypeEnum.java deleted 100644 → 0
package com.blt.other.common.rabbitmq.type;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in 2021-04-28
* @Modified by:
*/
public enum ExchangeTypeEnum {
/**
* direct
*/
DIRECT("direct"),
/**
* topic
*/
TOPIC("topic"),
/**
* fanout
*/
FANOUT("fanout"),
/**
* headers
*/
HEADERS("headers");
private String code;
private ExchangeTypeEnum(String code) {
this.code = code;
}
public String getCode() {
return this.code;
}
public void setCode(String code) {
this.code = code;
}
}
cost-service/src/main/java/com/blt/other/module/log/dto/SearchInput.java 0 → 100644
package com.blt.other.module.log.dto;
import lombok.Data;
import java.time.LocalDate;
import java.time.LocalDateTime;
/**
* @Author: li.yanlin
* @Description：
* @Date: Created in
* @Modified by:
*/
@Data
public class SearchInput {
private LocalDate createTime;
private String url;
private String userId;
private String sourceFrom;
private int pageNum;
private int pageSize;
}
cost-service/src/main/java/com/blt/other/module/log/service/OperationLogService.java
package com.blt.other.module.log.service;
import com.blt.other.module.log.dto.SearchInput;
import com.blt.other.module.log.model.OperationLog;
import java.util.List;
/**
* @Author: li.yanlin
* @Description：
......@@ -10,4 +13,6 @@ import com.blt.other.module.log.model.OperationLog;
*/
public interface OperationLogService {
public void insert(OperationLog log);
public List<OperationLog> select(SearchInput input);
}
cost-service/src/main/java/com/blt/other/module/log/service/impl/OperationLogServiceImpl.java
package com.blt.other.module.log.service.impl;
import com.blt.other.module.log.dao.OperationLogDao;
import com.blt.other.module.log.dto.SearchInput;
import com.blt.other.module.log.model.OperationLog;
import com.blt.other.module.log.service.OperationLogService;
import org.bson.BsonDocument;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.springframework.data.domain.*;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
/**
* @Author: li.yanlin
......@@ -23,10 +33,35 @@ public class OperationLogServiceImpl implements OperationLogService {
@Resource
private OperationLogDao operationLogDao;
@Resource
private MongoTemplate mongoTemplate;
@Override
@Async
public void insert(OperationLog log) {
log.setCreateTime(LocalDateTime.now());
operationLogDao.insert(log);
}
@Override
public List<OperationLog> select(SearchInput input) {
List<OperationLog> list = new ArrayList<>();
Query query = new Query();
/*query.addCriteria(Criteria.where("createTime").in(input.getCreateTime(),input.getCreateTime().plusDays(1)));
if(input.getSourceFrom() != null && !input.getSourceFrom().isEmpty()){
query.addCriteria(Criteria.where("sourceFrom").is(input.getSourceFrom()));
}
StringBuffer query = new StringBuffer("{}");*/
Document document = new Document();
document.put("createTime",String.format("{$lt :%s, $gt : %s}",input.getCreateTime(),input.getCreateTime().plusDays(1)));
/*mongoTemplate.getCollection("OperationLog")
.find(document)
.skip((input.getPageNum() - 1) * input.getPageSize())
.limit(input.getPageSize())
;
mongoTemplate.fi
mongoTemplate.find(x -> x.)*/
return null;
}
}
cost-service/src/test/java/com/blt/other/OtherApplicationTests.java
......@@ -10,17 +10,20 @@ import com.bailuntec.api.bailuntec.oa.response.OaDepartmentResp;
import com.bailuntec.api.bailuntec.oa.response.OaUserResp;
import com.bailuntec.common.JsonUtilByJackson;
import com.bailuntec.common.SpringContextUtil;
import com.bailuntec.common.rabbitmq.service.receiver.DefaultRegisterQueueService;
import com.bailuntec.common.rabbitmq.service.sender.DefaultSendService;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.blt.other.common.config.property.WxWebHook;
import com.blt.other.common.interceptor.mail.MailService;
import com.blt.other.common.rabbitmq.service.receiver.DefaultRegisterQueueService;
import com.blt.other.common.rabbitmq.service.sender.DefaultSendService;
import com.blt.other.common.util.OAuth1Util;
import com.blt.other.database.model.CostCashiercallbackDomain;
import com.blt.other.database.model.CostCompanyDomain;
import com.blt.other.database.model.CostTypeDomain;
import com.blt.other.database.model.UserDomain;
import com.blt.other.module.auth.dao.OaDepartmentMapper;
import com.blt.other.module.auth.dao.OaUserMapper;
import com.blt.other.module.auth.dao.UserDao;
import com.blt.other.module.auth.model.OaDepartment;
import com.blt.other.module.auth.model.OaUser;
import com.blt.other.module.auth.service.IOaDepartmentService;
......@@ -31,6 +34,8 @@ import com.blt.other.module.cost.model.AccountingSubject;
import com.blt.other.module.cost.model.CostCurrentReviewer;
import com.blt.other.module.cost.model.CostDetailDomain;
import com.blt.other.module.cost.model.CostDomain;
import com.blt.other.module.log.dao.OperationLogDao;
import com.blt.other.module.log.model.OperationLog;
import com.blt.other.module.netsuite.dao.NetsuiteLogDao;
import com.blt.other.module.netsuite.dto.NetsuiCostinfoDto;
import com.blt.other.module.netsuite.dto.NetsuiteDataDto;
......@@ -52,18 +57,23 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.http.converter.StringHttpMessageConverter;
import org.springframework.test.annotation.Rollback;
import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestTemplate;
import org.thymeleaf.util.StringUtils;
import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
......@@ -649,5 +659,52 @@ public class OtherApplicationTests {
}
@Resource
private MongoTemplate mongoTemplate;
@Resource
private OperationLogDao operationLogDao;
@Test
public void mongoDb(){
OperationLog log = operationLogDao.findById("614bf979283d810d3c0c27fe").get();
System.out.println(log);
String content = new String(log.getInput());
System.out.println(content);
}
@Resource
UserDao userDao;
@Resource
WxWebHook webHook;
@Resource
RestTemplate restTemplate;
@Test
public void wxMsg(){
CostDomain costDomain = costDao.selectByCostNo("F042350");
String msg = sendWxMsg(costDomain,3075);
System.out.println(msg);
restTemplate.getMessageConverters().set(1,new StringHttpMessageConverter(StandardCharsets.UTF_8));
Object obj = restTemplate.postForEntity(webHook.getUrl()+webHook.getTest(),msg,Object.class);
System.out.println(obj);
}
private String sendWxMsg(CostDomain costDomain, Integer currentUserId) {
UserDomain userDomain = userDao.selectByuserid(currentUserId);
DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
String msg = String.format("{\n" +
" \"msgtype\": \"markdown\",\n" +
" \"markdown\": {\n" +
" \"content\": \"单号【<font color=\\\"info\\\">%s</font>】通过最终审核。\\n" +
">审核人: <font color=\\\"info\\\">%s</font>\\n" +
">审核时间: <font color=\\\"info\\\">%s</font>\"" +
" }\n" +
"}", costDomain.getCostNo(),
userDomain == null ? currentUserId : userDomain.getUsername(),
LocalDateTime.now().format(dtf));
return msg;
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment