图示

1. 现状

场景: 假设设备上报不同类型的消息，我们要对不同类型的消息做不同的处理。如果我们通过if..else的方式处理的话会显得比较冗余。

例如：

if("alarmEvent".equals(msg)){
    // 处理告警消息逻辑 ...
}else if("deviceBase".equals(msg)){
    // 处理设备上报的基本信息 ...
}else if("heartBeat".equals(msg)){
    // 处理设备心跳消息 ...
}else {
    // ...
}

2. 消息处理Handler

那么对于不同消息的不同的处理逻辑我们可以单独放在一个实现类中，这些类有着相同的行为，所以我们可以定义一个接口：

public interface MessageHandler<T>{
    T invoke(String bizUpMessage); // 处理接收到的消息
    String getName();
    String getIdentification(); // 获取消息标志
}

针对于不同的消息，我们可以有各自的实现

实现类1：

@Component
public class DeviceBaseInfoHandler implements MessageHandler<String> {
    @Override
    public String invoke(String bizUpMessage) {
        // 处理设备上报的基本信息 ...
        return "";
    }
    @Override
    public String getName() {
        return "";
    }
    @Override
    public String getIdentification() {
        return "DeviceBaseInfoRequest";
    }
}

实现类2：

@Component
public class AlarmEventHandler implements MessageHandler<String>{
    @Override
    public String invoke(String bizUpMessage) {
       // 处理告警信息 ...
        return null;
    }
    @Override
    public String getName() {
        return "";
    }
    @Override
    public String getIdentification() {
        return "AlarmEventRequest";
    }
}

3. Handler注册处

我们可以将上一步的消息处理对象，根据消息的标志(Identification)不同，将消息放在一个Map中:

public class MessageHandlerRegisty {
    private Map<String, List<MessageHandler>> registerMap = new HashMap<>();
    // 注册各种 handler
    public void regist(String identification, MessageHandler handler){
        if(this.contains(identification)){
            get(identification).add(handler);
        } else {
            List<MessageHandler> list = new ArrayList<>();
            list.add(handler);
            registerMap.put(identification, list);
        }
    }
    public boolean contains(String identification){
        return registerMap.containsKey(identification);
    }
    public List<MessageHandler> get(String identification){
        return registerMap.get(identification);
    }
}

4. 消息接收Service

消息接收接口

public interface IMessageService {
    void messageReceived(String bizUpMessage); // 接收消息
}

5. 消息接收ServiceImpl

消息接收实现类

public class MessageServiceImpl implements IMessageService{
    
    public MessageHandlerRegisty registy;
    
    // 线程池
    public ExecutorService executorService = new ThreadPoolExecutor(3, 10, 15, TimeUnit.MINUTES, new LinkedBlockingQueue(1000), new LogdDiscardPolicy());
    
    public void messageReceived(String bizUpMessage){
        
        // 获取消息的标志Identification
        String identification = extractIdentification(bizUpMessage);
        
        if(StringUtils.isBlank(identification)){ 
            return;
        }
        
        if(!registy.contains(identification)){  
            return;
        }
        
        for(MessageHandler handler : registy.get(identification)){
            execute(handler, bizUpMessage); // 多线程处理消息
        }
    }
    
    public MessageHandlerRegisty getRegisty() {
        return registy;
    }
    public void setRegisty(MessageHandlerRegisty registy) {
        this.registy = registy;
    }
    
    private void execute(MessageHandler handler, String bizUpMessage){
        executorService.submit(new IotMessageTask(handler, bizUpMessage));
    }
    
    public final class LogdDiscardPolicy extends ThreadPoolExecutor.DiscardPolicy {
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            log.warn("MessageService pool is full !!!!!!!!");
        }
    }
    
    
    final class IotMessageTask implements Runnable {
        private MessageHandler handler;
        private String bizUpMessage;
        public IotMessageTask(MessageHandler handler, String bizUpMessage){
            this.handler = handler;
            this.bizUpMessage = bizUpMessage;
        }
        @Override
        public void run() {
            try {
                Stopwatch stopwatch = Stopwatch.createStarted();
                handler.invoke(bizUpMessage);
                stopwatch = stopwatch.stop();
                
                long cost = stopwatch.elapsed(TimeUnit.MICROSECONDS);
                logTime(cost);
            }catch (Exception e){
                log.error("handler execute error - ", e);
            }
        }
        private void logTime(long cost){
            if(cost > 50){
                log.warn(" handler -> {}, cost too much -> {}", handler.getName(), cost);
            } else {
                log.info("handler -> {}, cost -> {}", handler.getName(), cost);
            }
        }
        
    }
    
    
}

6. 将MessageServiceImpl对象配置到spring容器

@Configuration
public class ManagerConfig {
    @Bean
    public IMessageService messageService(){
        MessageServiceImpl messageService = new MessageServiceImpl();
        messageService.setRegisty(buildRegisty());
        return messageService;
    }
    // 消息处理handler
    @Autowired
    private DeviceBaseInfoHandler deviceBaseInfoHandler;
    @Autowired
    private AlarmEventHandler alarmEventHandler;
    
    private MessageHandlerRegisty buildRegisty(){
        MessageHandlerRegisty registy = new MessageHandlerRegisty();
        registy.regist(deviceBaseInfoHandler.getIdentification(), deviceBaseInfoHandler);
        registy.regist(alarmEventHandler.getIdentification(),alarmEventHandler);
        
        return registy;
    }
}