应用层

cache aside pattern

步骤

  1. 查找: 先查缓存，缓存没有查 db， db 查到重新设置到缓存，再响应请求
  2. 更新: 更新 db，删除缓存

为什么是删除缓存而不是更新缓存

  • 如果缓存中的数据是那种需要计算的数据，每次数据更新就需要重新计算一次再设置到缓存中，而该数据的访问却是冷数据，那么就造成性能浪费
  • 一种 懒加载 的思路

实时性数据的数据双写不一致

缓存系统 - 图3

非实时性数据的架构

缓存系统 - 图4

大数据

缓存系统 - 图5

storm 的入门案例

安装

  • 需要 jdk、zk、python2 环境

    配置

  • conf/storm.yaml

  • 列下主要的点
  • 主要要写域名
  • 一旦修改了配置，到 zk 中 rmr /storm 干掉原来的配置 ```yaml

    ZooKeeper服务器列表

    storm.zookeeper.servers:
    • “localhost”

      nimbus 运行位置

      nimbus.seeds: [“localhost”]

storm使用的本地文件系统目录(必须存在并且storm进程可读写)

storm.local.dir: “D:/dev_env/apache-storm-2.1.0/data”

supervisor上能够运行workers的端口列表。每个worker占用一个端口，且每个端口只运行一个worker。

通过这项配置可以调整每台机器上运行的worker数。(调整slot数/每机)

supervisor.slots.ports:

  • 6700
  • 6701
  • 6702
  • 6703

指定运行大小

to nimbus

nimbus.childopts: “-Xmx1024m”

to supervisor

supervisor.childopts: “-Xmx1024m”

to worker

worker.childopts: “-Xmx768m” 

  1. <a name="MClj7"></a>
  2. #### 运行
  3. > storm nimbus
  4. > storm supervisor
  5. > storm logviewer  // 注意是在 supervisor 上运行
  6. > storm ui
  8. 打开 `localhost:8080`
  9. <a name="XR9VB"></a>
  10. #### 依赖
  11. ```xml
  12.     <dependencies>
  13.         <!-- https://mvnrepository.com/artifact/org.apache.storm/storm-core -->
  14.         <dependency>
  15.             <groupId>org.apache.storm</groupId>
  16.             <artifactId>storm-core</artifactId>
  17.             <version>1.2.3</version>
  18.         </dependency>
  20.         <!--   解决 本地运行时 java.lang.NoClassDefFoundError: com/codahale/metrics/JmxReporter    -->
  21.         <!--        <dependency>-->
  22.         <!--            <groupId>io.dropwizard.metrics</groupId>-->
  23.         <!--            <artifactId>metrics-core</artifactId>-->
  24.         <!--            <version>3.2.4</version>-->
  25.         <!--        </dependency>-->
  26.     </dependencies>
  28.         <!-- 打包方式 -->
  29.     <build>
  30.         <plugins>
  31.             <plugin>
  32.                 <groupId>org.apache.maven.plugins</groupId>
  33.                 <artifactId>maven-shade-plugin</artifactId>
  34.                 <configuration>
  35.                     <createDependencyReducedPom>true</createDependencyReducedPom>
  36.                     <filters>
  37.                         <filter>
  38.                             <artifact>*:*</artifact>
  39.                             <excludes>
  40.                                 <exclude>META-INF/*.SF</exclude>
  41.                                 <exclude>META-INF/*.sf</exclude>
  42.                                 <exclude>META-INF/*.DSA</exclude>
  43.                                 <exclude>META-INF/*.dsa</exclude>
  44.                                 <exclude>META-INF/*.RSA</exclude>
  45.                                 <exclude>META-INF/*.rsa</exclude>
  46.                                 <exclude>META-INF/*.EC</exclude>
  47.                                 <exclude>META-INF/*.ec</exclude>
  48.                                 <exclude>META-INF/MSFTSIG.SF</exclude>
  49.                                 <exclude>META-INF/MSFTSIG.RSA</exclude>
  50.                             </excludes>
  51.                         </filter>
  52.                     </filters>
  53.                     <artifactSet>
  54.                         <excludes>
  55.                             <exclude>org.apache.storm:storm-core</exclude>
  56.                         </excludes>
  57.                     </artifactSet>
  58.                 </configuration>
  59.                 <executions>
  60.                     <execution>
  61.                         <phase>package</phase>
  62.                         <goals>
  63.                             <goal>shade</goal>
  64.                         </goals>
  65.                         <configuration>
  66.                             <transformers>
  67.                                 <transformer
  68.                                         implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
  69.                                 <transformer
  70.                                         implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
  71.                                 </transformer>
  72.                             </transformers>
  73.                         </configuration>
  74.                     </execution>
  75.                 </executions>
  76.             </plugin>
  77.         </plugins>
  78.     </build>

主要逻辑

  1. /**
  2.  * 随机句子拓扑
  3.  *
  4.  * @author lyl
  5.  */
  6. public class RandomSentenceTopology {
  7.     /**
  8.      * 生产句子的 Spout
  9.      */
  10.     public static class SentencesSpout extends BaseRichSpout {
  11.         /**
  12.          * 数据发射器
  13.          */
  14.         private SpoutOutputCollector collector;
  16.         private Random random;
  18.         /**
  19.          * open
  20.          * 对 spout 进行初始化，可以设置线程池、数据库连接池、httpClient 等用于连接数据源
  21.          *
  22.          * @param map
  23.          * @param topologyContext
  24.          * @param spoutOutputCollector 可以利用该对象发送数据出去
  25.          */
  26.         @Override
  27.         public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
  28.             this.collector = spoutOutputCollector;
  29.             this.random = new Random();
  30.         }
  32.         /**
  33.          * storm 中的 task 会一直调用该方法
  34.          */
  35.         @Override
  36.         public void nextTuple() {
  37.             try {
  38.                 TimeUnit.MILLISECONDS.sleep(200);
  39.             } catch (InterruptedException e) {
  40.                 e.printStackTrace();
  41.             }
  43.             // 随机获取一个句子
  44.             String[] sentences = new String[]{
  45.                     "hello, i am akarin",
  46.                     "maybe is silly",
  47.                     "but i am lucky"
  48.             };
  49.             String sentence = sentences[random.nextInt(sentences.length)];
  51.             System.err.println("spout 即将发送的句子: " + sentence);
  53.             // 发送一个 tuple 到下一个 task，无限的 task 组成一个概念 stream
  54.             this.collector.emit(new Values(sentence));
  55.         }
  58.         /**
  59.          * 定义 tuple 的 field 名称
  60.          *
  61.          * @param outputFieldsDeclarer
  62.          */
  63.         @Override
  64.         public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
  65.             outputFieldsDeclarer.declare(new Fields("sentence"));
  66.         }
  67.     }
  69.     public static class SplitBolt extends BaseRichBolt {
  70.         /**
  71.          * 数据发射器
  72.          */
  73.         private OutputCollector collector;
  75.         /**
  76.          * 初始化
  77.          *
  78.          * @param map
  79.          * @param topologyContext
  80.          * @param outputCollector
  81.          */
  82.         @Override
  83.         public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
  84.             this.collector = outputCollector;
  85.         }
  87.         /**
  88.          * storm 的 task 无限调用该方法
  89.          *
  90.          * @param tuple
  91.          */
  92.         @Override
  93.         public void execute(Tuple tuple) {
  94.             // 通过 tuple 的 field 获取内容
  95.             String sentence = tuple.getStringByField("sentence");
  96.             // 截取句子，貌似我这里搞错了，切错了
  97.             String[] words = sentence.split(" ");
  99.             // 发送 tuple 到下一个 task
  100.             for (String word : words) {
  101.                 this.collector.emit(new Values(word));
  102.             }
  103.         }
  105.         /**
  106.          * 也是定义 tuple 的 field名称
  107.          *
  108.          * @param outputFieldsDeclarer
  109.          */
  110.         @Override
  111.         public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
  112.             outputFieldsDeclarer.declare(new Fields("word"));
  113.         }
  114.     }
  116.     public static class CalculateBolt extends BaseRichBolt {
  117.         /**
  118.          * 数据发射器
  119.          */
  120.         private OutputCollector collector;
  122.         /**
  123.          * 记录词频
  124.          */
  125.         private Map<String, Integer> countMap = new ConcurrentHashMap<>();
  128.         /**
  129.          * 初始化
  130.          *
  131.          * @param map
  132.          * @param topologyContext
  133.          * @param outputCollector
  134.          */
  135.         @Override
  136.         public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
  137.             this.collector = outputCollector;
  138.         }
  140.         /**
  141.          * task 无限调用
  142.          *
  143.          * @param tuple
  144.          */
  145.         @Override
  146.         public void execute(Tuple tuple) {
  147.             String word = tuple.getStringByField("word");
  149.             Integer count = countMap.get(word);
  150.             if (Objects.isNull(count)) {
  151.                 count = 1;
  152.             } else {
  153.                 count++;
  154.             }
  155.             countMap.put(word, count);
  157.             System.err.println("单机统计，word: " + word + " count: " + count);
  159.             this.collector.emit(new Values(word, count));
  161.         }
  163.         /**
  164.          * 命名 tuple 的各个 field
  165.          *
  166.          * @param outputFieldsDeclarer
  167.          */
  168.         @Override
  169.         public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
  170.             outputFieldsDeclarer.declare(new Fields("word", "count"));
  171.         }
  172.     }
  174.     public static void main(String[] args) {
  175.         // 指定拓扑
  176.         TopologyBuilder builder = new TopologyBuilder();
  177.         builder.setSpout(
  178.                 // spout 的名称
  179.                 "randomSentenceSpout",
  180.                 // spout 实现类
  181.                 new RandomSentenceTopology.SentencesSpout(),
  182.                 // executor 个数
  183.                 2
  184.         );
  186.         builder.setBolt(
  187.                 // bolt 名字
  188.                 "splitSentenceBolt",
  189.                 // bolt 实现类
  190.                 new RandomSentenceTopology.SplitBolt(),
  191.                 // executor 个数
  192.                 3)
  193.                 // 默认 task 和 executor 个数保持一致，可以设置
  194.                 .setNumTasks(6)
  195.                 // 指定数据上游 spout/bolt；选择消费策略为负载均衡
  196.                 .shuffleGrouping("randomSentenceSpout");
  198.         builder.setBolt(
  199.                 "calculateCountBolt",
  200.                 new RandomSentenceTopology.CalculateBolt(),
  201.                 3)
  202.                 .setNumTasks(6)
  203.                 // 指定数据上游 spout/bolt；选择消费策略为根据 tuple field 固定数据流向同一个本 task
  204.                 // 避免在不同的 task 计算了相同的 word
  205.                 .fieldsGrouping("splitSentenceBolt", new Fields("word"));
  207.         // storm's config
  208.         Config config = new Config();
  210.         // 命令行执行
  211.         if (args != null && args.length > 0) {
  212.             // 指定 worker 数量
  213.             config.setNumWorkers(3);
  214.             try {
  215.                 StormSubmitter.submitTopology(
  216.                         // 拓扑名称
  217.                         args[0], config, builder.createTopology());
  218.             } catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
  219.                 e.printStackTrace();
  220.             }
  221.         } else {
  222.             // 本地执行
  223.             // 本地测试修改并行度，即分配总的 task 个数，让拓扑自己配置
  224.             config.setMaxTaskParallelism(20);
  226.             LocalCluster cluster = new LocalCluster();
  227.             cluster.submitTopology(
  228.                     // 拓扑名称
  229.                     "RandomSentenceTopology",
  230.                     config, builder.createTopology());
  232.             try {
  233.                 TimeUnit.SECONDS.sleep(60);
  234.             } catch (InterruptedException e) {
  235.                 e.printStackTrace();
  236.             }
  238.             cluster.shutdown();
  239.         }
  240.     }
  241. }

