最近项目中使用mqtt监听消息，再根据消息做各种处理，使用大量的if else，代码异常难维护，

参考的地址找不到了。。。

以下为实现：

接收mqtt的接口类

package com.mhm.mqttlistener;import org.apache.log4j.Logger;public interface MqttService {public void send(String topic, String content) throws Exception;public void handler(String message) throws Exception;}

接收mqtt消息的实现类

package com.mhm.mqttlistener;import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;import com.alibaba.fastjson.JSON;
import com.mhm.dto.api.mqtt.MqttCmdDto;
import com.mhm.mqttlistener.executor.MqttExecutor;@Component
public class MqttServiceImpl implements MqttService {@Autowiredprivate MqttPahoMessageHandler mqttHandler;@AutowiredMqttExecutor mqttExecutor;static final Logger log = Logger.getLogger(MqttService.class);@Overridepublic void send(String topic, String content) throws Exception {// 构建消息Message<String> messages = MessageBuilder.withPayload(content).setHeader(MqttHeaders.TOPIC, topic).build();// 发送消息mqttHandler.handleMessage(messages);}@Overridepublic void handler(String message) throws Exception {// log.info("收到消息:" + message);// KuaiyueCmdDto cmdDto = JSON.parseObject(message, KuaiyueCmdDto.class);if(!StringUtils.isBlank(message)) {// MqttMsgExecutor servcice = new MqttMsgExecutor();mqttExecutor.process(message);} else {log.info("消息为空");}}}

把打了注解的类遍历出来后放入map，(注解类在下面)

package com.mhm.mqttlistener.executor;import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;import javax.annotation.PostConstruct;import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;import com.mhm.enumeration.EnumMqttMsgtype;
import com.mhm.framework.aspect.mqttmsgtype.MqttmsgtypeAspect;
import com.mhm.utils.SpringContextHolder;@Service
public class MqttExecutorFactory {private static Map<EnumMqttMsgtype, MqttExecutorService> mqttMsgTypeMap = new ConcurrentHashMap<>();// SpringContextHolder springContextHolder;//工厂将 Spring 装配的相关的 Bean 用 Map 保存起来@PostConstructpublic void init() {Map<String, Object> beanMap = SpringContextHolder.getApplicationContext().getBeansWithAnnotation(MqttmsgtypeAspect.class);for(Object service : beanMap.values()) {MqttmsgtypeAspect annotation = service.getClass().getAnnotation(MqttmsgtypeAspect.class);mqttMsgTypeMap.put(annotation.value(), (MqttExecutorService)service);}}public MqttExecutorService createMqttExecutor(EnumMqttMsgtype msgtype) {return mqttMsgTypeMap.get(msgtype);}
}

pring-application.xml中加入SpringContextHolder的声明

<bean id="springContextHolder" class="com.mhm.utils.SpringContextHolder"lazy-init="false"></bean>

package com.mhm.utils;import java.util.Map;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;import com.mhm.mng.GroupInfoMng;public class SpringContextHolder implements ApplicationContextAware {private static ApplicationContext applicationContext;//实现ApplicationContextAware接口的context注入函数, 将其存入静态变量.public void setApplicationContext(ApplicationContext applicationContext) {SpringContextHolder.applicationContext = applicationContext;}//取得存储在静态变量中的ApplicationContext.public static ApplicationContext getApplicationContext() {checkApplicationContext();return applicationContext;}//从静态变量ApplicationContext中取得Bean, 自动转型为所赋值对象的类型.@SuppressWarnings("unchecked")public static <T> T getBean(String name) {checkApplicationContext();return (T) applicationContext.getBean(name);}//从静态变量ApplicationContext中取得Bean, 自动转型为所赋值对象的类型.//从静态变量ApplicationContext中取得Bean, 自动转型为所赋值对象的类型.//如果有多个Bean符合Class, 取出第一个.@SuppressWarnings("unchecked")public static <T> T getBean(Class<T> clazz) {checkApplicationContext();@SuppressWarnings("rawtypes")Map beanMaps = applicationContext.getBeansOfType(clazz);if (beanMaps!=null && !beanMaps.isEmpty()) {return (T) beanMaps.values().iterator().next();} else{return null;}}private static void checkApplicationContext() {if (applicationContext == null) {throw new IllegalStateException("applicaitonContext未注入,请在applicationContext.xml中定义SpringContextHolder");}}}

Enum消息类型

package com.mhm.enumeration;public enum EnumMqttMsgtype {/*** 开灯*/OPEN("开灯"),/*** 关灯*/CLOSE("关灯"),/*** 调光*/DIMMER("调光"),/*** NULL*/NULL("空");private String desc;EnumMqttMsgtype(String desc) {this.desc = desc;}public String getDesc() {return desc;}public void setDesc(String desc) {this.desc = desc;}}

MqttExecutor ，处理消息入口类

package com.mhm.mqttlistener.executor;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import com.alibaba.fastjson.JSON;
import com.cz.highland.dto.OnenetPushDataDto;
import com.mhm.dto.api.mqtt.MqttCmdDto;
import com.mhm.enumeration.EnumMqttMsgtype;@Component
public class MqttExecutor {@AutowiredMqttExecutorFactory mqttExecutorFactory;public void process(String message) {try {MqttCmdDto cmdDto = JSON.parseObject(message, MqttCmdDto.class);EnumMqttMsgtype msgtype = EnumMqttMsgtype.valueOf(cmdDto.getMsgType() == null ? "NULL" : cmdDto.getMsgType().toUpperCase());MqttExecutorService myMqttService = mqttExecutorFactory.createMqttExecutor(msgtype);myMqttService.execute(cmdDto);} catch (Exception ex) {ex.printStackTrace();}}
}

ExecutorBase ，每个子处理类的基类

package com.mhm.mqttlistener.executor;import org.apache.commons.lang3.StringUtils;import com.mhm.dto.api.mqtt.MqttCmdDto;public class ExecutorBase {protected String generateCodes(MqttCmdDto dto) {StringBuilder code = new StringBuilder();if(dto.getSnGroup() != null && !dto.getSnGroup().isEmpty()) {for (int i = 0; i < dto.getSnGroup().size(); i++) {if(StringUtils.isBlank(code)) {code.append(dto.getSnGroup().get(i).getSerialNumber());} else {code.append(",").append(dto.getSnGroup().get(i).getSerialNumber());}}}return code.toString();}
}

MqttExecutorService ，子处理类的接口

package com.mhm.mqttlistener.executor;import com.mhm.dto.api.mqtt.MqttCmdDto;public interface MqttExecutorService {public void execute(MqttCmdDto dto);}

开灯的处理子类

package com.mhm.mqttlistener.executor.impl;import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import com.mhm.dto.BusinessRtnMsg;
import com.mhm.dto.OperatorDto;
import com.mhm.dto.api.mqtt.MqttCmdDto;
import com.mhm.entity.Userinfo;
import com.mhm.enumeration.EnumMqttMsgtype;
import com.mhm.framework.aspect.mqttmsgtype.MqttmsgtypeAspect;
import com.mhm.framework.common.Constants;
import com.mhm.mng.CommandMng;
import com.mhm.mqttlistener.executor.ExecutorBase;
import com.mhm.mqttlistener.executor.MqttExecutorService;
import com.mhm.utils.LedConstants;@Component("myMqttOpen")
@MqttmsgtypeAspect(value = EnumMqttMsgtype.OPEN)
public class ExecutorServiceImpl_Open extends ExecutorBase implements MqttExecutorService {@Autowiredprivate CommandMng commandMng;@Overridepublic void execute(MqttCmdDto dto) {System.out.println("open...");// 需判断该topic是否对该资源有open操作String codes = generateCodes(dto);OperatorDto operatorDto = new OperatorDto();operatorDto.setOperatorType(Constants.INT_VALUE2);operatorDto.setClient(LedConstants.LOG_CLIENT_MQTT);operatorDto.setNodeLevel(LedConstants.CONTROLLER_NODE_TYPE); // setNodeLevel必须在SetControllerType之前operatorDto.setControllerType(Constants.INT_VALUE0);operatorDto.setParam1(codes);operatorDto.setParam2("1");Userinfo user = new Userinfo(1);BusinessRtnMsg brmsg = commandMng.cmd_api_v2_switchLight(user, operatorDto);}}

关灯的处理子类：

package com.mhm.mqttlistener.executor.impl;import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import com.mhm.dto.BusinessRtnMsg;
import com.mhm.dto.OperatorDto;
import com.mhm.dto.api.mqtt.MqttCmdDto;
import com.mhm.entity.Userinfo;
import com.mhm.enumeration.EnumMqttMsgtype;
import com.mhm.framework.aspect.mqttmsgtype.MqttmsgtypeAspect;
import com.mhm.framework.common.Constants;
import com.mhm.mng.CommandMng;
import com.mhm.mqttlistener.executor.ExecutorBase;
import com.mhm.mqttlistener.executor.MqttExecutorService;
import com.mhm.utils.LedConstants;@Component("myMqttClose")
@MqttmsgtypeAspect(value = EnumMqttMsgtype.CLOSE)
public class ExecutorServiceImpl_Close extends ExecutorBase implements MqttExecutorService {@Autowiredprivate CommandMng commandMng;@Overridepublic void execute(MqttCmdDto dto) {System.out.println("close...");// 需判断该topic是否对该资源有open操作String codes = generateCodes(dto);OperatorDto operatorDto = new OperatorDto();operatorDto.setOperatorType(Constants.INT_VALUE2);operatorDto.setClient(LedConstants.LOG_CLIENT_MQTT);operatorDto.setNodeLevel(LedConstants.CONTROLLER_NODE_TYPE); // setNodeLevel必须在SetControllerType之前operatorDto.setControllerType(Constants.INT_VALUE0);operatorDto.setParam1(codes);operatorDto.setParam2("0");Userinfo user = new Userinfo(1);BusinessRtnMsg brmsg = commandMng.cmd_api_v2_switchLight(user, operatorDto);}}