EventConstants類， 主要是用來定義事件相關常量

/**
 * 事件相關常量
 * @author victor
 *
 */
public class EventConstants {
	
	public static final String KEY_EVENT_AUTO_ID = "KEY_EVENT_AUTO_ID"; // event 計數器， 用於生成EVENT_ID

	public static final String KEY_EVENT_DATA_MAP = "KEY_EVENT_DATA_MAP"; //用來儲存使用者自定義data的 Hash
	
	public static final String EVENT_KEY = "KEY_EVENT"; // 用來做event redis唯一標識字首
	
	public static final String EVENT_DATA_KEY = "DATA";// hash  中儲存data字首
	
	public static final String DEFAULT_EMPTY_DATA = "KEY_EVENT_DATA_EMPTY"; // 空data時常量替換串
	
	public static final String EVENT_META_KEY = "META";//用來儲存 事件型別字首
}
 

/**
 * 事件型別列舉
 * @author victor
 *
 */
public enum EventEnum implements IndexedEnum<EventEnum>{
	
	EXAMPLE(1,"example","示例事件");
	
	EventEnum(int index,String code,String name){
		this.index = index;
		this.name = name;
		this.code = code;
	}

	private int index;
	
	private String name;
	
	private String code;
	
	
	private static final ImmutableMap<Integer, EventEnum> INDEXS = IndexedEnumUtil.toIndexes(values());
	
	public static EventEnum indexOf(int index){
		return INDEXS.get(index);
	}


	public int getIndex() {
		return index;
	}

	public String getName() {
		return name;
	}

	public String getCode() {
		return code;
	}
	
}
 

訊息註冊Service以及介面：負責註冊事件，移出事件等實現：

/**
 * 事件服務介面
 * @author victor
 *
 */
public interface IEventService {

	/**
	 * 註冊事件,相同事件不會覆蓋，註冊成功將返回全域性唯一事件ID
	 * @param event  事件
	 * @param data	業務資料
	 * @return 事件ID,全域性唯一
	 */
	public String register(EventEnum event,String data);
	
	/**
	 * 註冊事件,相同事件不會覆蓋，註冊成功將返回全域性唯一事件ID
	 * @param event  事件
	 * @param data	業務資料
	 * @param seconds 事件執行時間, 單位秒
	 * @return 事件ID,全域性唯一
	 */
	public String register(EventEnum event,String data,int seconds);
	
	/**
	 * 註冊事件,相同事件不會覆蓋，註冊成功將返回全域性唯一事件ID
	 * @param event  事件
	 * @param data	業務資料
	 * @param seconds 事件執行時間
	 * @return 事件ID,全域性唯一
	 */
	public String register(EventEnum event,String data,Date date);
	
	
	/**
	 * 移除事件,如果事件已經被執行,移除無意義,如果事件未執行,則返回移除成功或者失敗
	 * @param eventId  事件ID
	 * @return 是否移除成功,true 表示成功，false表示失敗(如事件正在被執行)
	 */
	public boolean remove(String eventId);
	
}
 

實現類：

@Service
public class EventServiceImpl implements IEventService{
	
	private static Logger logger = Logger.getLogger(EventServiceImpl.class);
	
	@Autowired
	private IRedisService redisService;
	
	@Autowired
	private IDistributedLock distributedLock;

	@Override
	public String register(EventEnum event, String data) {
		return register(event, data,1);
	}

