京东到家开放平台消息系统-进阶之路

京东到家开放平台，是一个面向商家以及第三方开发者-开放服务、持续赋能O2O业务的到家服务集成平台。是商家数据与到家数据打通的桥梁，开放平台、商家、到家三者之间的关系如图所示。

既然是数据通讯的桥梁，必然涉及到双向通信，开放平台提供大量的可调用的API接口，同时要求商家提供必要的接口供到家调用，比如新订单的产生，订单状态变更，新增门店信息，促销信息审核等等。

其中API接口由于是到家提供，相对来讲比较稳定，而通知接口是商家提供，大量不确定因素都需要考虑，比如接口挂掉了，服务器宕机，网络中断等等，出了问题后还面临如何快速发现，重新推送，推送频次，重试时长，消息间会不会相互影响等问题。这些在消息系统设计之初都应该考虑到。

消息系统核心机制的设计

开放平台的消息系统（Business Message Queue，简称BMQ）是基于内部系统的MQ，这些MQ被内部各业务系统订阅的同时，也被开放平台订阅，平台进行格式转化（标准消息或非标准消息）后，发送给商家，基本概念如图所示。

MQ系统是京东自主研发的一款消息中间件系统，具有高可用、数据高可靠等特性。所以可以直接使用MQ来保证平台自动重新推送消息，推送频次，重试时长，以及消息的不丢失。那直接利用有什么弊端呢？

以新订单消息为例，当一个商家的接口挂了，就会不断重试，甚至积压，一旦积压就会影响到没有问题的商家。那怎么解决这个问题？能不能针对商家维度进行消息推送？

开放平台的做法是，将重点ka与普通商家分离，所有ka商家独立通道，ka商家是可扩展的，当有必要时可以从非ka商家中提取单独通道进行处理，这样就将业务维度转换成了商家维度，具体做法如图所示。

这样简单做可以了吗？其实大家会发现一个问题，虽然ka商家间的影响小了，但是非ka商家仍然存在相互影响的问题，随着业务增长，非ka商家间相互影响会越来越大，ka商家通道的不同业务之间也会互相影响吗？

以沃尔玛为例，一个新订单消息接口挂了，会不会影响售后消息呢？这是非常有可能的，同时也有一个非常简单的做法就是按业务再次分离，因为消息的种类非常多，每个商家的每个消息都单独开辟消息通道，显然要申请大量的topic，不太利于维护和管理。

开放平台的做法是，每个通道增加专门的重试通道，一旦消息有问题，直接丢到重试通道即可，这样就不影响正常的业务了，具体做法如图所示。

这样就解决了大部分问题，在实际的运行过程中也发现是可行的。以为可以高枕无忧了吗？直到一次出现了常规通道中大量积压，可是重试通道却没有数据，原来，我们httpclient允许的超时时间是3秒，超过3秒被认为是失败的，会丢到重试通道中，但实际上商家出现了大量慢响应，可能都在2秒多才返回，这样引发不了发送失败的条件，造成常规通道中的积压。

开放平台针对这种情况的做法是，每个通道增加专门的降级通道，我们实时监控统计（kafka+flink）单商家单消息响应时间超过阈值的次数，对于规定时间内单商家单消息统计次数超过阈值的打上降级标。开放平台在消费到家内部业务MQ时，有降级标的商家消息会直接进入对应的降级通道，不会进入常规通道。实时监控统计也会对恢复正常的单商家单消息清除降级标记，降级通道中执行失败的消息处理方式同常规通道一样，都会进入重试通道。具体策略如图所示。

以上基本上解决了大部分实际的场景，并且还具有可扩展空间，比如发现非ka的量较大时，可以将消息较多的商家提出单独通道即可，总结就是采用三种通道结合具体策略达到了应对实际情况的方案，归纳如图所示。

到家开放平台是如何保证消息顺序问题？因为平台具备自动重试能力，所以没有严格确保消息顺序，只是使用了MQ自身保证的顺序，平台对调用商家接口定义为仅消息通知，需要商家接到消息通知后反查到家接口，要求商家提供的消息接口具备排重能力。

消息系统的预警机制

核心的设计虽然解决了实际环境中的各种问题，但这主要解决了平台的风险，仍然面临的问题有：

1、接口出现了问题怎们能第一时间通知给商家，毕竟要求所有商家建立及时有效的报警也不太现实。

2、非ka商家共用通道还是有一些相互影响的风险。

