实际项目中的消息中心

前一篇文章讲到我们项目的工作流，这一篇我们扒一扒项目中的消息中心，msgcenter。消息可以分成很多种消息：留存可重复查看的DB消息，短暂保存在redis的comet消息，短信形式的msg消息，推送到手机的push消息等等。但是不管是哪种类型的消息，在我们的项目中都有统一的格式。（不管是db，msg或者是push都是一样的格式），这样就能保证存储和显示已经推送的一致性。

项目中的消息中心也是独立的模块，用maven依赖被各个网关层使用，当然简单方便，不过后来看了一篇文章，讲到微服务，我就在想是不是可以把消息模块独立出来做一个微服务，这样对消息中心做的优化就不用再修改别的网关应用了。不过这个是架构师哥哥处理的问题，我就是这么简单的想想。

废话不说，我们看msgcenter的层次

还是统一的层次，因为是模型，没有集成web层。dal是数据库持久层，domain是模型层（最接近模型的层），facade层是对外提供能力的层（外观模式，定义了对外提供能力的接口），biz是实现facade层的实例层，integration是集成其他模块的层（这一层很有意思，到时候我们单独写一篇文章介绍），test是测试层

那既然我们的消息中心是一个模型，供其他系统调用，那我们就看下facade对外提供的能力吧。

看facade层只有两个接口类（我们只关注MsgSenderFacade接口），

public interface MsgSenderFacade {
   /**
   * 发送消息
   *
   * @param list
   * @return
   */
   List<MsgBodyDTO> sendMsgs(List<MsgBodyDTO> list);

我们简单的看下发送消息的方法，系统中所有的消息调用的都是这个方法，是不是很屌，原因上面我们讲到过，不管什么类型的消息，他们都有统一的格式，那就是MsgBodyDTO,里面就是我们平常类的ID，消息模板ID，模板的替换数组，接收用户的ID，接收用户的电话号码，业务号，标题，真实内容，创建时间，更新时间，状态等等属性，我们看下实现方法

     fillIdAndStatus(allMsgList);//填充ID，创建时间，初始状态等等
       Map<MsgTypeEnum, List<MsgBodyDTO>> map = splitByMsgType(allMsgList);//根据消息队列的不同消息模板，分开（之后分别处理）
       try {
           for (Map.Entry<MsgTypeEnum, List<MsgBodyDTO>> entry : map
                   .entrySet()) {
               MsgTypeEnum msgType = entry.getKey();
               List<MsgBodyDTO> msgList = entry.getValue();
               switch (msgType) {
               case COMET:
                   LogUtil.info(logger,
                           "start send comet msgs[" + msgList.size() + "]");
                   if (logger.isDebugEnabled()) {
                       LogUtil.debug(logger, JSON.toJSONString(msgList));
                   }
                   cometMsgSenderProcessor.checkParams(msgList);
                   cometMsgSenderProcessor.doSend(msgList);//不同的消息类型我们用不同的消息处理器进行验证和分发。
                   break;
               case DB:
                   LogUtil.info(logger, "start send db msgs[" + msgList.size()
                           + "]");
                   dbMsgSenderProcessor.checkParams(msgList);
                   dbMsgSenderProcessor.doSend(msgList);
                   break;
               case MSG:
                   LogUtil.info(logger,
                           "start send mobile msgs[" + msgList.size() + "]");
                   mobileMsgSenderProcessor.checkParams(msgList);
                   mobileMsgSenderProcessor.doSend(msgList);
                   break;
               case PUSH:
                   LogUtil.info(logger,
                           "start send push msgs[" + msgList.size() + "]");
                   pushMsgSenderProcessor.checkParams(msgList);
                   pushMsgSenderProcessor.doSend(msgList);
                   break;
               default:
                   throw new BizException(
                           ErrorConst.ERROR_SUCH_MESSAGE_NOT_SUPPORT,
                           "不支持当前的消息类型[" + msgType.name() + "]");
               }
           }
           return allMsgList;

这四个处理器都是Spring注入的，我们看下配置，

我们接下来跟进代码查看具体实现方法以mobileMsgSenderProcessor为例：

        //检查消息不能重复
        checkMsgsValid(domainMsgs);
        for (MsgBody msgBody : domainMsgs) {
            if (realSendMobileMsg) {
                MsgChannel channel = channelMap.get(msgBody.getChannelId());
                if (channel == null) {
                    StringBuilder buffer = new StringBuilder();
                    buffer.append("没有找到channelId=[").append(msgBody.getChannelId()).append("]的定义");
                    throw new CriticalSystemError(buffer.toString());
                } else {
                    String mobile = msgBody.getMobile();
                    try {
                        boolean success = channel.doSend(msgBody);
                        if (success) {
                            LogUtil.info(logger, "send mobile msg success[" + msgBody.getMobile()
                                                 + "][" + msgBody.getRealContent() + "]");
                            kvClient.save(
                                "msg_ok_" + mobile,
                                "["
                                        + DateUtil.format(new Date(),
                                            DateFormatterEnum.DATE_WITH_TIME.getCode()) + "]"
                                        + msgBody.getRealContent(), Integer.MAX_VALUE);
                        } else {
                            LogUtil.error(
                                Logs.ERROR_LOGGER,
                                "send mobile msg fail[" + msgBody.getMobile() + "]["
                                        + msgBody.getRealContent() + "]");
                        }
                    } catch (Exception e) {
                        LogUtil.error(
                            Logs.ERROR_LOGGER,
                            "send mobile msg fail[" + msgBody.getMobile() + "]["
                                    + msgBody.getRealContent() + "]", e);
                    } finally {
                        kvClient.save("msg_idx_" + mobile, msgBody.getChannelId(), 86400);
                    }
                }