SpringBoot整合RabbitMQ——消息的发送和接收
2019-07-03 本文已影响0人
AmosZhu
上篇博文我们整理了RabbitMQ的交换机、队列以及路由绑定等相关知识，并且了解了RabbitMQ是如何发送消息给队列的，以及重要的RoutingKey等重要知识点，这篇博文我们来重点了解下RabbitMQ是如何发送消息，消费消息的，本片博文我们还是以代码为主，简要说明为辅，同时对消息的发送和接收功能进行简单的封装，可以作为一个jar包给第三方进行使用。
本博文是在上一篇博文《SpringBoot整合RabbitMQ——交换机和队列的管理和绑定》的基础上进行重构和新增消息的发送和接收的功能
如果我们要提供一个类似jar包的可以让第三方来作为依赖引入，从而在代码中简单集成我们提供的rabbitMQ的通用的功能，我们的项目需要满足以下的需求：
- 消息发送方需要知道消息是否真正的到达了RabbitMQ,如果发送不到rabbitMQ，如何保证发送方能够处理发送失败的业务数据
- 如何保证消息可靠的达到消费者，并且消费者能否成功消费消息，并且告知发送者消费者的具体消费情况（改业务需求主要是满足消息发送方需要知道消息消费方是否消费成功），对于是否通知消息发送方具体的消费情况需要做到用户自定义
- 为了确保消息的正确和安全，对于消息要做到持久化，防止出现异常导致消息丢失
- 保证客户端对于消息能自主的进行消息的确认，并且对了消费失败的数据能在业务方保存日志、
- 客户端引入我们的依赖，可以很方便的在程序中动态实现队列的注册和交换机的绑定，并且很方便的实现消息的发送以及发送失败的处理
- 客户端能很方便的继承消息的消费，动态的指定处理器自定义处理消息，并且提供消息消费失败（包括业务流程的失败和系统代码错误引起的失败）后消息回传的功能，回传功能做到业务可控
- 在客户端消费消息存在日志记录，并且能与业务逻辑解耦，实现业务与日志分离
- 能针对客户端发出的各类不同类型的消息进行处理，并且是实现各种不同业务的处理扩展
消息的发送
在进行消息的发送之前，我们需要了解下参数mandatory：
- 当其值为true时，交换器无法根据自身的类型和路由键匹配到符合条件的队列，这是rabbitMQ就会通过回调函数将消息返回给生产者。
- 当其值为false时，如果出现上述情形，则消息会丢失
如果需要处理发送rabbitMQ失败的话，在SpringBoot中我们需要在配置文件中配置如下：
spring:
rabbitmq:
template:
mandatory: true
publisher-confirms: true
publisher-returns: true
对应的代码如下：
// RabbitMQConfig类中添加属性
/**
* 消息发送失败，是否回调给发送者
*/
@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.addresses);
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;
}
以上我们就完成了配置类的修改，下面我们来对发送消息的方法进行封装，并且支持用户自定义相关属性
在发送消息之前，我们需要先创建队列，并且将交换机（这里采用默认的交换机mq.direct）和队列进行绑定，路由键就设置成队列名，方法中提供自定义的绑定方法，如有需要可以自行进行封装使用
/**
* Copyright © 2018 五月工作室. All rights reserved.
*
* @Package com.amos.common.send
* @ClassName SendService
* @Description 发送消息的抽象类，子类可以实现该类来处理对应的业务逻辑
* <p/>
* 抽象类实现了ConfirmCallback和ReturnCallback接口，
* confirmCallback来实现业务日志记录，并且自定义处理各自的业务处理逻辑
* returnCallback来实现消息发送失败时的业务处理，并且自定义各自的业务处理逻辑
* @Author Amos
* @Modifier
* @Date 2019/7/1 15:11
* @Version 1.0
**/
public abstract class AbstractSendService implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
public final Log logger = LogFactory.getLog(this.getClass());
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)) {
RabbitMQExceptionUtils.throwRabbitMQException("发送的队列不能为空");
}
if (StringUtils.isEmpty(content)) {
RabbitMQExceptionUtils.throwRabbitMQException("内容不能为空");
}
this.send(MqExchange.DEFAULT_DIRECT_EXCHANGE, queue, content, null, UUIDUtils.generateUuid());
}
/**
* 发送一条有过期时间的消息
*
* @param queue 队列，默认是跟路由键相同的
* @param content 发送的内容
* @param expireTime 过期时间 时间毫秒
*/
public void send(String queue, String content, int expireTime) {
if (StringUtils.isEmpty(queue)) {
RabbitMQExceptionUtils.throwRabbitMQException("发送的队列不能为空");
}
if (StringUtils.isEmpty(content)) {
RabbitMQExceptionUtils.throwRabbitMQException("内容不能为空");
}
MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
// 设置消息的过期时间
message.getMessageProperties().setExpiration(expireTime + "");
return message;
}
};
this.send(MqExchange.DEFAULT_DIRECT_EXCHANGE, queue, content, messagePostProcessor, UUIDUtils.generateUuid());
}
/**
* 按照给定的交换机、路由键、发送内容、发送的自定义属性来发送消息
* 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)) {
RabbitMQExceptionUtils.throwRabbitMQException("交换机不能为空");
}
if (StringUtils.isEmpty(routingKey)) {
RabbitMQExceptionUtils.throwRabbitMQException("路由键不能为空");
}
if (StringUtils.isEmpty(object)) {
RabbitMQExceptionUtils.throwRabbitMQException("发送的内容不能为空");
}
CorrelationData correlationData = new CorrelationData();
correlationData.setId(StringUtils.isEmpty(messageId) ? UUIDUtils.generateUuid() : 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() {
this.logger.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) {
this.logger.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) {
this.logger.info("return-----message:" + message.toString() + "---replyCode:" + replyCode + "----replyText:" + replyText + "----exchange:" + exchange + "----routingKey:" + routingKey);
// TODO 记录日志（数据库或者es）
this.handleReturnCallback(message, replyCode, replyText, routingKey);
}
}
消息的接收
消息发送RabbitMQ之后，我们需要定义监听来监控队列， 并且消费队列上的消息，本类方法中对消息消费进行了封装，添加了消费信息日志和状态的记录，并且支持用户自定义消费方法。消费完成之后，可以自定义设置是否返回给消息发送者消息消费的具体情况，并且针对不同类型的消息 ，封装了命令模式来处理不同类型的消息，方便用户后期对消息的处理的扩展，具体的代码可以参考Gitee上项目
rabbitmq
主要代码逻辑如下:
- 注册队列，并且使用队列名来与交换机进行绑定
- 为该队列添加好消息接收处理
- 封装消息接收的处理逻辑，并且对于消息接收失败的进行重发
下面是重要代码说明：
注册队列，并且为该队列设置消息监听
/**
* Copyright © 2018 五月工作室. All rights reserved.
*
* @Project: rabbitmq
* @ClassName: RegisterQueue
* @Package: com.amos.common.register
* @author: zhuqb
* @Description: 注册队列并且设置监听
* @date: 2019/7/2 0002 下午 15:32
* @Version: V1.0
*/
@Data
public abstract class AbstractRegisterQueue {
public final Log logger = LogFactory.getLog(this.getClass());
@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 (this.logger.isDebugEnabled()) {
e.printStackTrace();
}
return Boolean.FALSE;
}
}
return tag;
}
}
上面类主要是用来注册队列，并且注册成功之后为其新增消息监听类
我们再来新增消息监听以及消息接收处理的代码：
/**
* Copyright © 2018 五月工作室. All rights reserved.
*
* @Package com.amos.common.listen
* @ClassName AbstractMessageHandle
* @Description 队列设置监听基类
* @Author Amos
* @Modifier
* @Date 2019/7/1 20:21
* @Version 1.0
**/
@Component
public class MessageListen {
public final Log logger = LogFactory.getLog(this.getClass());
@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();
this.logger.info("------ 已成功监听异步消息触发通知队列：" + queue + " ------");
}
}
指明队列的监听类，并且维护是否手动ack消息
/**
* Copyright © 2018 五月工作室. All rights reserved.
*
* @Package com.amos.common.listen
* @ClassName AbstractMessageHandler
* @Description 消息接收处理类
* <p/>
* 实现 ChannelAwareMessageListener接口 重写onMessage方法来实现业务的处理
* @Author Amos
* @Modifier
* @Date 2019/7/1 20:09
* @Version 1.0
**/
@Component
public abstract class AbstractMessageHandler implements ChannelAwareMessageListener {
public final Log logger = LogFactory.getLog(this.getClass());
@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 {
this.logger.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 (this.logger.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) {
this.logger.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.this.logger.info("消息" + mqMessage.toString() + "已签收");
return ++this.count;
}
}, new RecoveryCallback<Integer>() {
@Override
public Integer recover(RetryContext context) throws Exception { //重试多次后都失败了
AbstractMessageHandler.this.logger.info("消息" + mqMessage.toString() + "签收失败");
return Integer.MAX_VALUE;
}
});
if (result.intValue() <= retryTimes) {
//消息签收成功 更改状态
} else {
//MQ服务器或网络出现问题，签收失败 更改状态
}
} catch (Exception e) {
this.logger.error("消息" + mqMessage.toString() + "签收出现异常：" + e.getMessage());
}
} else {
this.logger.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);
this.logger.info("注入队列 " + queue + " 消息签收模式: " + ack.name());
}
}
上面代码主要是封装了消息接收处理的代码逻辑
- 定义抽象类方便让子类来继承实现基类中的方法，其次类实现了
ChannelAwareMessageListener接口，实现onMessage方法的重写 - 该方法中除了进行自身业务的处理，同时也调用业务的自定义消息处理逻辑
- 对于消息接收处理失败后，进行消息重发，并且可以业务进行日志记录
- 在对队列进行设置监听时，指定是否手动ack消息
- 业务新增类继承
AbstractMessageHandler基类，并且将该监听类与队列动态绑定即可
以上就是本博文对消息的发送和接收处理进行的简单的封装，其中核心的业务都已经实现，待后期与elasticsearch集合完善日志记录相关的功能
附： 不同类型消息的处理
之所以不跟上面的代码整合在一起，主要是因为本rabbitMQ的项目主要是为了对rabbitMQ的常用业务进行封装，消息的处理大多数是业务方面的工作，如果整合在一起的会造成代码的耦合，不利于rabbitMQ功能代码的剥离。
不同消息类型的处理的业务流程如下：
- 声明一个接口定义消息处理的通用方法
/**
* Copyright © 2018 五月工作室. All rights reserved.
*
* @Package com.amos.common.send
* @ClassName SendService
* @Description 对消息进行处理
* @Author Amos
* @Modifier
* @Date 2019/7/1 15:11
* @Version 1.0
**/
public interface Receiver {
/**
* 对消息进行处理
*
* @param messageData
* @return
*/
HandleResult handleMessage(MessageData messageData);
}
- 定义消息接收基类，该类实现
Receiver接口,积累中定义不同类型处理的自定义方法以及处理成功和处理失败的业务逻辑
/**
* Copyright © 2018 五月工作室. All rights reserved.
*
* @Package com.amos.common.send
* @ClassName SendService
* @Description 定义通用消息接收处理基类
* @Author Amos
* @Modifier
* @Date 2019/7/1 15:11
* @Version 1.0
**/
public abstract class AbstractReceiver implements Receiver {
private static final Logger logger = LoggerFactory.getLogger(AbstractReceiver.class);
/**
* 用户自定义消息处理
*
* @param messageData
* @return
*/
public abstract HandleResult exec(MessageData messageData) throws Exception;
/**
* 用户自定义验证
*
* @param messageData
* @return
*/
public abstract Result validate(MessageData messageData);
/**
* 成功处理
*
* @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) {
logger.info(this.getClass().getSimpleName() + "-->handleMessage()参数 unicomData:{}", messageData.toString());
HandleResult handleResult = null;
try {
// 如果自定义验证不通过
Result result = this.validate(messageData);
if (!ResultEnum.success().equals(result.getCode())) {
// 如果验证失败 进行失败处理
return this.handleFail(messageData);
}
// 根据自行处理的返回结果
handleResult = this.exec(messageData);
// 执行成功处理的逻辑
handleResult = this.handleSuccess(messageData);
} catch (Exception e) {
e.printStackTrace();
messageData.setContent(e.getMessage());
return this.handleFail(messageData);
}
return handleResult;
}
}
- 针对不同的消息类型实体进行处理 ，这里可以采用命令模式来封装代码，首先定义命令的基类，基类中定义好处理的方法，子类实现该基类，并且设置对应的命令来处理其业务逻辑，该种设计模式将命令与处理者之间进行了松耦合，可以很方便的维护命令与处理者之间关系
//命令的基类
/**
* Copyright © 2018 五月工作室. All rights reserved.
*
* @Package com.amos.common.send
* @ClassName SendService
* @Description 命令抽象类
* <p/>
* 把接收消息的类型封装成一个命令 并且交给指定的接收者出处理
* 方便扩展每个命令的处理
* @Author Amos
* @Modifier
* @Date 2019/7/1 15:11
* @Version 1.0
**/
public abstract class AbstractCommand {
/**
* 每个命令都必须被处理
*
* @param messageData
* @return
*/
public abstract HandleResult execute(MessageData messageData);
}
// 回调命令的处理，在回调命令中设置了抽象处理者，处理者交由子类去具现
/**
* Copyright © 2018 五月工作室. All rights reserved.
*
* @Package com.amos.common.send
* @ClassName SendService
* @Description 回调函数 消息处理
* @Author Amos
* @Modifier
* @Date 2019/7/1 15:11
* @Version 1.0
**/
@Component
public class CallBackCommand extends AbstractCommand {
/**
* 定义handler来进行命令处理
*/
private AbstractHandler handler;
public CallBackCommand(AbstractHandler handler) {
this.handler = handler;
}
public CallBackCommand init(AbstractHandler handler) {
this.handler = handler;
return this;
}
/**
* 执行业务处理
*
* @param unicomData
* @return
*/
@Override
public HandleResult execute(MessageData unicomData) {
return this.handler.handle(unicomData);
}
}
- 定义具体的处理者来处理不同的消息
// 定义处理者基类
/**
* Copyright © 2018 五月工作室. All rights reserved.
*
* @Package com.amos.common.send
* @ClassName SendService
* @Description 处理消息 抽象类
* <p/>
* 业务处理需要继承该基类，实现处理的方法
* @Author Amos
* @Modifier
* @Date 2019/7/1 15:11
* @Version 1.0
**/
public abstract class AbstractHandler {
/**
* 自定义处理
*
* @param data
* @return
*/
public abstract HandleResult handle(MessageData data);
}
// 子类实现处理者基类并且实现具体的处理方法
/**
* Copyright © 2018 五月工作室. All rights reserved.
*
* @Package com.amos.common.send
* @ClassName SendService
* @Description 回调消息处理者
* @Author Amos
* @Modifier
* @Date 2019/7/1 15:11
* @Version 1.0
**/
@Component
public class CallBackHandler extends AbstractHandler {
/**
* 修改消息
*
* @param data
* @return
*/
@Override
public HandleResult handle(MessageData data) {
// TODO 自定义业务逻辑处理
return new HandleResult.CallBack(true).callback(false).msg("处理成功").builder();
}
}
