100字范文 > 阿里程序员工作小技巧：理解CPU分支预测提高代码效率

阿里程序员工作小技巧：理解CPU分支预测提高代码效率

时间：2023-05-02 06:10:05

技术传播的价值，不仅仅体现在通过商业化产品和开源项目来缩短我们构建应用的路径，加速业务的上线速率，体现也会在优秀程序员在工作效率提升，产品性能优化和用户体验改善等小技巧方面的分享，以提高我们的工作能力。

本文来自阿里巴巴中间件技术团队的程序员断岭，他是阿里微服务开源项目Dubbo的项目组成员，也是Java线上诊断开源项目Arthas的负责人。

一、基础概念

a. Dubbo: 是一款高性能、轻量级的开源Java RPC框架，提供了三大核心能力：面向接口的远程方法调用，智能容错和负载均衡，以及服务自动注册和发现；

b. ChannelEventRunnable： Dubbo 里所有网络事件的回调接口；

c. JMH：即Java Microbenchmark Harness，是专门用于代码微基准测试的工具套件。在性能优化的过程中，可以使用JMH对优化的结果进行量化的分析。

二、需求缘起：

在Stack Overflow上有一个非常著名的问题：为什么处理有序数组要比非有序数组快？从问题的结论来看，是分支预测对代码运行效率的提升起到了非常重要的作用。

现今的CPU是都支持分支预测(branch prediction)和指令流水线(instruction pipeline)，这俩的结合可以极大的提高CPU的工作效率，从而提高代码执行效率。但这仅适用于简单的if跳转，但对于Switch跳转，CPU则没有太好的解决办法，因为Switch本质上是据索引，是从地址数组里取地址再跳转。

三、思考和方案假设

要提高代码执行效率，一个重要的实现原则就是尽量避免CPU把流水线清空，从Stack Overflow上的讨论结果来看，通过提高分支预测的成功率，是可以降低CPU对流水线清空的概率。那么，除了在硬件层面，是否可以考虑代码层面帮CPU把判断提前，来提高代码执行效率呢？

四、方案验证

在Dubbo的ChannelEventRunnable里有一个Switch来判断channel state。当一个channel建立起来之后，超过99.9%的情况，它的state都是ChannelState.RECEIVED，我们可以考虑，把这个判断提前。

以下通过JMH来验证，把判断提前后是否就可以提高代码执行效率。

public class TestBenchMarks {\tpublic enum ChannelState {\t\tCONNECTED, DISCONNECTED, SENT, RECEIVED, CAUGHT\t}\t@State(Scope.Benchmark)\tpublic static class ExecutionPlan {\t\t@Param({ \u0026quot;1000000\u0026quot; })\t\tpublic int size;\t\tpublic ChannelState[] states = null;\t\t@Setup\t\tpublic void setUp() {\t\t\tChannelState[] values = ChannelState.values();\t\t\tstates = new ChannelState[size];\t\t\tRandom random = new Random(new Date().getTime());\t\t\tfor (int i = 0; i \u0026lt; size; i++) {\t\t\t\tint nextInt = random.nextInt(1000000);\t\t\t\tif (nextInt \u0026gt; 100) {\t\t\t\t\tstates[i] = ChannelState.RECEIVED;\t\t\t\t} else {\t\t\t\t\tstates[i] = values[nextInt % values.length];\t\t\t\t}\t\t\t}\t\t}\t}\t@Fork(value = 5)\t@Benchmark\t@BenchmarkMode(Mode.Throughput)\tpublic void benchSiwtch(ExecutionPlan plan, Blackhole bh) {\t\tint result = 0;\t\tfor (int i = 0; i \u0026lt; plan.size; ++i) {\t\t\tswitch (plan.states[i]) {\t\t\tcase CONNECTED:\t\t\t\tresult += ChannelState.CONNECTED.ordinal();\t\t\t\tbreak;\t\t\tcase DISCONNECTED:\t\t\t\tresult += ChannelState.DISCONNECTED.ordinal();\t\t\t\tbreak;\t\t\tcase SENT:\t\t\t\tresult += ChannelState.SENT.ordinal();\t\t\t\tbreak;\t\t\tcase RECEIVED:\t\t\t\tresult += ChannelState.RECEIVED.ordinal();\t\t\t\tbreak;\t\t\tcase CAUGHT:\t\t\t\tresult += ChannelState.CAUGHT.ordinal();\t\t\t\tbreak;\t\t\t}\t\t}\t\tbh.consume(result);\t}\t@Fork(value = 5)\t@Benchmark\t@BenchmarkMode(Mode.Throughput)\tpublic void benchIfAndSwitch(ExecutionPlan plan, Blackhole bh) {\t\tint result = 0;\t\tfor (int i = 0; i \u0026lt; plan.size; ++i) {\t\t\tChannelState state = plan.states[i];\t\t\tif (state == ChannelState.RECEIVED) {\t\t\t\tresult += ChannelState.RECEIVED.ordinal();\t\t\t} else {\t\t\t\tswitch (state) {\t\t\t\tcase CONNECTED:\t\t\t\t\tresult += ChannelState.CONNECTED.ordinal();\t\t\t\t\tbreak;\t\t\t\tcase SENT:\t\t\t\t\tresult += ChannelState.SENT.ordinal();\t\t\t\t\tbreak;\t\t\t\tcase DISCONNECTED:\t\t\t\t\tresult += ChannelState.DISCONNECTED.ordinal();\t\t\t\t\tbreak;\t\t\t\tcase CAUGHT:\t\t\t\t\tresult += ChannelState.CAUGHT.ordinal();\t\t\t\t\tbreak;\t\t\t\t}\t\t\t}\t\t}\t\tbh.consume(result);\t}}

验证说明：

benchSiwtch里是纯Switch判断；benchIfAndSwitch里用一个如果提前判断状态是否ChannelState.RECEIVED。

基准测试结果是：

Result \u0026quot;io.github.hengyunabc.jmh.TestBenchMarks.benchSiwtch\u0026quot;: 576.745 ±(99.9%) 6.806 ops/s [Average] (min, avg, max) = (490.348, 576.745, 618.360), stdev = 20.066 CI (99.9%): [569.939, 583.550] (assumes normal distribution)# Run complete. Total time: 00:06:48Benchmark(size) Mode CntScore Error UnitsTestBenchMarks.benchIfAndSwitch 1000000 thrpt 100 1535.867 ± 61.212 ops/sTestBenchMarks.benchSiwtch 1000000 thrpt 100 576.745 ± 6.806 ops/s

可以看到，提前if判断提高了近3倍的代码效率，这种技巧可以放在性能要求严格的地方。。