Skip to content

M
midjourney-proxy
Commit 911a7f82 by songxiang
Options

RabbitMQ迁移

parent b48a6003
Showing with 119 additions and 290 deletions
pcloud-common-core/src/main/java/com/pcloud/common/core/biz/ExapiQueueBizImpl.java
package com.pcloud.common.core.biz;
import com.pcloud.common.core.dto.ExapiMessageDto;
import com.pcloud.common.exceptions.BizException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Component;
import javax.jms.JMSException;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import com.pcloud.common.core.aspect.ParamLog;
import com.pcloud.common.core.constant.MQQueueConstant;
import com.pcloud.common.core.dto.ExapiMessageDto;
@Component("exapiQueueBiz")
public class ExapiQueueBizImpl implements ExapiQueueBiz{
private static final Logger LOGGER = LoggerFactory.getLogger(ExapiQueueBiz.class);
public class ExapiQueueBizImpl implements ExapiQueueBiz {
@Autowired(required = false)
@Qualifier("jmsExapiTemplate")
private JmsTemplate jmsExapiTemplate;
@Autowired
private AmqpTemplate amqpTemplate;
@Override
public void sendMessageQueue(ExapiMessageDto exapiMessageDto) {
MessageCreator messageCreator = new MessageCreator() {
@Override
public ObjectMessage createMessage(Session session) throws JMSException {
return session.createObjectMessage(exapiMessageDto);
}
};
@Override
@ParamLog("发送对外API queue")
public void sendMessageQueue(ExapiMessageDto exapiMessageDto) {
amqpTemplate.convertAndSend(MQQueueConstant.EXAPI, exapiMessageDto);
}
try {
jmsExapiTemplate.send(messageCreator);
} catch (Exception e) {
LOGGER.error("发送失败，" + e.getMessage() + "," + exapiMessageDto, e);
throw BizException.SEND_QUEUE_FAIL;
}
LOGGER.info("发送成功，" + exapiMessageDto);
}
}
pcloud-common-core/src/main/java/com/pcloud/common/core/biz/FrontEventBiz.java
package com.pcloud.common.core.biz;
import com.pcloud.common.core.dto.ConvertQueueDto;
import com.pcloud.common.core.dto.FrontEventDto;
import com.pcloud.common.exceptions.BizException;
......@@ -10,10 +9,10 @@ import com.pcloud.common.exceptions.BizException;
* @description:
*/
public interface FrontEventBiz {
/**
* 发送事件消息
*
* @param frontEventDto
*/
public void sendFrontEventQueue(FrontEventDto frontEventDto) throws BizException;
/**
* 发送事件消息
*
* @param frontEventDto
*/
public void sendFrontEventQueue(FrontEventDto frontEventDto) throws BizException;
}
pcloud-common-core/src/main/java/com/pcloud/common/core/biz/FundQueueBizImpl.java
package com.pcloud.common.core.biz;
import javax.jms.JMSException;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Component;
import com.pcloud.common.core.aspect.ParamLog;
import com.pcloud.common.core.constant.MQQueueConstant;
import com.pcloud.common.core.dto.FundMessageDto;
import com.pcloud.common.exceptions.BizException;
/**
* @description 基金资源变更记录消息队列实现类
* @author PENG
* @date 2017年10月18日 下午2:11:52
* @version 1.0
* @description 基金资源变更记录消息队列实现类
* @date 2017年10月18日 下午2:11:52
*/
@Component("fundQueueBiz")
public class FundQueueBizImpl implements FundQueueBiz {
......@@ -29,29 +24,20 @@ public class FundQueueBizImpl implements FundQueueBiz {
*/
private final static Logger logger = LoggerFactory.getLogger(FundQueueBizImpl.class);
@Autowired(required = false)
@Qualifier("jmsFundTemplate")
private JmsTemplate jmsFundTemplate;
@Autowired
private AmqpTemplate amqpTemplate;
/**
* 发送队列
*/
@Override
@ParamLog("发送基金资源变更记录消息QUEUE")
public void sendFundMessage(FundMessageDto fundMessageDto) throws BizException {
MessageCreator messageCreator = new MessageCreator() {
@Override
public ObjectMessage createMessage(Session session) throws JMSException {
return session.createObjectMessage(fundMessageDto);
}
};
try {
jmsFundTemplate.send(messageCreator);
amqpTemplate.convertAndSend(MQQueueConstant.FUND, fundMessageDto);
} catch (Exception e) {
logger.error("发送失败，" + e.getMessage() + "," + fundMessageDto.toString(), e);
throw BizException.SEND_QUEUE_FAIL;
}
logger.info("发送成功，" + fundMessageDto.toString());
}
}
pcloud-common-core/src/main/java/com/pcloud/common/core/biz/MerchantDynamicQueueBizImpl.java
package com.pcloud.common.core.biz;
import com.pcloud.common.core.dto.DynamicDto;
import com.pcloud.common.exceptions.BizException;
import org.apache.activemq.command.ActiveMQQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Component;
import javax.jms.JMSException;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import com.pcloud.common.core.aspect.ParamLog;
import com.pcloud.common.core.constant.MQQueueConstant;
import com.pcloud.common.core.dto.DynamicDto;
import com.pcloud.common.exceptions.BizException;
/**
* @描述：
* @作者：lucas
* @创建时间：2017年12月14日,15:58
* @版本：1.0
*
* @作者：lucas @创建时间：2017年12月14日,15:58 @版本：1.0
*/
@Component("merchantDynamicQueueBiz")
public class MerchantDynamicQueueBizImpl implements MerchantDynamicQueueBiz {
private final static Logger logger= LoggerFactory.getLogger(MerchantDynamicQueueBizImpl.class);
/**
* 消息模板
*/
@Autowired(required=false)
@Qualifier("jmsDynamicTemplate")
private JmsTemplate jmsDynamicTemplate;
@Autowired
private AmqpTemplate amqpTemplate;
@Override
public void send(DynamicDto dynamicDto) throws BizException {
MessageCreator messageCreator = new MessageCreator() {
@Override
public ObjectMessage createMessage(Session session) throws JMSException {
return session.createObjectMessage(dynamicDto);
}
};
try {
jmsDynamicTemplate.send(messageCreator);
} catch (Exception e) {
logger.error("发送失败，"+e.getMessage()+","+dynamicDto,e);
throw BizException.SEND_QUEUE_FAIL;
}
@Override
@ParamLog("动态监控QUEUE")
public void send(DynamicDto dynamicDto) throws BizException {
amqpTemplate.convertAndSend(MQQueueConstant.MERCHANT_DYNAMIC, dynamicDto);
}
logger.info("发送成功，"+dynamicDto);
}
}
pcloud-common-core/src/main/java/com/pcloud/common/core/biz/MessageBiz.java
......@@ -2,6 +2,7 @@ package com.pcloud.common.core.biz;
import com.pcloud.common.core.dto.SendAppMessageDto;
import com.pcloud.common.core.dto.SendEmailDto;
import com.pcloud.common.core.dto.SendMessageDto;
import com.pcloud.common.core.dto.SendNotifyDto;
......@@ -37,5 +38,10 @@ public interface MessageBiz {
* @param sendSMSDto
*/
void sendSMS(SendMessageDto smsParam);
/**
* 短信推送
*/
void sendAppMessage(SendAppMessageDto sendAppMessageDto);
}
pcloud-common-core/src/main/java/com/pcloud/common/core/biz/MessageBizImpl.java
package com.pcloud.common.core.biz;
import java.io.IOException;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Component;
import com.pcloud.common.core.aspect.ParamLog;
import com.pcloud.common.core.constant.MQQueueConstant;
import com.pcloud.common.core.dto.SendAppMessageDto;
import com.pcloud.common.core.dto.SendEmailDto;
import com.pcloud.common.core.dto.SendMessageDto;
import com.pcloud.common.core.dto.SendNotifyDto;
......@@ -20,148 +16,82 @@ import com.pcloud.common.utils.mq.MqMessage;
import com.pcloud.common.utils.mq.MqMessageTypeEnum;
/**
*
* @描述：邮件业务实现
* @作者：shichunshan
* @创建时间：2016年6月24日,下午2:19:27
* @版本：1.0
* @创建时间：2016年6月24日,下午2:19:27 @版本：1.0
*/
@Component("mailBiz")
public class MessageBizImpl implements MessageBiz{
private final static Logger logger=LoggerFactory.getLogger(MessageBizImpl.class);
@Component("messageQueueBiz")
public class MessageBizImpl implements MessageBiz {
private final static Logger logger = LoggerFactory.getLogger(MessageBizImpl.class);
@Autowired
private AmqpTemplate amqpTemplate;
/**
* 消息模板
* 发送邮件
*
* @param sendEmailDto
*/
@Autowired(required = false)
@Qualifier("jmsMessageTemplate")
private JmsTemplate jmsMessageTemplate;
@Override
@ParamLog("发送邮件")
public void sendEmail(SendEmailDto sendEmailDto) {
MessageCreator messageCreator = new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
MqMessage mqMessage = null;
try {
mqMessage = formateMessage(sendEmailDto);
logger.info("mqMessage+++++++++++" + mqMessage);
} catch (IOException e) {
logger.error("邮件发送失败，" + sendEmailDto);
return null;
}
return session.createObjectMessage(mqMessage);
}
};
try {
jmsMessageTemplate.send(messageCreator);
ObjectMapper objectMapper = new ObjectMapper();
String content = objectMapper.writeValueAsString(sendEmailDto);
MqMessage mqMessage = new MqMessage(MqMessageTypeEnum.EMAIL, content);
amqpTemplate.convertAndSend(MQQueueConstant.MESSAGE, mqMessage);
} catch (Exception e) {
logger.error("邮件发送失败，" + e.getMessage() + "," + sendEmailDto, e);
return;
}
logger.info("邮件发送成功，" + sendEmailDto);
}
/**
* 格式化邮件内容
*
* @param mailParam
* 邮件参数
* @return 队列消息
* @throws IOException
* 序列化为json错误
* 发送站内信
*/
private MqMessage formateMessage(SendEmailDto sendEmailDto) throws IOException {
ObjectMapper objectMapper = new ObjectMapper();
String content = objectMapper.writeValueAsString(sendEmailDto);
return new MqMessage(MqMessageTypeEnum.EMAIL, content);
}
@Override
@ParamLog("发送站内信")
public void sendLetter(SendNotifyDto sendNotifyDto) {
MessageCreator messageCreator = new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
MqMessage mqMessage = null;
try {
mqMessage = formateMessage(sendNotifyDto);
logger.info("mqMessage+++++++++++" + mqMessage);
} catch (IOException e) {
logger.error("站内信发送失败，" + sendNotifyDto);
return null;
}
return session.createObjectMessage(mqMessage);
}
};
try {
jmsMessageTemplate.send(messageCreator);
ObjectMapper objectMapper = new ObjectMapper();
String content = objectMapper.writeValueAsString(sendNotifyDto);
MqMessage mqMessage = new MqMessage(MqMessageTypeEnum.LETTER, content);
amqpTemplate.convertAndSend(MQQueueConstant.MESSAGE, mqMessage);
} catch (Exception e) {
logger.error("站内信发送失败，" + e.getMessage() + "," + sendNotifyDto, e);
return;
}
logger.info("站内信发送成功，" + sendNotifyDto);
}
/**
* 格式化站内信内容
*
* @param sendNotifyDto
* 站内信参数
* @return 队列消息
* @throws IOException
* 序列化为json错误
* 发送短信
*/
private MqMessage formateMessage(SendNotifyDto sendNotifyDto) throws IOException {
ObjectMapper objectMapper = new ObjectMapper();
String content = objectMapper.writeValueAsString(sendNotifyDto);
return new MqMessage(MqMessageTypeEnum.LETTER, content);
}
@Override
public void sendSMS(SendMessageDto smsParam) {
MessageCreator messageCreator = new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
MqMessage mqMessage = null;
try {
mqMessage = formateMessage(smsParam);
logger.info("mqMessage+++++++++++" + mqMessage);
} catch (IOException e) {
logger.error("短信发送失败，" + smsParam);
return null;
}
return session.createObjectMessage(mqMessage);
}
};
@ParamLog("发送短信")
public void sendSMS(SendMessageDto sendMessageDto) {
try {
jmsMessageTemplate.send(messageCreator);
ObjectMapper objectMapper = new ObjectMapper();
String content = objectMapper.writeValueAsString(sendMessageDto);
MqMessage mqMessage = new MqMessage(MqMessageTypeEnum.SHORT_MESSAGE, content);
amqpTemplate.convertAndSend(MQQueueConstant.MESSAGE, mqMessage);
} catch (Exception e) {
logger.error("短信发送失败，" + e.getMessage() + "," + smsParam, e);
return;
logger.error("短信发送失败，" + e.getMessage() + "," + sendMessageDto, e);
}
logger.info("短信发送成功，" + smsParam);
}
/**
* 格式化内容
*
* @param sendSMSDto
* 邮件参数
* @return 队列消息
* @throws IOException
* 序列化为json错误
* 发送app推送
*/
private MqMessage formateMessage(SendMessageDto smsParam) throws IOException {
ObjectMapper objectMapper = new ObjectMapper();
String content = objectMapper.writeValueAsString(smsParam);
return new MqMessage(MqMessageTypeEnum.SHORT_MESSAGE, content);
@Override
@ParamLog("发送APP推送")
public void sendAppMessage(SendAppMessageDto sendAppMessageDto) {
try {
ObjectMapper objectMapper = new ObjectMapper();
String content = objectMapper.writeValueAsString(sendAppMessageDto);
MqMessage mqMessage = new MqMessage(MqMessageTypeEnum.APP_MESSAGE, content);
amqpTemplate.convertAndSend(MQQueueConstant.MESSAGE, mqMessage);
} catch (Exception e) {
logger.error("APP推送发送失败，" + e.getMessage() + "," + sendAppMessageDto, e);
}
}
}
pcloud-common-core/src/main/java/com/pcloud/common/core/biz/OperationLogQueueBizImpl.java
package com.pcloud.common.core.biz;
import javax.jms.JMSException;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;
import com.pcloud.common.core.aspect.ParamLog;
import com.pcloud.common.core.constant.MQQueueConstant;
import com.pcloud.common.core.dto.OperationLogDto;
import com.pcloud.common.exceptions.BizException;
@Service("operationLogQueueBiz")
public class OperationLogQueueBizImpl implements OperationLogQueueBiz {
/**
*
*/
private final static Logger logger = LoggerFactory.getLogger(OperationLogQueueBizImpl.class);
/**
* 消息模板
*/
@Autowired(required = false)
@Qualifier("jmsLogTemplate")
private JmsTemplate jmsLogTemplate;
@Autowired
private AmqpTemplate amqpTemplate;
@Override
@ParamLog("日志记录QUEUE")
public void send(OperationLogDto operationLog) throws BizException {
MessageCreator messageCreator = new MessageCreator() {
@Override
public ObjectMessage createMessage(Session session) throws JMSException {
return session.createObjectMessage(operationLog);
}
};
try {
jmsLogTemplate.send(messageCreator);
} catch (Exception e) {
logger.error("发送失败，" + e.getMessage() + "," + operationLog, e);
throw BizException.SEND_QUEUE_FAIL;
}
logger.info("发送成功，" + operationLog);
amqpTemplate.convertAndSend(MQQueueConstant.LOG, operationLog);
}
}
pcloud-common-core/src/main/java/com/pcloud/common/core/biz/WeektaskQueueBizImpl.java
package com.pcloud.common.core.biz;
import com.pcloud.common.core.dto.WeektaskMessageDto;
import com.pcloud.common.exceptions.BizException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Component;
import javax.jms.JMSException;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import com.pcloud.common.core.aspect.ParamLog;
import com.pcloud.common.core.constant.MQQueueConstant;
import com.pcloud.common.core.dto.WeektaskMessageDto;
import com.pcloud.common.exceptions.BizException;
/**
* 编辑周任务队列实现
......@@ -21,37 +16,18 @@ import javax.jms.Session;
* @create 2017-09-21 09:33
**/
@Component("weektaskQueueBiz")
public class WeektaskQueueBizImpl implements WeektaskQueueBiz{
private final static Logger logger= LoggerFactory.getLogger(WeektaskQueueBizImpl.class);
@Autowired(required=false)
@Qualifier("jmsWeektaskTemplate")
private JmsTemplate jmsWeekstaskTemplate;
// @Autowired(required= false)
// @Qualifier("raystask")
// private ActiveMQQueue templateQueue;
/**
* 发送队列
*/
@Override
public void sendMessageQueue(WeektaskMessageDto weektaskMessageDto) throws BizException {
MessageCreator messageCreator = new MessageCreator() {
@Override
public ObjectMessage createMessage(Session session) throws JMSException {
return session.createObjectMessage(weektaskMessageDto);
}
};
try {
jmsWeekstaskTemplate.send(messageCreator);
} catch (Exception e) {
logger.error("发送失败，" + e.getMessage() + "," + weektaskMessageDto, e);
throw BizException.SEND_QUEUE_FAIL;
}
logger.info("发送成功，" + weektaskMessageDto);
}
public class WeektaskQueueBizImpl implements WeektaskQueueBiz {
@Autowired
private AmqpTemplate amqpTemplate;
/**
* 发送队列
*/
@Override
@ParamLog("红榜任务QUEUE")
public void sendMessageQueue(WeektaskMessageDto weektaskMessageDto) throws BizException {
amqpTemplate.convertAndSend(MQQueueConstant.WEEK_TASK, weektaskMessageDto);
}
}
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