Spring 事件监听器示例及源码解析

浅浅的花香味﹌ 2022-12-08 05:44 162阅读 0赞

# Spring 事件监听器示例及源码解析 #

--------------------

## 源码解析 ##

--------------------

### ApplicationEvent ###

public abstract class ApplicationEvent extends EventObject { 
        //序列化
        private static final long serialVersionUID = 7099057708183571937L;
        //时间戳
        private final long timestamp = System.currentTimeMillis();
        //source事件推送源
        public ApplicationEvent(Object source) { 
            super(source);
        }
    
        public final long getTimestamp() { 
            return this.timestamp;
        }
    }

### ApplicationContext - publishEvent() ###

protected void publishEvent(Object event, @Nullable ResolvableType eventType) { 
            //事件非null
            Assert.notNull(event, "Event must not be null");
            Object applicationEvent;
            //是否事件继承自ApplicationEvent
            if (event instanceof ApplicationEvent) { 
                applicationEvent = (ApplicationEvent)event;
            } else { 
                applicationEvent = new PayloadApplicationEvent(this, event);
                if (eventType == null) { 
                    eventType = ((PayloadApplicationEvent)applicationEvent).getResolvableType();
                }
            }
    		
    	    //早期事件是否不为空
            if (this.earlyApplicationEvents != null) { 
                this.earlyApplicationEvents.add(applicationEvent);
            } else { 
                //事件保存
                this.getApplicationEventMulticaster().multicastEvent((ApplicationEvent)applicationEvent, eventType);
            }
    
            if (this.parent != null) { 
                if (this.parent instanceof AbstractApplicationContext) { 
                    ((AbstractApplicationContext)this.parent).publishEvent(event, eventType);
                } else { 
                    this.parent.publishEvent(event);
                }
            }
    
        }

### SimpleApplicationEventMulticaster ###

public void multicastEvent(ApplicationEvent event, @Nullable ResolvableType eventType) { 
            ResolvableType type = eventType != null ? eventType : this.resolveDefaultEventType(event);
            //获取线程池
            Executor executor = this.getTaskExecutor();
            //获取指定事件监听器的迭代器
            Iterator var5 = this.getApplicationListeners(event, type).iterator();
            
            while(var5.hasNext()) { 
                ApplicationListener<?> listener = (ApplicationListener)var5.next();
                //线程池不为null
                if (executor != null) { 
                    //线程异步执行
                    executor.execute(() -> { 
                        this.invokeListener(listener, event);
                    });
                } else { 
                    //同步执行
                    this.invokeListener(listener, event);
                }
            }
    
        }
    
    protected void invokeListener(ApplicationListener<?> listener, ApplicationEvent event) { 
            //异常处理器
            ErrorHandler errorHandler = this.getErrorHandler();
            if (errorHandler != null) { 
                try { 
                    //事件处理
                    this.doInvokeListener(listener, event);
                } catch (Throwable var5) { 
                    errorHandler.handleError(var5);
                }
            } else { 
                //事件处理
                this.doInvokeListener(listener, event);
            }
    
        }
    
     private void doInvokeListener(ApplicationListener listener, ApplicationEvent event) { 
            try { 
                //调用事件监听器处理事件
                listener.onApplicationEvent(event);
            } catch (ClassCastException var6) { 
                String msg = var6.getMessage();
                if (msg != null && !this.matchesClassCastMessage(msg, event.getClass())) { 
                    throw var6;
                }
    
                Log logger = LogFactory.getLog(this.getClass());
                if (logger.isTraceEnabled()) { 
                    logger.trace("Non-matching event type for listener: " + listener, var6);
                }
            }
    
        }

## 代码 ##

--------------------

### RedisPubSubConfig：Redis 订阅发布配置 ###

@Configuration
    public class RedisPubSubConfig { 
        public final static String tokenChannel = "ztx";
        @Bean
        RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory) { 
            RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
            redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
            redisMessageListenerContainer.addMessageListener(pubSubRefreshListener(), new ChannelTopic(tokenChannel));
            return redisMessageListenerContainer;
        }
    
        @Autowired
        EventPubSubService eventPubSubService;
        MessageListener pubSubRefreshListener() { 
            //事件监听器
            return new MessageListener() { 
                @SneakyThrows
                @Override
                public void onMessage(Message message, byte[] pattern) { 
                    //将message中的数据进行反序列化
                    try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(message.getBody()))) { 
                        //将事件强转为父接口类型
                        EventBase event = (EventBase) in.readObject();
                        //getEvent()获取的类型是子类型接口 此处泛型
                        eventPubSubService.subEvent(event.getEvent());
                    }
                }
            };
        }

### EventPubSubService：消息发布与事件发布 ###