	@Override
	public String register(EventEnum event, String data, int seconds) {
		seconds = seconds <= 0 ? 1 : seconds;
		data = StringUtils.isNullOrEmpty(data) ? EventConstants.DEFAULT_EMPTY_DATA : data;
		String eventId = String.valueOf(redisService.incyby(EventConstants.KEY_EVENT_AUTO_ID));
		Map<String,String> hash = new HashMap<>();
		hash.put(EventUtils.getDataKey(eventId), data);
		hash.put(EventUtils.getMetaKey(eventId), event.getIndex() + "");
		String res = redisService.hmset(EventConstants.KEY_EVENT_DATA_MAP,hash);
		if(StringUtils.isNullOrEmpty(data) || StringUtils.isNullOrEmpty(hash) || StringUtils.isNullOrEmpty(res)){
			logger.error("Set " + EventConstants.KEY_EVENT_DATA_MAP + " error . eventId:" + eventId + ",data: " + data + ",meta:" + event.getIndex());
		}
		redisService.setex(EventUtils.getEventKey(eventId), event.getIndex() + "", seconds);
		logger.info("register event :" + eventId);
		/*ThreadPoolUtil.executor(ThreadPoolEnum.KEY_EVENT_REGISTER).execute(new KeyEventRegisterRunable(redisService,eventId,event,data,seconds));*/
		return eventId;
	}

	@Override
	public String register(EventEnum event, String data, Date date) {
		int seconds = DateUtils.calculateDifferenceSeconds(date, new Date());
		return register(event, data,seconds);
	}

	@Override
	public boolean remove(String eventId) {
		if(distributedLock.lock(eventId, 5)){
			redisService.del(EventUtils.getEventKey(eventId));
			redisService.hdel(EventConstants.KEY_EVENT_DATA_MAP, EventUtils.getDataKey(eventId),EventUtils.getMetaKey(eventId));
			distributedLock.unlock(eventId);
		}
		logger.info("Remove key event error : event is executed!");
		return false;
	}

}

移出事件： 直接刪除事件, 防止刪除事件同時事件正在執行，加鎖

訂閱監聽類：

/**
 * Key event 頻道訂閱<br/>
 * 需要redis版本2.8以上，並且修改conf配置檔案引數: <cite>notify-keyspace-events Ex</cite><br/>
 * 開啟配置之後redis將會在key失效時，將訊息傳送到<cite>[email protected]<db>__:expired</cite>頻道，其中db代表key所在資料庫的index
 * @author victor
 *
 */
@Component
@Configuration
public class KeyEventMessageListener implements MessageListener{
	
	private static Logger logger = Logger.getLogger(KeyEventMessageListener.class);
	
	private final RedisSerializer<String> serializer = new StringRedisSerializer();
	
	@Autowired
	private EventHandlerServiceFacotry eventHandlerServiceFacotry;
	
	@Autowired
	private IRedisService redisService;
	
	@Autowired
	private IRedisLock redisLock;
	
	@Override
	public void onMessage(Message message, byte[] pattern) {
		String key = serializer.deserialize(message.getBody());
		String eventId = EventUtils.getEventId(key);
		if(!StringUtils.isNullOrEmpty(eventId)){
			if(redisLock.lock(eventId, 300)){//must executed in 300S
				logger.info("Handle event :" + eventId + " start !");
				try{
					handler(eventId);
				}catch (Exception ex) {
					logger.error("Key event hadppend bug : eventId:" + eventId + ", message :" + ex.getMessage() , ex);
					ex.printStackTrace();
				}finally{
					//delete data and meta
					redisService.del(EventUtils.getEventKey(eventId));
					redisService.hdel(EventConstants.KEY_EVENT_DATA_MAP, EventUtils.getDataKey(eventId),EventUtils.getMetaKey(eventId));
					redisLock.unlock(eventId);
				}
				logger.info("Handle event :" + eventId + " end!");
			}else{// event was handled or is handling
				logger.debug("Event is handled");
			}
		} 
	}
	
