大报文问题实战

Tech

导读

大报文问题，在京东物流内较少出现，但每次出现往往是大事故，甚至导致上下游多个系统故障。大报文的背后，是不同商家业务体量不同，特别是B端业务的采购及销售出库单，一些头部商家对京东系统支持业务复杂度及容量能力的要求越来越高。因此有必要把这个问题重视起来，从组织上根本上解决。

大报文问题，是指不同的系统通过网络进行数据交互时payload size过大导致的系统可用性下降问题。

图1.大报文示意

对于大报文的产生方，过大的报文在序列化时消耗更多内存和CPU，在传输时(JSF/MQ)可能超过中间件的大小限制导致传输失败；对于大报文的消费方，过大的报文在反序列化时会产生大对象，消耗更多的内存和CPU，容易触发FullGC甚至OOM，而在处理过程中要遍历的内容更多，造成响应变慢，如果涉及数据库操作容易产生大事务、慢SQL，这些容易触发超时，如果客户端有重试机制，会进一步加重大报文消费方负载，严重时导致服务集群整体不可用。

此外，由于大报文与小报文是在一个接口上完成的，使用相同的UMP key，它会导致监控失真，报警阈值无效。如果日志记录了原始报文，也可能磁盘打满和响应变慢。

在京东物流技术体系内，具体表现为：

📌 JMQ/JSF对payload大小的限制都属于防御性保护措施，目前的值是科学的，它们都已经足够大了。在紧急止血情况下可以调整配置参数来暂时提高payload大小限制，但长期看它会加重系统的风险，应该从设计入手避免超过payload大小限制。

1.1.1 JMQ限制

根据JMQ的官方文档，单条消息大小：JMQ4不要超过4M，JMQ2不要超过2M。

