使用策略模式 + 函数式接口优化 MQ 消息处理

原创

Java king

修改于 2025-05-28 13:44:49

2900

文章被收录于专栏：后端Java后端Java

一、背景说明

在传统消息队列（MQ）消费处理场景中，我们通常采用大量 if-else 或 switch-case 结构来根据不同的消息类型执行不同的业务逻辑。

存在的问题：

if-else 嵌套层级深、代码不易维护
新增类型需要修改原有逻辑，违反开闭原则
业务逻辑与消息处理耦合度高

二、实现目标

✅ 消除冗余的 if-else 分支结构
✅ 解耦消息处理逻辑与调用方式
✅ 利用 Java 8 函数式接口优雅传递处理方法
✅ 实现更强的可维护性和扩展性

三、核心实现结构

1. 定义处理逻辑枚举（策略集合）

public enum EvaluationEnum {
    SURVEY_DATA_GENERATE("XXX", EvaluationUtil::surveyDataGenerate),
    PERSONNEL_DATA_GENERATE("XXX", EvaluationUtil::personnelDataGenerate);

    private final String desc;
    private final Consumer<EvaluationDto> setter;

    EvaluationEnum(String desc, Consumer<EvaluationDto> setter) {
        this.desc = desc;
        this.setter = setter;
    }

    public Consumer<EvaluationDto> getSetter() {
        return setter;
    }
}

每个枚举值代表一种处理策略
使用 Consumer<EvaluationDto> 函数式接口传入对应处理方法
描述信息 desc 用于前端展示或日志用途

2. 消息监听者中调用策略

@Service("evaluationListener")
public class EvaluationListener implements ChannelAwareMessageListener {

    private static final Logger log = LogManager.getLogger(EvaluationListener.class);

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        try {
            BasQueueTask basQueueTask = JSONObject.parseObject(new String(message.getBody()), BasQueueTask.class);
            EvaluationDto evaluationDto = JsonUtil.fromJsonToBean(basQueueTask.getMsgBody().toString(), EvaluationDto.class);
            EvaluationEnum.valueOf(basQueueTask.getScene()).getSetter().accept(evaluationDto);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            log.error("[EvaluationListener]消息发送异常，已入死信队列。消息内容：{},e:", JsonUtil.toJson(message), e);
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
    }
}

3. 发送端设置场景值

BasQueueTask basQueueTask = new BasQueueTask();
basQueueTask.setMsgBody(evaluationDto);
basQueueTask.setScene(
    evaluationDto.getReportType().equals(ReportManageTypeEnum.SURVEY.getId()) 
    ? EvaluationEnum.SURVEY_DATA_GENERATE.name() 
    : EvaluationEnum.PERSONNEL_DATA_GENERATE.name()
);

rabbitTemplate.convertAndSend(
    MsgQueueEnum.RPT_EVALUATION.getExchangeName(),
    MsgQueueEnum.RPT_EVALUATION.getQueueName(),
    basQueueTask
);

四、函数式接口说明

常用接口

接口名	参数数量	返回值	用途
`Consumer<T>`	1	无	消费一个输入参数
`BiConsumer<T,U>`	2	无	消费两个输入参数（例如带条件处理）

示例：

Consumer<String> log = System.out::println;
log.accept("Hello World");

BiConsumer<String, Integer> printer = (label, value) -> 
    System.out.println(label + ": " + value);

printer.accept("数量", 10);