(精华)2020年10月7日高并发高可用 Redis实现异步架构

前言

在后端编程时，对需要立即返回的数据我们应当立刻返回，而对于可以慢慢处理而业务复杂的我们可以选择延迟返回。这个实现使用到了异步消息队列。

异步消息队列

主要用于实现生产者-消费者模式。也就是说，这个队列应当是可以阻塞的，否者会带来大量的性能浪费。

生产者-消费者模式

实现
1.定义事件类型 – 定义Enum类 – EnumType
用于表示该事件的类型

public enum EventType {//这里列举了四种类型LIKE(0),COMMENT(1),LOGIN(2),MAIL(3);private int value;EventType(int value) {this.value = value;}public int getValue() {return value;}
}

2.定义事件的实体 – EventModel
这里说明一下entityOwnerId的必要性。举个例子，当我们给一个人点赞时，系统要给那个人（也就是entityOwnerId）发送一个站内信，通知那个人他被点赞了。当然，我们也可以把entityOwnerId包装在exts里，但因为几乎每一个事件都需要这个字段，所以这里我们开一个字段给他。

public class EventModel {//之前定义的事件类型private EventType type;//触发者的idprivate int actorId;//entityId和entityType共同组成了所触发的事件private int entityId;private int entityType;//该事件的拥有者private int entityOwnerId;//需要传输的额外信息private Map<String, String> exts = new HashMap<>();public Map<String, String> getExts() {return exts;}public EventModel() {}public EventModel(EventType type) {this.type = type;}public String getExt(String name) {return exts.get(name);}public EventModel setExt(String name, String value) {exts.put(name, value);return this;}public EventType getType() {return type;}public EventModel setType(EventType type) {this.type = type;return this;}public int getActorId() {return actorId;}public EventModel setActorId(int actorId) {this.actorId = actorId;return this;}public int getEntityId() {return entityId;}public EventModel setEntityId(int entityId) {this.entityId = entityId;return this;}public int getEntityType() {return entityType;}public EventModel setEntityType(int entityType) {this.entityType = entityType;return this;}public int getEntityOwnerId() {return entityOwnerId;}public EventModel setEntityOwnerId(int entityOwnerId) {this.entityOwnerId = entityOwnerId;return this;}
}

3.生产者的实现 – EventProducer
这里的队列我们使用Redis的阻塞双向队列list来实现。
a) 我们先用JSON把事件序列化
b) 再通过lpush把事件推进队列里
EventPr oducer

@Service
public class EventProducer {@AutowiredJedisAdapter jedisAdapter;public boolean fireEvent(EventModel eventModel) {try {String json = JSONObject.toJSONString(eventModel);String key = RedisKeyUtil.getEventQueueKey();jedisAdapter.lpush(key, json);return true;} catch (Exception e) {return false;}}
}

RedisKeyUtil – 用于统一的管理Redis的Key

public class RedisKeyUtil {private static String BIZ_EVENT = "EVENT";public static String getEventQueueKey() {return BIZ_EVENT;}
}

JedisAdapter – 对Jedis的函数进行一层封装

@Service
public class JedisAdapter implements InitializingBean {private static final Logger logger = LoggerFactory.getLogger(JedisAdapter.class);private Jedis jedis = null;private JedisPool pool = null;@Overridepublic void afterPropertiesSet() throws Exception {pool = new JedisPool("localhost", 6379);}private Jedis getJedis() {return pool.getResource();}public long lpush(String key, String value) {Jedis jedis = null;try {jedis = getJedis();return jedis.lpush(key, value);} catch (Exception e) {logger.error("发生异常" + e.getMessage());return 0;} finally {if (jedis != null) {jedis.close();}}}public List<String> brpop(int timeout, String key) {Jedis jedis = null;try {jedis = pool.getResource();return jedis.brpop(timeout, key);} catch (Exception e) {logger.error("发生异常" + e.getMessage());return null;} finally {if (jedis != null) {jedis.close();}}}
}

4.定义一个事件处理器的接口 – EventHandler