3、知道了这些风险后，需要哪些工具来定位和处理问题呢。

这时一套及时有效便捷的报警机制显得尤为重要。毕竟早知道，早通知，早处理是解决系统风险的第一要务，越早预警越可以规避不必要的风险。

报警是建立在统计之上进行的，开放平台的统计流程如图。

任何商家任何消息出现连续失败达到一定阈值，就会发出报警，报警内容会以短信方式发送给商家负责人与开放平台研发，如下所示。

未有商家达到失败阈值，但是多位商家失败叠加导致JMQ报警，无法定位商家与接口时，我们有工具可以进行定位：按总失败的量定位风险最大的商家。商家统计节点可以点进入看商家下每个消息接口总失败次数。

知道某个商家某个接口导致的失败报警，经过与商家确认后，不能短时间修复，或者不再订阅的，可以快速禁用此消息（消息将会被丢弃）。

商家得知系统问题后，修复过程中没必要持续报警，还可以手动停止报警（消息仍然保持重试，只停止报警，避免短信费用浪费）。

动态消息发布的产生

从上面核心功能的概念图中可以看出，消息系统是需要订阅大量底层MQ消息的，首先咱们先看一下以前京东到家开放平台BMQ系统中是如何发布一个消息接口的。

同时也可以看下代码维护的情况，BMQ系统中有很多相似的java类，不利于代码的观赏与维护。

总结弊端如下：

1、接入新消息流程繁琐复杂

2、需要耗费最少1名研发的资源和时间

3、每次新消息接入都是重复工作

4、每次新消息接入都需系统上线

为了解决这些问题，开放平台引入动态配置的思路，开放平台的标准消息占大部分，标准消息是和非标准消息相对的一个概念，标准消息就是无论原有消息的内容是什么，最终转化成标准格式，推送内容的字段统一，业务含义标准化。可配置的内容包括简单的业务处理选择，字段映射关系，默认值处理，还有简单的逻辑。可配置的界面如图所示。

字段可配置细节如下图。

消息管理除了基本的增删改功能，还需要通知服务端动态订阅mq（包括停止），这样连listener都不用写了，如果一旦发现有线上问题，还可以做到及时回退版本达到回滚的目的，管理界面如下。

服务端实现动态订阅mq的核心代码如下。

// ips字段为空为广播所有机器处理，不为空为指定某些机器执行
if (StringUtils.isEmpty(dynamicLoadingMessage.getIps()) || dynamicLoadingMessage.getIps().contains(IPUtil.getIp())) {
  // type=0需要加载的mq，非0需要卸载的mq
  if (dynamicLoadingMessage.getType() == 0) {
    openPlatformBMQEngine.getMessageConsumer().unsubscribe(dynamicLoadingMessage.getTopic());
    openPlatformBMQEngine.getMessageConsumer().subscribe(dynamicLoadingMessage.getTopic(), commonBMQListener);
    LOGGER.error("DynamicLoadingListener->onMessage->subscribe->dynamicLoadingMessage:{}", JsonUtils.toJson(dynamicLoadingMessage));
  } else {
    openPlatformBMQEngine.getMessageConsumer().unsubscribe(dynamicLoadingMessage.getTopic());
    LOGGER.error("DynamicLoadingListener->onMessage->unsubscribe->dynamicLoadingMessage:{}", JsonUtils.toJson(dynamicLoadingMessage));
  }
}

通过以上处理后，发布消息的流程如下。

总结下动态bmq的特点：

1、消息系统平台化

取消原有定制化接入流程，使用规则配置动态解析处理，实现平台化统一入口接入新消息。

2、接入低风险

规避原有定制化手工编写代码上线接入方式，基于规则配置完成消息动态化，动态发布消息从而有效避免开发上线过程中可能导致的风险，从而把风险降到最低。

3、提高接入效率

原有接入流程需要耗费约2个工作日的开发、测试和上线成本，新流程基于动态配置规避原有的开发、测试、上线流程，接入新消息时间缩短至1分钟，后期甚至可达到使非开发人员有能力自行接入。

总结

本篇着重介绍了BMQ系统是如何保证稳定性，易用性及简洁性，从实践的角度结合理论解决了诸多问题，不仅满足了当前系统的实际需要，也可以应对未来一段时间的业务增长需求。从理论的角度上看，平台还有很多不足，在未来的日子里我们也会持续结合实践进行优化。