提交任务

  • 本地，直接运行即可
  • 提交到 storm 集群
  1. 打包，需要使用 maven-shade-plugin ，见 打包方式

  2. 提交命令, 分别是 jar 包，入口类全类名，命名(main 函数里面指定要求给拓扑一个名称，即使用 arg[0])

    storm jar testStorm-1.0-SNAPSHOT.jar storm.RandomSentenceTopology mytopology

  3. 在 storm.ui 或者 storm list 可以查看任务

    运行显示

  • 本地可以看见

    1. 单机统计，word: silly count: 22
    2. spout 即将发送的句子: maybe is silly
    3. 单机统计，word: maybe count: 22
    4. 单机统计，word: is count: 22
    5. 单机统计，word: maybe count: 23
    6. 单机统计，word: silly count: 23
    7. 单机统计，word: is count: 23

  • 提交集群可以到时候看对应 storm ui 中对应拓扑的 worker 的 port 日志

    测试项目下载

    testStorm.zip

问题大全

  1. supervisor.log 中显示 客户端没有所需的特权
    1. win下用管理员权限的 cmd 运行 supervisor，如果不行，就全部都用管理员权限的 cmd
  2. 点击 storm ui 中 nimbus的 port 查看日志，发现显示[ERROR] No available slots for topology

image.png

  1. 配置文件中设置 woker 的运行大小 worker.childopts
    1. 咋看日志啊
  2. 目录下的 /log/workers-artifacts 查看各个端口的日志
  3. 或者 storm ui 上点击对应的 拓扑，到下面的 worker 页面点击对应 port image.png

WorkerResources
Port
suprvisorld
Host
3426834tb023-192.168.1.11
18a28794-37b6-4a8~
6700

  1. 改了配置没反应啊
    1. todo
    2. 这里直接去 zk，用 rmr /storm 干掉元数据了

缓存冷启动和预热

缓存系统 - 图8

热数据

缓存系统 - 图9

预防缓存雪崩

缓存系统 - 图10