public interface EventHandler {//事件处理函数void doHandle(EventModel model);//获取该事件处理器所支持的事件类型List<EventType> getSupportEventTypes();
}

5.消费者的实现 – EventConsumer
a）创建一个类型为Map<EventType, List>的map，用于存放所有的Handler。
b）在afterPropertiesSet函数中（这个函数在sping在初始化完该Bean后会执行），我们通过applicationContext获取实现了EventHandler接口的全部Handler。
b.1）通过for循环，分门别类的把各个Handler放到map中。
b.2）启动线程去消化事件
b.2.1）该线程使用死循环让其不间断的运行。
b.2.2）用brpop把事件拉出来
b.2.3）过滤掉key之后，剩下value，把value用JSON的api转化为EventModel
b.2.4）在map中寻找是否有能处理EventModel的Handler，判断方法是看EventType是否支持。
b.2.5）过滤掉不支持的EventType之后，调用每一个支持该EventType的doHandle方法。
具体代码实现

@Service
public class EventConsumer implements InitializingBean, ApplicationContextAware {private static final Logger logger = LoggerFactory.getLogger(EventConsumer.class);private Map<EventType, List<EventHandler>> config = new HashMap<>();private ApplicationContext applicationContext;@Autowiredprivate JedisAdapter jedisAdapter;@Overridepublic void afterPropertiesSet() throws Exception {Map<String, EventHandler> beans = applicationContext.getBeansOfType(EventHandler.class);if (beans != null) {for (Map.Entry<String, EventHandler> entry : beans.entrySet()) {List<EventType> eventTypes = entry.getValue().getSupportEventTypes();for (EventType type : eventTypes) {if (!config.containsKey(type)) {config.put(type, new ArrayList<EventHandler>());}// 注册每个事件的处理函数config.get(type).add(entry.getValue());}}}// 启动线程去消费事件Thread thread = new Thread(new Runnable() {@Overridepublic void run() {// 从队列一直消费while (true) {String key = RedisKeyUtil.getEventQueueKey();List<String> messages = jedisAdapter.brpop(0, key);// 第一个元素是队列名字for (String message : messages) {if (message.equals(key)) {continue;}EventModel eventModel = JSON.parseObject(message, EventModel.class);// 找到这个事件的处理handler列表if (!config.containsKey(eventModel.getType())) {logger.error("不能识别的事件");continue;}for (EventHandler handler : config.get(eventModel.getType())) {handler.doHandle(eventModel);}}}}});thread.start();}@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}
}

6.写一个实现了EventHandler接口的实现

@Component
public class LikeHandler implements EventHandler {@AutowiredMessageService messageService;@AutowiredUserService userService;@Overridepublic void doHandle(EventModel model) {Message message = new Message();User user = userService.getUser(model.getActorId());message.setToId(model.getEntityOwnerId());message.setContent("用户" + user.getName() +" 赞了你的资讯,http://127.0.0.1:8080/news/"+ String.valueOf(model.getEntityId()));// SYSTEM ACCOUNTmessage.setFromId(3);message.setCreatedDate(new Date());messageService.addMessage(message);}@Overridepublic List<EventType> getSupportEventTypes() {return Arrays.asList(EventType.LIKE);}
}

7.在Controller中调用Producer的fireEvent – 用于产生一个事件

@Controller
public class LikeController {@AutowiredLikeService likeService;@AutowiredHostHolder hostHolder;@AutowiredNewsService newsService;@AutowiredEventProducer eventProducer;@RequestMapping(path = {"/like"}, method = {RequestMethod.GET, RequestMethod.POST})@ResponseBodypublic String like(@Param("newId") int newsId) {long likeCount = likeService.like(hostHolder.getUser().getId(), EntityType.ENTITY_NEWS, newsId);// 更新喜欢数News news = newsService.getById(newsId);newsService.updateLikeCount(newsId, (int) likeCount);eventProducer.fireEvent(new EventModel(EventType.LIKE).setEntityOwnerId(news.getUserId()).setActorId(hostHolder.getUser().getId()).setEntityId(newsId));return ToutiaoUtil.getJSONString(0, String.valueOf(likeCount));}
}