@Service
    public class EventPubSubService { 
        @Autowired
        RedisTemplate redisTemplate;
        @Autowired
        ApplicationContext applicationContext;
        //消息推送 将消息推送到redis
        public void pubEvent(ApplicationEvent event) { 
            redisTemplate.convertAndSend(RedisPubSubConfig.tokenChannel, event);
        }
    	
        //将redis发布的事件进行发布
        public <T> void subEvent(T event) { 
            applicationContext.publishEvent(event);
        }
    
    }

### EventBase：事件接口 ###

//泛型 用于统一发布事件 不同监听器各自监听
    public interface EventBase<T> { 
        T getEvent();
    }

### ZTXEvent：事件类 ###

@Getter
    public class ZTXEvent extends ApplicationEvent implements EventBase<ZTXEvent>, Serializable { 
        private String carId;
        private Integer type;
        private byte param;
    
    
        public ZTXEvent() { 
            super("ztx");
        }
    
        public ZTXEvent(String carId, Integer type, byte param) { 
            super("ztx");
            this.carId = carId;
            this.type = type;
            this.param = param;
        }
    
    
        @Override
        public ZTXEvent getEvent() { 
            return this;
        }
    }

### ZTXEventListener：事件监听器 ###

@Service
    public class ZTXEventListener implements ApplicationListener<ZTXEvent> { 
        @Autowired
        EquipmentServiceImpl equipmentService;
        
        //处理事件
        @Override
        public void onApplicationEvent(ZTXEvent ztxEvent) { 
            ChannelHandlerContext ctx = equipmentService.carCtx.get(ztxEvent.getCarId()).getCtx();
            ProtocolOfZTX protocolOfZTX = new ProtocolOfZTX();
            protocolOfZTX.setAgvId(Byte.parseByte(ztxEvent.getCarId()));
            switch (ztxEvent.getType()){ 
                case 1:
                    setProtocolOfZTXData(protocolOfZTX,new byte[]{ 5,2});  //前
                    break;
                case 2:
                    setProtocolOfZTXData(protocolOfZTX,new byte[]{ 5,3});  //后
                    break;
                case 3:
                    setProtocolOfZTXData(protocolOfZTX,new byte[]{ 7,2});  //左
                    break;
                case 4:
                    setProtocolOfZTXData(protocolOfZTX,new byte[]{ 7,3});  //右
                    break;
                case 5:
                    setProtocolOfZTXData(protocolOfZTX,new byte[]{ 7,3});  //右
                    break;
                case 6:
                    setProtocolOfZTXData(protocolOfZTX,new byte[]{ 6,0});  //停止
                    break;
                case 7:
                    setProtocolOfZTXData(protocolOfZTX,new byte[]{ 8,ztxEvent.getParam()});  //速度
                    break;
                case 8:
                    setProtocolOfZTXData(protocolOfZTX,new byte[]{ 0x0B,1,0,0,0});  //急停
                    break;
                case 9:
                    setProtocolOfZTXData(protocolOfZTX,new byte[]{ 0x0B,0,1,0,0});  //准备
                    break;
                case 10:
                    setProtocolOfZTXData(protocolOfZTX,new byte[]{ 0x0B,0,0,1,0});  //故障清除
                    break;
                case 11:
                    setProtocolOfZTXData(protocolOfZTX,new byte[]{ 0x0B,0,0,0,1});  //驱动提升
                    break;
                default:
                    break;
            }
            ctx.writeAndFlush(protocolOfZTX);
        }
    
        private void setProtocolOfZTXData(ProtocolOfZTX protocolOfZTXData,byte[] bytes){ 
            byte[] data = new byte[20];
            System.arraycopy(bytes, 0, data, 0, bytes.length);
            protocolOfZTXData.setData(data);
        }
    }

--------------------

## 流程 ##

### 类分布 ###

*  A系统包含类
    
     *  ZTXEvent
     *  RedisPubSubConfig
     *  EventBase
 *  B系统包含类
    
     *  EventBase
     *  ZTXEvent
     *  ZTXEventListener

### 工作流程 ###

*  A系统
    
     *  界面按钮点击
     *  redis推送时间ZTXEvent（自定义事件）
 *  B系统
    
     *  接受到Redis发布的时间
     *  发布事件ZTXEvent
     *  ZTXEventListener对ZTXEvent进行事件处理

## 设计原理 ##

A系统需要和B系统进行通信，A系统推送事件后，B系统收到事件并将该事件发布，B系统实现了该事件的监听器，C系统没有实现该事件的监听器，所以B该事件只对B系统有效。

在这种情况下，所有需要相互交互的系统，都只需要通过redis发布事件即可，若某个系统对该事件感兴趣则只需要实现该事件的监听器即可。