class ClusterManager {    protected volatile int maxSize = 4194304; // 4MB}
class MessageProducer implement Producer { // Producer接口的具体实现类    ClusterManager clusterManager;
    // producer.send时做校验    int checkMessages(List<Message> messages) {        int size = 0;        for (Message message : messages) {            size += message.getSize() // 压缩后的大小        }        if (size > this.clusterManager.getMaxSize()) {            throw new IllegalArgumentException("the total bytes of message body must be less than " + this.clusterManager.getMaxSize());        }    }}

图2.JMQ消息大小的限制，以代码实现为准

1.1.2 JSF限制

根据JSF官方文档，JSF可以在server和consumer端分别设置payload size，默认都是8MB。

图3.数据包大小限制和调用压缩

class LengthFieldBasedFrameDecoder { // 基于netty io.netty.handler.codec.LengthFieldBasedFrameDecoder的改动    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {        // 从JSF协议的报文头里获取本次请求的payload size，此时还没有读取8MB的body        long frameLength = getUnadjustedFrameLength(in, actualLengthFieldOffset, lengthFieldLength, byteOrder);        if (frameLength > maxFrameLength) { // maxFrameLength即8MB限制            throw new TooLongFrameException();        }    }}
class ServerChannelHandler implements ChannelHandler {    public void exceptionCaught(ChannelHandlerContext ctx, final Throwable cause) {        if (cause instanceof IOException) {            // ...        } else if (cause instanceof RpcException) {            // 这里可以看到遇到这种异常，JSF是如何给consumer端响应的            ResponseMessage responseMessage = new ResponseMessage(); // 给consumer的响应            responseMessage.getMsgHeader().setMsgType(Constants.RESPONSE_MSG);            String causeMsg = cause.getMessage();            String channelInfo = BaseServerHandler.getKey(ctx.channel());            String causeMsg2 = "Remote Error Channel:" + channelInfo + " cause: " + causeMsg;            ((RpcException) cause).setErrorMsg(causeMsg2);            responseMessage.setException(cause); // 异常传递给consumer            // socket.write回consumer            ChannelFuture channelFuture = ctx.writeAndFlush(responseMessage);        } else {            // TooLongFrameException会走到这里，它的继承关系如下：            // TooLongFrameException -> DecoderException -> CodecException -> RuntimeException            // 异常被吃掉了，不给consumer响应            logger.warn("catch " + cause.getClass().getName() + " at {} : {}",                    NetUtils.channelToString(channel.remoteAddress(), channel.localAddress()),                    cause.getMessage());        }    }}

📌 经与JSF团队确认，consumer端或provider端发出的消息过大(超过playload)时consumer端得不到正确的异常响应只提示请求超时的问题，已经在1.7.5版本修复：需要provider端升级。升级后，如果consumer端发送的消息过大，provider会立即响应RpcException。

图4.已修复consumer端或provider端发出的消息过大(超过playload)时consumer端得不到正确的异常响应只提示请求超时的问题

图5.consumer使用了默认的5秒超时，但consumer抛出超时异常总用时是48秒

class JSFClientTransport {    // consumer同步调用provider    ResponseMessage send(BaseMessage msg, int timeout) {        MsgFuture<ResponseMessage> future = doSendAsyn(msg, timeout);        return future.get(timeout, TimeUnit.MILLISECONDS);    }
    MsgFuture doSendAsyn(final BaseMessage msg, int timeout) {        final MsgFuture resultFuture = new MsgFuture(getChannel(), msg.getMsgHeader(), timeout);        Protocol protocol = ProtocolFactory.getProtocol(msg.getProtocolType(), msg.getMsgHeader().getCodecType());        byteBuf = protocol.encode(request, byteBuf); // 发送报文前的序列化        RequestMessage request = (RequestMessage) msg;        request.setMsg(byteBuf);        channel.writeAndFlush(request, channel.voidPromise()); // socket.write，异步IO        resultFuture.setSentTime(JSFContext.systemClock.now());    }}
class MsgFuture implements java.util.concurrent.Future {    final long genTime = JSFContext.systemClock.now(); // new的时候就赋值了    volatile long sentTime;
    // 抛出超时异常逻辑    ClientTimeoutException clientTimeoutException() {        Date now = new Date();        String errorMsg = "[JSF-22110]Waiting provider return response timeout . Start time: " + DateUtils.dateToMillisStr(new Date(genTime))                + ", End time: " + DateUtils.dateToMillisStr(now)                + ", Client elapsed: " + (sentTime - genTime) // 它包括：序列化时间，由于异步IO因此不包括socket.write时间                + "ms, Server elapsed: " + (now.getTime() - sentTime);        return new ClientTimeoutException(errorMsg);    }}

1.1.3 物流网关限制

物流网关在nginx层通过client_max_body_size（http://nginx.org/en/docs/http/ngx_http_core_module.html#client_max_body_size）做了5MB限制。这意味着，JSF限制了8MB，但通过物流网关对外开放成HTTP JSON API时，调用者实际的限制是5MB。

1.1.4 MySQL限制

max_allowed_packet，net_buffer_length等参数在底层控制TCP层的报文长度，京东物流体系内该值足够大，研发不必关注。

研发需要关注的是字段长度的定义，主要是varchar的长度。MySQL通过sql_mode（https://dev.mysql.com/doc/refman/5.7/en/sql-mode.html）参数控制字段超过长度后的行为是字段截断还是中断事务。对于京东物流业务执行链路比较长的场景来讲，同一个字段可能多处保存，例如订单行里的skuName，就会在OFC/WMS等系统保存，sku_name varchar长度的不一致，特殊场景下可能造成上下游交互出现问题。

1.1.5 其他限制

DUCC value 的长度默认限制为 4W 字符。

UMP Key的限制128。

JMQ的businessId长度限制100，Producer在发送是默认超时2秒，Producer发送失败默认重试2次。

JMQ消费者抛出异常会导致重试(进入retry-db)，首次重试10分钟，如果重试还不成功会越来越慢推送直至过期。过期时间：JMQ2为3天，JMQ4为30天。

JSF如果不配置consumer timeout，则使用默认值：5秒。

Zookeeper ZNode限制长度 1MB。虽然可以通过jute.maxbuffer这个Java系统属性修改，但强烈不建议。

原则上，所有依赖的中间件都要确认其限制约束，提升健壮性，避免边界条件被触发而产生出乎意料的错误。

1.2.1 集合类字段无约束

导致京东物流线上事故的大报文问题中，绝大部分都属于该类问题。而这又可以细分为两种场景：

interface JsfAPI {    // 场景1：批量接口，对批量的大小无限制    void foo(List<Request> requests);    }
class Request {    // 场景2：对一个类内部的集合类字段大小无限制    // JMQ产生大报文，绝大部分属于该场景    List<Item> items;}

当数据量增大时，报文也会增大，造成几MB到几十MB的报文传输，系统为了处理这样大数据量的报文，必然会产生大对象，并且这种对象会一直处于内存中，在数据保存处理时，会造成内存不能释放，可能触发频繁FullGC，CPU使用率飙升。同时，处理集合数据，往往会有数据遍历过程，如果无并发则时间复杂度是O(N)，大的数据集必然带来更慢的响应速度，而consumer端不会根据payload大小动态设置超时时间，它可能导致consumer端超时，超时可能带来多次重试，进而加重服务端压力。

1.2.2 大字段无约束

Amazon FBA的SP-API(Sell Partner API)，对可能出现风险的字段都做了长度限制，例如：

String displayableOrderComment; // maxLength: 1000String sellerSku; // maxLength: 50String giftMessage; // maxLength: 512String displayableComment; // maxLength: 250

图6.对可能出现风险的字段做了长度限制

1.2.3 查询接口返回大量数据

ECLP主数据有个接口：导出所有warehouse list，调用方很多，访问频率不高，每次响应长度3MB。该接口在线上出现过多次事故(2019年)。这个接口显然是不该存在的，但把它下线需要推动所有的调用方改动，这个周期很长阻力也很大。

最开始，直接查数据库，出现事故后加入JimDB，再次出现事故后配置了JimDB的local cache，后又加入JSF限流等措施。

出现故障时，ECLP CPU飙升，导致服务超时，京东零售调用方配置的超时设置很短，这导致越来越多的请求打过来，加重了ECLP负担。

1.2.4 导出问题

这个问题与【1.2.3 查询接口返回大量数据】看上去类似，但有很大不同：一个同步调用，返回的数据量相对少，另一个异步执行，返回数据量巨大。

WMS6.0的报表都有导出的需求，例如导出最近3个月的明细数据。贴近商家的OFC(如ECLP)，也有类似需求，商家要求导出明细数据。系统执行过程大致是：根据用户指定的条件异步执行SQL，把数据库返回的数据集写入Excel，并存放到blob storage(指定TTL)，用户在规定时间(TTL)内根据storage key去blob storage下载，完成整个导出过程。

图7.如何查询数据库的8个可选项

condition = DateRange(from = "2022-01-01 00:00:00", to = "2022-04-01 00:00:00") // 用户指定的时间范围：3个月// sql = select * from ob_shipment_order where xxx and update_time between condition.from and condition.toList<DateRange> chunks = explode(condition)for (DateRange chunk : chunks) {    // 该chunk的时间范围已经变成了1天，甚至是1小时，具体值是根据SQL执行计划估算得来的：数据量越大则拆分越细    sql = select * from ob_shipment_order where xxx and update_time between chunk.from and chunk.to    mysql.query(sql)}

1.2.5 payload约束不一致产生的问题

例如，aws的API Gateway与Lambda对payload size有不同的约束，最终用户必须知道限制最严格的那一个环节。

图8.aws的API Gateway与Lambda对payload size有不同的约束

对于京东物流，JSF与JMQ的限制不同，理论上可能产生这样的问题：JSF调用者发送8MB的请求，JSF提供者处理时采用同步转异步机制，异步把该请求8MB发送MQ，它会导致MQ发送永远无法成功，而JSF的调用方却浑然不觉。

如果通过物流网关对外开放，网关nginx限制是5MB，而JSF是8MB，设计上没问题（fail fast），但可能造成服务方承诺与调用者感知端到端的不一致。

JSF对provider(jsf:server)和consumer可以分别设置不同的报文大小限制，理论上也可能出现问题，但在京东物流尚未出现，可不必关注。

1.2.6 其他非入口场景

public class ListResultInterceptor implements org.apache.ibatis.plugin.Interceptor {    private static final int RESULTSET_SIZE_THRESHOLD = 10000;
    @Override    public Object intercept(Invocation invocation) throws Throwable {        Object result = invocation.proceed();        if (result != null && result instanceof List) {            int resultSetSize = ((List) result).size();            if (resultSetSize > RESULTSET_SIZE_THRESHOLD) {                // 报警            }        }
        return result;    }}

设计原则

2.2  Fail Fast

2.3  上下游对齐隐式约束

2.4  大报文产生方负责拆分

具体办法

3.1  报文分页

class Payload {    List<Item> items;    int currentPage, totalPages;}
void sendPayload(Payload payload) {    int currentPage = 1;    int totalPages = payload.getItems().size() / batchSize;    Lists.partition(payload.getItems, batchSize).forEach(subItems -> {        Payload subPayload = new Payload(subItems)        subPayload.setPageInfo(currentPage, totalPages)        producer.send(subPayload)        currentPage++;    });}

3.2  报文转存

class BigPayload {    List<Item> items;}
class SmallPayload {    List<String> itemBlobKeys;}
void sendPayload(BigPayload bigPayload) {    SmallPayload smallPayload = new SmallPayload();    Lists.partition(bigPayload.getItems(), batchSize).forEach(subItems -> {        List<String> itemBlobKeys = blogStore.putObjects(subItems)        smallPayload.addItemBlobKeys(itemBlobKeys);    });
    producer.send(JSON.encode(smallPayload);}

3.3  报文截断

3.4  分页调用

class FooRequest {    @javax.validation.constraints.Size(min = 1, max = 200)    private List<Bar> barItems;}
interface JsfAPI {    // 场景1：批量接口    void foo(@javax.validation.constraints.Size(min = 1, max = 200) List<FooRequest> requests)
    // 场景2：请求对象里有集合字段    void bar(FooRequest request);}

int maxJsfRetries = 3; // JSF async下的自动重试只能应用层自己做了int retried = 0;do {    List<ResponseFuture<Result<ObLocatingResultDto>>> futures = new LinkedList();    Lists.partition(voList, batchSize).forEach(subVoList -> {        ObLocatingOrderDto dto = mapper.INSTANCE.toDTO(subVoList);        locatingAppService.outboundOrderLocate(dto); // async JSF call        ResponseFuture<Result<ObLocatingResultDto>> future = RpcContext.getContext().getFuture();        futures.add(future);    });
    for (ResponseFuture<Result<ObLocatingResultDto>> future : futures) {        try {            Result<ObLocatingResultDto> result = future.get();        } catch (RpcException jsfException) {            retried++;        } catch (Throwable e) {            // 额外的业务逻辑：与JSF并发同步调用相同的处理逻辑        }    }} while (retried <= maxJsfRetries);

3.5  MQ代替JSF

3.6  总结

4.1  单个接口与批量接口分离

interface JsfAPI {    Result<SkuInfo> getSkuInfo(String sku);    Result<List<SkuInfo>> listSkuInfo(List<String> skus);}

Result<SkuInfo> result = jsfAPI.listSkuInfo(Lists.newArrayList("EMG1800752592"));

interface JsfAPI {    Result<SkuInfo> getSkuInfo(String sku);}
interface JsfBulkAPI {    Result<List<SkuInfo>> listSkuInfo(List<String> skus);}

4.2  线程池隔离

4.3  大报文与小报文分离

4.4  JMQ设置合理的批量大小

interface MessageListener {    void onMessage(List<Message> messages) throws Exception;}

interface Producer {    void send(List<Message> messages) throws JMQException;}

/** * 发送分播结果消息 */public void send(List<CheckResultDto> checkResultDtos) {    List<Message> messageList = Lists.newArrayList();    for (CheckResultDto checkResultDto : checkResultDtos) {        String messageText = JmqMessage.createReportBody(checkResultDto.getUuid(), Lists.newArrayList(checkResultDto));        messageList.add(JmqMessage.create(topic, messageText, checkResultDto.getUuid(), checkResultDto.getWarehouseNo()));    }    producer.send(messageList);}

4.5  避免大日志

if (logger.isInfoEnabled()) {    log.info(JsonUtil.toJson(request); // CPU intensive and disk I/O intensive(虽然日志是顺序写)}

4.6  显式约束由严开始

5.1  识别大报文场景

@Slf4jpublic final class PayloadSizeFilter extends AbstractFilter {    private static final int PAYLOAD_SIZE_THRESHOLD = 4 << 20; // 4MB = 8MB(JSF限制) * 50%    private static final int BATCH_SIZE_THRESHOLD = 1000;
    @Override    public ResponseMessage invoke(RequestMessage requestMessage) {        if (!RpcContext.getContext().isProviderSide()) {            // 只在provider端检查大报文：它才是我们要保护的对象            return getNext().invoke(requestMessage);        }
        // 自动识别潜在的大报文场景：针对报文大小        Integer payloadSize = requestMessage.getMsgHeader().getLength();        if (payloadSize != null && payloadSize > PAYLOAD_SIZE_THRESHOLD) {            // 这里使用最简单的日志把潜在大报文暴露出来，各团队可以做更细化的机制            // 由于logbook限制只有error level日志才能配置"关键字报警"，这里使用log.error            // 如果不想自动报警，只是人工巡检，可以log.warn            String methodName = requestMessage.getMethodName();            String className = requestMessage.getClassName();            log.error("Suspected BIG payload: {}.{}, {}>{}", className, methodName, payloadSize, PAYLOAD_SIZE_THRESHOLD);        }
        // 自动识别潜在的大报文场景：报文字节小，但仍会导致处理慢，例如 List<String> orderNos，如果发来1万个单号？        // 这里只能识别出入参是List的场景，对于字段类型是List的场景无效        Invocation invocation = requestMessage.getInvocationBody();        Class[] argClasses = invocation.getArgClasses();        Object[] args = invocation.getArgs();        for (int i = 0; i < argClasses.length; i++) {            Class argClass = argClasses[i];            if (Collection.class.isAssignableFrom(argClass)) {                // 入参类型是Collection                Collection collection = (Collection) args[i];                if (collection.size() > BATCH_SIZE_THRESHOLD) {                    log.error("Too BIG Collection argument: {}>{}", collection.size(), BATCH_SIZE_THRESHOLD);                }            }        }
        return getNext().invoke(requestMessage);    }}

public interface BigPayloadTrait extends MessageListener {    int THRESHOLD_BIG_PAYLOAD = 2 << 20; // 2MB = 4MB(JMQ限制) * 50%
    default boolean suspectedBigPayload(List<Message> messages) {        for (Message message : messages) {            if (message.getSize() > THRESHOLD_BIG_PAYLOAD) {                return true;            }        }
        return false;    }}

5.2  有效的监控

5.3  设计应急预案

5.4  常态化的大报文捣乱演练

5.5  团队执行

▪功能支撑：会员成长体系、等级计算策略、权益体系、营销底层能力支持

▪用户活跃：会员关怀、用户触达、活跃活动、业务线交叉获客、拉新促活