	private void handler(String eventId){
		String data = redisService.hget(EventConstants.KEY_EVENT_DATA_MAP,EventUtils.getDataKey(eventId));
		String meta = redisService.hget(EventConstants.KEY_EVENT_DATA_MAP,EventUtils.getMetaKey(eventId));
		if(StringUtils.isNullOrEmpty(data) || StringUtils.isNullOrEmpty(meta)){// event was handled
			logger.info("Event was handled, eventId:" + eventId + ".");
			return;
		}
		int event = Integer.parseInt(meta);
		data = data.equals(EventConstants.DEFAULT_EMPTY_DATA) ? null : data;
		IEventHandlerService service = eventHandlerServiceFacotry.instance(event);
		if(service == null){
			logger.error("Event handler not found, eventId:" + eventId + ",data:" + data + ",meta:" + meta + "!");
			throw new EventNotMappingHandlerException("Event handler service not mapping!",eventId,event,data);
		}
		try{
			service.handle(EventEnum.indexOf(event), eventId, data);
		}catch (Exception e) {
			logger.error("Handler event error, eventId : " + eventId, e);
			throw new EventHandleException("Handler event error.",eventId,event,data);
		}
	}
}

此類主要完成功能， 接受redis推送的訂閱訊息， 並且根據key來找到對應的 事件型別， 事件data,  加鎖防止重複消費， 通過事件執行工廠根據事件型別獲取不同的事件最終業務service.

正常情況， 不同的事件可能各自都有自己不同的事件， 比如訂單取消事件 和 商品自動上架兩種事件執行的業務邏輯就不同， 所以此處使用多型動態獲取不同的事件業務實現例項。

/**
 * 事件處理工廠
 * @author victor
 *
 */
@Service
public class EventHandlerServiceFacotry {
	
	private List<IEventHandlerService> services;
	
	@Autowired
	public EventHandlerServiceFacotry(List<IEventHandlerService> services){
		this.services = services;
	}
	
	public IEventHandlerService instance(int event){
		for(IEventHandlerService service : services){
			if(service.getEvent().getIndex() == event){
				return service;
			}
		}
		return null;
	}
	
	public IEventHandlerService instance(EventEnum event){
		return instance(event.getIndex());
	}
}

IEventHandlerService介面， 定義事件執行介面方法以及獲取時間型別介面：

/**
 * 事件觸發服務 自定義
 * @author victor
 *
 */
public interface IEventHandlerService {

	/**
	 * 執行
	 * @param eventId
	 * @param event
	 * @param data
	 */
	public void handle(EventEnum event,String eventId,String data);
	
	/**
	 * 事件型別
	 * @return
	 */
	public EventEnum getEvent();
}

@Service
public class ExampleEventHandlerServiceImpl implements IEventHandlerService{
	
	@Override
	public void handle(EventEnum event, String eventId, String data) {
		System.out.println("執行事件:" + event.getName() + ",使用者自定義data:" + data);
	}

	@Override
	public EventEnum getEvent() {
		return EventEnum.EXAMPLE;
	}

}

輔助類EventUtils：

public final class EventUtils {

	private EventUtils(){
		
	}
	
	public static String getEventKey(String eventId){
		return EventConstants.EVENT_KEY + "_" + eventId;
	}
	
	public static String getDataKey(String eventId){
		return eventId + "_" + EventConstants.EVENT_DATA_KEY;
	}
	
	public static String getMetaKey(String eventId){
		return eventId + "_" + EventConstants.EVENT_META_KEY;
	}
	
	public static boolean isEvent(String key){
		if(!StringUtils.isNullOrEmpty(key) && key.toUpperCase().indexOf(EventConstants.EVENT_KEY + "_") == 0){
			return true;
		}
		return false;
	}
	
	
	public static String getEventId(String key){
		if(isEvent(key)){
			String eventId = key.replace(EventConstants.EVENT_KEY + "_", "");
			return eventId;
		}
		return null;
	}
	
}

上述程式碼中， 部分類可能沒有， 比如redisService ： 為redis操作命令封裝   distributionLock ：分散式鎖，筆者部落格前面有自己實現的redis鎖。

呼叫註冊事件：註冊一個型別為EXAMPLE型別事件，自定義data為 hello,example字串， 並且在60秒之後執行：

@Autowired
private  IEventService eventService;

eventService.register(EventEnum.EXAMPLE,"hello , example !" ,60);