[分布式架构下的并行数据处理机制 ]的搜索结果

...Spark应用在执行分布式缓存操作时出现问题 一、问题初现 分布式缓存的初衷与现状 嘿，朋友们！今天我们来聊聊Spark在分布式缓存操作中遇到的一些坑。说到Spark，它可是大数据处理界的明星选手，性能强大，功能丰富。但即使是这么优秀的框架，有时候也会让我们头疼不已。 分布式缓存是Spark的一个重要特性，它的核心目标是减少重复计算，提升任务执行效率。简单来说，就是把一些频繁使用的数据放到内存里，供多个任务共享。听起来是不是很美好？但实际上，我在实际开发过程中遇到了不少麻烦。 比如有一次，我正在做一个数据分析项目，需要多次对同一份数据进行操作。我寻思着，这不就是常规操作嘛，直接用Spark的分布式缓存功能得了，这样岂不是能省掉好多重复加载的麻烦？嘿，事情是这样的——我辛辛苦苦搞完了任务，满怀期待地提交上去，结果发现这运行速度简直让人无语，不仅没达到预期的飞快效果，反而比啥缓存都不用的时候还慢！当时我就蒙圈了，心里直嘀咕：“卧槽，这是什么神仙操作？”没办法，只能硬着头皮一点点去查问题，最后才慢慢搞清楚了分布式缓存里到底藏着啥猫腻。 二、深入分析 为什么缓存反而变慢？ 经过一番折腾，我发现问题出在以下几个方面： 2.1 数据量太大导致内存不足 首先，大家要明白一点，Spark的分布式缓存本质上是将数据存储在集群节点的内存中。要是数据量太大，超出了单个节点能装下的内存容量，那就会把多余的数据写到磁盘上，这个过程叫“磁盘溢写”。但这样一来，任务的速度就会被拖慢，变得特别磨叽。 举个例子吧，假设你有一份1GB大小的数据集，而你的集群节点只有512MB的可用内存。你要是想把这份数据缓存起来，Spark会自己挑个序列化的方式给数据“打包”，顺便还能压一压体积。不过呢，就算是这样，还是有可能会出现溢写这种烦人的情况，挡都挡不住。唉，真是没想到啊，本来想靠着缓存省事儿提速呢，结果这操作反倒因为磁盘老是读写（频繁I/O）变得更卡了，简直跟开反向加速器似的！ 解决办法也很简单——要么增加节点的内存配置，要么减少需要缓存的数据规模。当然，这需要根据实际情况权衡利弊。 2.2 序列化方式的选择不当 另一个容易被忽视的问题是序列化方式的选择。Spark提供了多种序列化机制，包括JavaSerializer、KryoSerializer等。不同的序列化方式会影响数据的大小以及读取效率。 我曾经试过直接使用默认的JavaSerializer，结果发现性能非常差。后来改用了KryoSerializer之后，才明显感觉到速度有所提升。话说回来啊，用 KryoSerializer 的时候可别忘了先给所有要序列化的类都注册好，不然程序很可能就“翻车”报错啦！ java import org.apache.spark.serializer.KryoRegistrator; import com.esotericsoftware.kryo.Kryo; public class MyRegistrator implements KryoRegistrator { @Override public void registerClasses(Kryo kryo) { kryo.register(MyClass.class); // 注册其他需要序列化的类... } } 然后在SparkConf中设置： java SparkConf conf = new SparkConf(); conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer"); conf.set("spark.kryo.registrator", "MyRegistrator"); 2.3 缓存时机的选择失误 还有一个关键点在于缓存的时机。有些人一启动任务就赶紧给数据加上.cache()，觉得这样数据就能一直乖乖待在内存里，不用再费劲去读了。但实际上，这种做法并不总是最优解。 比如，在某些情况下，数据可能只会在特定阶段被频繁访问，而在其他阶段则很少用到。要是你提前把这部分数据缓存了，不光白白占用了宝贵的内存空间，搞不好后面真要用缓存的地方还找不到足够的空位呢！ 因此，合理规划缓存策略非常重要。比如说，在某个任务快开始了，你再随手调用一下.cache()这个方法，这样就能保证数据乖乖地待在内存里，别到时候卡壳啦！ 三、实践案例 如何正确使用分布式缓存？ 接下来，我想分享几个具体的案例，帮助大家更好地理解和运用分布式缓存。 案例1：简单的词频统计 假设我们有一个文本文件，里面包含了大量的英文单词。我们的目标是统计每个单词出现的次数。为了提高效率，我们可以先将文件内容缓存起来，然后再进行处理。 scala val textFile = sc.textFile("hdfs://path/to/input.txt") textFile.cache() val wordCounts = textFile.flatMap(_.split(" ")) .map(word => (word, 1)) .reduceByKey(_ + _) wordCounts.collect().foreach(println) 在这个例子中，.cache()方法确保了textFile RDD的内容只被加载一次，并且可以被后续的操作共享。其实嘛，要是没用缓存的话，每次你调用flatMap或者map的时候，都得重新去原始数据里翻一遍，这就跟每次出门都得把家里所有东西再检查一遍似的，纯属给自己找麻烦啊！ 案例2：多步骤处理流程 有时候，一个任务可能会涉及到多个阶段的处理，比如过滤、映射、聚合等等。在这种情况下，合理安排缓存的位置尤为重要。 python from pyspark.sql import SparkSession spark = SparkSession.builder.appName("WordCount").getOrCreate() df = spark.read.text("hdfs://path/to/input.txt") 第一步：将文本拆分为单词 words = df.selectExpr("split(value, ' ') as words").select("words.") 第二步：缓存中间结果 words.cache() 第三步：统计每个单词的出现次数 word_counts = words.groupBy("value").count() word_counts.show() 这里，我们在第一步处理完之后立即调用了.cache()方法，目的是为了保留中间结果，方便后续步骤复用。要是不这么干啊，那每走一步都得把上一步的算一遍，想想就费劲，效率肯定低得让人抓狂。 四、总结与展望 通过今天的讨论，相信大家对Spark的分布式缓存有了更深刻的认识。虽然它能带来显著的性能提升，但也并非万能药。其实啊，要想把它用得溜、用得爽，就得先搞懂它是怎么工作的，再根据具体的情况去灵活调整。不然的话，它的那些本事可就都浪费啦！ 未来，随着硬件条件的不断改善以及算法优化的持续推进，相信Spark会在更多领域展现出更加卓越的表现。嘿，咱们做开发的嘛，就得有颗永远好奇的心！就跟追剧似的，新技术一出就得赶紧瞅两眼，说不定哪天就用上了呢。别怕麻烦，多学点东西总没错，说不定哪天就能整出个大招儿来！ 最后，感谢大家耐心阅读这篇文章。如果你有任何疑问或者想法，欢迎随时交流！让我们一起努力，共同进步吧！

...。 另外，针对大规模分布式系统中的定时任务调度，Apache Eagle和Apollo等开源项目也提供了强大的解决方案。例如，Apollo拥有丰富的定时任务调度策略以及灵活的分片、依赖处理机制，能够有效应对高并发场景下的定时任务管理需求。 与此同时，云原生环境下的Kubernetes CronJob也是一个值得关注的方向。CronJob作为Kubernetes的一部分，可以根据Cron表达式在集群中调度容器化的定时任务，实现了与容器编排平台的高度集成。 此外，在深入研究定时任务原理时，可以追溯到操作系统级别的定时器和调度算法，如Linux系统的timerfd和POSIX信号定时器机制，这些底层技术为上层应用提供精确且高效的定时服务。 总之，随着技术的演进与发展，Java定时任务的实现方式日趋丰富多样，开发者应根据实际应用场景选择最适合的技术方案，同时关注社区前沿动态，以确保所采用的定时任务技术始终与时俱进。

...C10K 是单机同时处理 1 万个请求（并发连接 1 万）的问题 C1000K 是单机支持处理 100 万个请求（并发连接 100 万）的问题。 C10K C10K 问题最早由 Dan Kegel 在 1999 年提出。那时的服务器还只是 32 位系统，运行着 Linux 2.2 版本（后来又升级到了 2.4 和 2.6，而 2.6 才支持 x86_64），只配置了很少的内存（2GB）和千兆网卡。 怎么在这样的系统中支持并发 1 万的请求呢？ 从资源上来说，对 2GB 内存和千兆网卡的服务器来说，同时处理 10000 个请求，只要每个请求处理占用不到 200KB（2GB/10000）的内存和 100Kbit （1000Mbit/10000）的网络带宽就可以。 物理资源是足够的，是软件的问题，特别是网络的 I/O 模型问题。 I/O 的模型，文件 I/O和网络 I/O 模型也类似。 在 C10K 以前，Linux 中网络处理都用同步阻塞的方式，也就是每个请求都分配一个进程或者线程。 请求数只有 100 个时，这种方式自然没问题，但增加到 10000 个请求时，10000 个进程或线程的调度、上下文切换乃至它们占用的内存，都会成为瓶颈。 每个请求分配一个线程的方式不合适，为了支持 10000 个并发请求，有两个问题需要我们解决 第一，怎样在一个线程内处理多个请求，也就是要在一个线程内响应多个网络 I/O。以前的同步阻塞方式下，一个线程只能处理一个请求，到这里不再适用，是不是可以用非阻塞 I/O 或者异步 I/O 来处理多个网络请求呢？ 第二，怎么更节省资源地处理客户请求，也就是要用更少的线程来服务这些请求。是不是可以继续用原来的 100 个或者更少的线程，来服务现在的 10000 个请求呢？ I/O 模型优化 异步、非阻塞 I/O 的解决思路是我们在网络编程中经常用到的 I/O 多路复用（I/O Multiplexing） 两种 I/O 事件通知的方式：水平触发和边缘触发，它们常用在套接字接口的文件描述符中。 水平触发：只要文件描述符可以非阻塞地执行 I/O ，就会触发通知。也就是说，应用程序可以随时检查文件描述符的状态，然后再根据状态，进行 I/O 操作。 边缘触发：只有在文件描述符的状态发生改变（也就是 I/O 请求达到）时，才发送一次通知。这时候，应用程序需要尽可能多地执行 I/O，直到无法继续读写，才可以停止。如果 I/O 没执行完，或者因为某种原因没来得及处理，那么这次通知也就丢失了。 I/O 多路复用的方法有很多实现方法，我带你来逐个分析一下。 第一种，使用非阻塞 I/O 和水平触发通知，比如使用 select 或者 poll。 根据刚才水平触发的原理，select 和 poll 需要从文件描述符列表中，找出哪些可以执行 I/O ，然后进行真正的网络 I/O 读写。由于 I/O 是非阻塞的，一个线程中就可以同时监控一批套接字的文件描述符，这样就达到了单线程处理多请求的目的。所以，这种方式的最大优点，是对应用程序比较友好，它的 API 非常简单。 但是，应用软件使用 select 和 poll 时，需要对这些文件描述符列表进行轮询，这样，请求数多的时候就会比较耗时。并且，select 和 poll 还有一些其他的限制。 select 使用固定长度的位相量，表示文件描述符的集合，因此会有最大描述符数量的限制。比如，在 32 位系统中，默认限制是 1024。并且，在 select 内部，检查套接字状态是用轮询的方法，再加上应用软件使用时的轮询，就变成了一个 O(n^2) 的关系。 而 poll 改进了 select 的表示方法，换成了一个没有固定长度的数组，这样就没有了最大描述符数量的限制（当然还会受到系统文件描述符限制）。但应用程序在使用 poll 时，同样需要对文件描述符列表进行轮询，这样，处理耗时跟描述符数量就是 O(N) 的关系。 除此之外，应用程序每次调用 select 和 poll 时，还需要把文件描述符的集合，从用户空间传入内核空间，由内核修改后，再传出到用户空间中。这一来一回的内核空间与用户空间切换，也增加了处理成本。 有没有什么更好的方式来处理呢？答案自然是肯定的。 第二种，使用非阻塞 I/O 和边缘触发通知，比如 epoll。既然 select 和 poll 有那么多的问题，就需要继续对其进行优化，而 epoll 就很好地解决了这些问题。 epoll 使用红黑树，在内核中管理文件描述符的集合，这样，就不需要应用程序在每次操作时都传入、传出这个集合。 epoll 使用事件驱动的机制，只关注有 I/O 事件发生的文件描述符，不需要轮询扫描整个集合。 不过要注意，epoll 是在 Linux 2.6 中才新增的功能（2.4 虽然也有，但功能不完善）。由于边缘触发只在文件描述符可读或可写事件发生时才通知，那么应用程序就需要尽可能多地执行 I/O，并要处理更多的异常事件。 第三种，使用异步 I/O（Asynchronous I/O，简称为 AIO）。 在前面文件系统原理的内容中，我曾介绍过异步 I/O 与同步 I/O 的区别。异步 I/O 允许应用程序同时发起很多 I/O 操作，而不用等待这些操作完成。而在 I/O 完成后，系统会用事件通知（比如信号或者回调函数）的方式，告诉应用程序。这时，应用程序才会去查询 I/O 操作的结果。 异步 I/O 也是到了 Linux 2.6 才支持的功能，并且在很长时间里都处于不完善的状态，比如 glibc 提供的异步 I/O 库，就一直被社区诟病。同时，由于异步 I/O 跟我们的直观逻辑不太一样，想要使用的话，一定要小心设计，其使用难度比较高。 工作模型优化 了解了 I/O 模型后，请求处理的优化就比较直观了。 使用 I/O 多路复用后，就可以在一个进程或线程中处理多个请求，其中，又有下面两种不同的工作模型。 第一种，主进程 + 多个 worker 子进程，这也是最常用的一种模型。这种方法的一个通用工作模式就是：主进程执行 bind() + listen() 后，创建多个子进程；然后，在每个子进程中，都通过 accept() 或 epoll_wait() ，来处理相同的套接字。 比如，最常用的反向代理服务器 Nginx 就是这么工作的。它也是由主进程和多个 worker 进程组成。主进程主要用来初始化套接字，并管理子进程的生命周期；而 worker 进程，则负责实际的请求处理。我画了一张图来表示这个关系。 这里要注意，accept() 和 epoll_wait() 调用，还存在一个惊群的问题。换句话说，当网络 I/O 事件发生时，多个进程被同时唤醒，但实际上只有一个进程来响应这个事件，其他被唤醒的进程都会重新休眠。 其中，accept() 的惊群问题，已经在 Linux 2.6 中解决了； 而 epoll 的问题，到了 Linux 4.5 ，才通过 EPOLLEXCLUSIVE 解决。 为了避免惊群问题， Nginx 在每个 worker 进程中，都增加一个了全局锁（accept_mutex）。这些 worker 进程需要首先竞争到锁，只有竞争到锁的进程，才会加入到 epoll 中，这样就确保只有一个 worker 子进程被唤醒。 不过，根据前面 CPU 模块的学习，你应该还记得，进程的管理、调度、上下文切换的成本非常高。那为什么使用多进程模式的 Nginx ，却具有非常好的性能呢？ 这里最主要的一个原因就是，这些 worker 进程，实际上并不需要经常创建和销毁，而是在没任务时休眠，有任务时唤醒。只有在 worker 由于某些异常退出时，主进程才需要创建新的进程来代替它。 当然，你也可以用线程代替进程：主线程负责套接字初始化和子线程状态的管理，而子线程则负责实际的请求处理。由于线程的调度和切换成本比较低，实际上你可以进一步把 epoll_wait() 都放到主线程中，保证每次事件都只唤醒主线程，而子线程只需要负责后续的请求处理。 第二种，监听到相同端口的多进程模型。在这种方式下，所有的进程都监听相同的接口，并且开启 SO_REUSEPORT 选项，由内核负责将请求负载均衡到这些监听进程中去。这一过程如下图所示。 由于内核确保了只有一个进程被唤醒，就不会出现惊群问题了。比如，Nginx 在 1.9.1 中就已经支持了这种模式。 不过要注意，想要使用 SO_REUSEPORT 选项，需要用 Linux 3.9 以上的版本才可以。 C1000K 基于 I/O 多路复用和请求处理的优化，C10K 问题很容易就可以解决。不过，随着摩尔定律带来的服务器性能提升，以及互联网的普及，你并不难想到，新兴服务会对性能提出更高的要求。 很快，原来的 C10K 已经不能满足需求，所以又有了 C100K 和 C1000K，也就是并发从原来的 1 万增加到 10 万、乃至 100 万。从 1 万到 10 万，其实还是基于 C10K 的这些理论，epoll 配合线程池，再加上 CPU、内存和网络接口的性能和容量提升。大部分情况下，C100K 很自然就可以达到。 那么，再进一步，C1000K 是不是也可以很容易就实现呢？这其实没有那么简单了。 首先从物理资源使用上来说，100 万个请求需要大量的系统资源。比如， 假设每个请求需要 16KB 内存的话，那么总共就需要大约 15 GB 内存。 而从带宽上来说，假设只有 20% 活跃连接，即使每个连接只需要 1KB/s 的吞吐量，总共也需要 1.6 Gb/s 的吞吐量。千兆网卡显然满足不了这么大的吞吐量，所以还需要配置万兆网卡，或者基于多网卡 Bonding 承载更大的吞吐量。 其次，从软件资源上来说，大量的连接也会占用大量的软件资源，比如文件描述符的数量、连接状态的跟踪（CONNTRACK）、网络协议栈的缓存大小（比如套接字读写缓存、TCP 读写缓存）等等。 最后，大量请求带来的中断处理，也会带来非常高的处理成本。这样，就需要多队列网卡、中断负载均衡、CPU 绑定、RPS/RFS（软中断负载均衡到多个 CPU 核上），以及将网络包的处理卸载（Offload）到网络设备（如 TSO/GSO、LRO/GRO、VXLAN OFFLOAD）等各种硬件和软件的优化。 C1000K 的解决方法，本质上还是构建在 epoll 的非阻塞 I/O 模型上。只不过，除了 I/O 模型之外，还需要从应用程序到 Linux 内核、再到 CPU、内存和网络等各个层次的深度优化，特别是需要借助硬件，来卸载那些原来通过软件处理的大量功能。 C10M 显然，人们对于性能的要求是无止境的。再进一步，有没有可能在单机中，同时处理 1000 万的请求呢？这也就是 C10M 问题。 实际上，在 C1000K 问题中，各种软件、硬件的优化很可能都已经做到头了。特别是当升级完硬件（比如足够多的内存、带宽足够大的网卡、更多的网络功能卸载等）后，你可能会发现，无论你怎么优化应用程序和内核中的各种网络参数，想实现 1000 万请求的并发，都是极其困难的。 究其根本，还是 Linux 内核协议栈做了太多太繁重的工作。从网卡中断带来的硬中断处理程序开始，到软中断中的各层网络协议处理，最后再到应用程序，这个路径实在是太长了，就会导致网络包的处理优化，到了一定程度后，就无法更进一步了。 要解决这个问题，最重要就是跳过内核协议栈的冗长路径，把网络包直接送到要处理的应用程序那里去。这里有两种常见的机制，DPDK 和 XDP。 第一种机制，DPDK，是用户态网络的标准。它跳过内核协议栈，直接由用户态进程通过轮询的方式，来处理网络接收。 说起轮询，你肯定会下意识认为它是低效的象征，但是进一步反问下自己，它的低效主要体现在哪里呢？是查询时间明显多于实际工作时间的情况下吧！那么，换个角度来想，如果每时每刻都有新的网络包需要处理，轮询的优势就很明显了。比如： 在 PPS 非常高的场景中，查询时间比实际工作时间少了很多，绝大部分时间都在处理网络包； 而跳过内核协议栈后，就省去了繁杂的硬中断、软中断再到 Linux 网络协议栈逐层处理的过程，应用程序可以针对应用的实际场景，有针对性地优化网络包的处理逻辑，而不需要关注所有的细节。 此外，DPDK 还通过大页、CPU 绑定、内存对齐、流水线并发等多种机制，优化网络包的处理效率。 第二种机制，XDP（eXpress Data Path），则是 Linux 内核提供的一种高性能网络数据路径。它允许网络包，在进入内核协议栈之前，就进行处理，也可以带来更高的性能。XDP 底层跟我们之前用到的 bcc-tools 一样，都是基于 Linux 内核的 eBPF 机制实现的。 XDP 的原理如下图所示： 你可以看到，XDP 对内核的要求比较高，需要的是 Linux 4.8 以上版本，并且它也不提供缓存队列。基于 XDP 的应用程序通常是专用的网络应用，常见的有 IDS（入侵检测系统）、DDoS 防御、 cilium 容器网络插件等。 总结 C10K 问题的根源，一方面在于系统有限的资源；另一方面，也是更重要的因素，是同步阻塞的 I/O 模型以及轮询的套接字接口，限制了网络事件的处理效率。Linux 2.6 中引入的 epoll ，完美解决了 C10K 的问题，现在的高性能网络方案都基于 epoll。 从 C10K 到 C100K ，可能只需要增加系统的物理资源就可以满足；但从 C100K 到 C1000K ，就不仅仅是增加物理资源就能解决的问题了。这时，就需要多方面的优化工作了，从硬件的中断处理和网络功能卸载、到网络协议栈的文件描述符数量、连接状态跟踪、缓存队列等内核的优化，再到应用程序的工作模型优化，都是考虑的重点。 再进一步，要实现 C10M ，就不只是增加物理资源，或者优化内核和应用程序可以解决的问题了。这时候，就需要用 XDP 的方式，在内核协议栈之前处理网络包；或者用 DPDK 直接跳过网络协议栈，在用户空间通过轮询的方式直接处理网络包。 当然了，实际上，在大多数场景中，我们并不需要单机并发 1000 万的请求。通过调整系统架构，把这些请求分发到多台服务器中来处理，通常是更简单和更容易扩展的方案。 本篇文章为转载内容。原文链接：https://blog.csdn.net/qq_23864697/article/details/114626793。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

... 17 年了， 担任架构师的职位也超过了 10 年，担任过像 HP、Amazon 这样的世界级团队的架构师，也担任过像汇量科技这样快速成长的中小企业的技术领导。应 InfoQ 邀请分享一下我的工作感悟，分享内容部分来自成功总结，更多是来自失败的反思，希望我踩过的坑大家可以不用再踩。 “提出问题”难于“解决问题” 作为技术人员，我们已经习惯于作为问题的解决者给出设计方案，而很少以问题提出者的身份去思考设计方案。团队中常见的典型矛盾，就是产品团队和研发团队之间的矛盾。作为研发团队，我们常吐槽产品团队的需求不合理、不懂技术等。其实我们可以试着把自己的工作再往前移一下，不仅仅是去设计架构、实现产品的需求，同时也试着去实现客户的需求，甚至发现潜在的需求。 这时我们就变成了在设计上提出问题的人，你会发现提出问题的同时，在很多时候也需要同样深入的思考。设计一个好的问题，甚至比解决问题更难。 其实即便是软件开发领域的大神 Frederick P. Brooks Jr.（《人月神话》的作者）也会有同样的感叹。 “The hardest part of design is deciding what to design.” – 《The design of design》, by Frederick P. Brooks Jr. 决定“不要什么”比“要什么”更难 也许是由于人性的贪婪，对于软件系统我们同样想要更多：更多功能、更好的性能、更好的伸缩性、扩展性等等。作为软件架构师要明白软件架构设计就是一种取舍或平衡。当大家都在往里面加东西的时候，架构师更应该来做这个说“不”的人。 软件设计和定义过程中存在很多取舍，例如： 完善功能和尽早发布的取舍。 伸缩性和性能的取舍。 著名的 CAP 原则，就是一个很好的取舍指导策略。为了更好的取舍，保持架构风格的一致性，在一开始架构师就应该根据系统的实际需求来定义一些取舍的原则，如： 数据一致性拥有最高优先级。 提前发布核心功能优于完整发布等。 非功能性需求决定架构 因为软件是为了满足客户的功能性需求的，所以很多设计人员可能会认为架构是由要实现的功能性需求决定的。但实际上真正决定软件架构的其实是非功能性需求。 架构师要更加关注非功能性需求，常见的非功能性包括：性能，伸缩性，扩展性和可维护性等，甚至还包括团队技术水平和发布时间要求。能实现功能的设计总是有很多，考虑了非功能性需求后才能筛选出最合适的设计。 以上架构模式来自《面向模式的软件架构》的第一卷，这套书多年来一直是架构师的必读经典。面向架构的模式就是为不同的非功能性需求提供了很好的参考和指导。图中的 Micro-Kernel 模式，更加关注可扩展性和可用性（错误隔离）。 “简单”并不“容易” 很多架构师都会常常提到保持简单，但是有时候我们会混淆简单和容易。简单和容易在英语里也是两个词“simple”和“easy”。 “Simple can be harder than complex: You have to work hard to get your thinking clean to make it simple. But it’s worth it in the end because once you get there, you can move mountains. To be truly simple, you have to go really deep.” –SteveJobs 真正的一些简单的方法其实来自于对问题和技术更深入的理解。这些方案往往不是容易获得的、表面上的方法。简单可以说蕴含着一种深入的技巧在其中。 下面我来举一个例子。 首先我们来回顾一下软件生命周期中各个阶段的成本消耗占比。以下是来一个知名统计机构的分析报告。我们可以看到占比最大的是维护部分，对于这一部分的简化将最具有全局意义。 我曾经开发过一个设备管理系统，移动运营商通过这个系统来管理移动设备，实现包括设备的自动注册、固件和软件的同步等管理功能。这些功能是通过一些管理系统与移动设备间的预定义的交互协议来完成的。 电信专家们会根据业务场景及需求来调整和新增这些交互协议。起初我们采用了一种容易实现的方式，即团队中的软件工程会根据电信专家的说明，将协议实现为对应代码。 之后我们很快发现这样的方式，让我们的工作变得没那么简单。 “I believe that the hardest part of software projects, the most common source of project failure, is communication with the customers and users of that software.” –Martin Fowler 正如软件开发大师 MartinFowler 提到的，“沟通”往往是导致软件项目失败的主要原因。前面这个项目最大的问题是在系统上线后的运行维护阶段，电信专家和开发工程师之间会不断就新的协议修改和增加进行持续的沟通，而他们的领域知识和词汇都有很大的差别，这会大大影响沟通的效率。因此这期间系统的运行维护（协议的修改）变得十分艰难，不仅协议更新上线时间慢，而且由于软件工程对于电信协议理解程度有限，很多问题都要在实际上线使用后才能被电信专家发现，导致了很多的交换和反复。 针对上面提到的问题，后来我们和电信专家一起设计了一种协议设计语言（并提供可视化的工具），这种设计语言使用的电信专家所熟悉的词汇。然后通过一个类似于编译器的程序将电信专家定义好的协议模型转换为内存中的 Java 结构。这样整个项目的运行和维护就变得简单高效了，省去了低效的交流和不准确人工转换。 我们可以看到一开始按电信专家的说明直接实现协议是更为容易的办法，但就整个软件生命周期来看却并不是一个简单高效的方法。 永远不要停止编码 架构师也是程序员，代码是软件的最终实现形态，停止编程会逐渐让你忘记作为程序员的感受，更重要的是忘记其中的“痛”，从而容易产生一些不切实际的设计。 大家可能听说过在 Amazon，高级副总裁级别的 Distinguish Engineer（如：James Gosling，Java 之父），他们每年的编码量也非常大，常在 10 万行以上。 风险优先 架构设计很重要的一点是识别可能存在的风险，尤其是非功能性需求实现的风险。因为这些风险往往没有功能性需求这么容易在初期被发现，但修正的代价通常要比修正功能性需求大非常多，甚至可能导致项目的失败，前面我们也提到了非功能性需求决定了架构，如数据一致性要求、响应延迟要求等。 我们应该通过原型或在早期的迭代中确认风险能够通过合理的架构得以解决。 绝对不要把风险放到最后，就算是一个项目要失败也要让它快速失败，这也是一种敏捷。 从“问题”开始，而不是“技术” 技术人员对于新技术的都有着一种与身俱来的激情，总是乐于去学习新技术，同时也更有激情去使用新技术。但是这也同样容易导致一个通病，就是“当我们有一个锤子的时候看什么都是钉子”，使用一些不适合的技术去解决手边的问题，常常会导致简单问题复杂化。 我曾经的一个团队维护过这样一个简单的服务，起初就是一个用 MySQL 作数据存储的简单服务，由团队的一个成员来开发和维护。后来，这位成员对当时新出的 DynamoDB 产生了兴趣，并学习了相关知识。 然后就发生下面这样的事： 用DynamoDB替换了MySQL。 很快发现DynamoDB并不能很好的支持事务特性，在当时只有一个性能极差的客户端类库来支持事物，由于采用客户端方式，引入了大量的额外交互，导致性能差别达7倍之多。这时候，这个同学就采用了当时在NoSQL领域广泛流行的最终一致技术，通过一个Pub-Sub消息队列来实现最终一致（即当某对象的值发生改变后会产生一个事件，然后关注这一改变的逻辑，就会订阅这个通知，并改变于其相关数据，从而实现不同数据的最终一致）。 接着由于DynamoDB无法提供SQL那样方便的查询机制，为了实现数据分析就又引入了EMR/MapReduceJob。 到此，大家可以看到实现一样的功能，但是复杂性大大增加，维护工作也由一个人变成了一个团队。 过度忙碌使你落后 对于 IT 人而言忙碌已成为了习惯，加班常挂在嘴边。“996”工作制似乎也变成了公司高效的标志。而事实上过度的忙碌使你落后。经常遇见一些朋友，在一个公司没日没夜的干了几年，没有留一点学习时间给自己。几年之后倒是对公司越来越“忠诚”了，但忙碌的工作同时也导致了没有时间更新知识，使得自己已经落后了，连跳槽的能力和勇气都失去了。 过度忙碌会导致没有时间学习和更新自己的知识，尤其在这个高速发展的时代。我在工作经历中发现过度繁忙通常会带来以下问题： 缺乏学习导致工作能力没有提升，而面对的问题却变得日益复杂。 技术和业务上没有更大的领先优势，只能被动紧紧追赶。试想一下，要是你都领先同行业五年了，还会在乎通过加班来早一个月发布吗？ 反过来上面这些问题会导致你更加繁忙，进而更没有时间提高自己的技术技能，很快就形成了一个恶性循环。 练过健身的朋友都知道，光靠锻炼是不行的，营养补充和锻炼同样重要。个人技术成长其实也一样，实践和学习是一样重要的，当你在一个领域工作了一段时间以后，工作对你而言就主要是实践了，随着你对该领域的熟悉，能学习的到技术会越来越少。所以每个技术人员都要保证充足的学习时间，否则很容易成为井底之蛙，从而陷入前面提到的恶性循环。 最后，以伟大诗人屈原的诗句和大家共勉：“路漫漫其修远兮，吾将上下而求索“。希望我们大家都可以不忘初心，保持匠心！ 作者简介： 蔡超，Mobvista 技术 VP 兼首席架构师，SpotMax 云服务创始人。拥有超过 15 年的软件开发经验，其中 9 年任世界级 IT 公司软件架构师/首席软件架构师。2017 年加入 Mobvista，任公司技术副总裁及首席架构师，领导公司的数字移动营销平台的开发，该平台完全建立于云计算技术之上，每天处理来自全球不同 region 的超过 600 亿次的请求。 在加入 Mobvista 之前，曾任亚马逊全球直运平台首席架构师，亚马逊（中国）首席架构师，曾领导了亚马逊的全球直运平台的开发，并领导中国团队通过 AI 及云计算技术为中国客户打造更好的本地体验；曾任 HP（中国）移动设备管理系统首席软件架构师，该系统曾是全球最大的无线设备管理系统（OMA DM）（客户包括中国移动，中国联通，中国电信等）；曾任北京天融信网络安全技术公司，首席软件架构师，领导开发的网络安全管理系统（TopAnalyzer）至今仍被政府重要部门及军队广为采用，该系统也曾成功应用于 2008 北京奥运，2010 上海世博等重要事件的网络安全防护。 本篇文章为转载内容。原文链接：https://blog.csdn.net/Honnyee/article/details/111896981。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

...人工智能、云计算到大数据分析，新兴技术领域的崛起为技术人员提供了更为广阔的发展舞台。例如，在字节跳动等头部互联网企业中，技术研发人员不仅能够参与到创新项目中，还可能获得丰厚的股权激励，实现个人财富和职业价值的双重提升（参考《字节跳动员工持股计划：释放的价值与激励机制解读》）。 此外，对于技术人员而言，持续学习与技术积累显得尤为关键。近日，《中国程序员技术能力现状与发展趋势报告》指出，随着容器化、微服务架构的广泛应用，以及Docker、Kubernetes等技术的普及，对底层网络工作原理的理解和应用能力成为衡量开发人员技术水平的重要标准之一。 同时，团队协作与领导力培养也日益受到重视。据《哈佛商业评论》近期文章所述，在现代企业中，具有主动发现问题、解决问题意识，并能带领团队共同创新的技术管理者，往往能在组织内部获得更快的成长和更高的认可度。 综上所述，无论是在选择行业赛道还是在实际工作中，技术人才都应紧跟时代潮流，深化专业技能，积极主动地挖掘潜在问题并寻求解决方案，从而在快速变化的互联网行业中取得长足发展。而类似于张彦飞allen这样的经历分享，无疑为后来者提供了宝贵的经验借鉴和启示。

...案例 需求分析和技术架构 广告点击系统实时分析 广告来自于广告或者移动App等，广告需要设定在具体的广告位，当用户点击广告的时候，一般都会通过ajax或Socket往后台发送日志数据，在这里我们是要做基于SparkStreaming做实时在线统计。那么数据就需要放进消息系统（Kafka）中，我们的Spark Streaming应用程序就会去Kafka中Pull数据过来进行计算和消费，并把计算后的数据放入到持久化系统中（MySQL） 广告点击系统实时分析的意义：因为可以在线实时的看见广告的投放效果，就为广告的更大规模的投入和调整打下了坚实的基础，从而为公司带来最大化的经济回报。 核心需求： 1、实时黑名单动态过滤出有效的用户广告点击行为：因为黑名单用户可能随时出现，所以需要动态更新； 2、在线计算广告点击流量； 3、Top3热门广告； 4、每个广告流量趋势； 5、广告点击用户的区域分布分析 6、最近一分钟的广告点击量； 7、整个广告点击Spark Streaming处理程序724小时运行； 数据格式： 时间、用户、广告、城市等 技术细节： 在线计算用户点击的次数分析，屏蔽IP等； 使用updateStateByKey或者mapWithState进行不同地区广告点击排名的计算； Spark Streaming+Spark SQL+Spark Core等综合分析数据； 使用Window类型的操作； 高可用和性能调优等等； 流量趋势，一般会结合DB等； Spark Core / /package com.tom.spark.SparkApps.sparkstreaming;import java.util.Date;import java.util.HashMap;import java.util.Map;import java.util.Properties;import java.util.Random;import kafka.javaapi.producer.Producer;import kafka.producer.KeyedMessage;import kafka.producer.ProducerConfig;/ 数据生成代码，Kafka Producer产生数据/public class MockAdClickedStat {/ @param args/public static void main(String[] args) {final Random random = new Random();final String[] provinces = new String[]{"Guangdong", "Zhejiang", "Jiangsu", "Fujian"};final Map<String, String[]> cities = new HashMap<String, String[]>();cities.put("Guangdong", new String[]{"Guangzhou", "Shenzhen", "Dongguan"});cities.put("Zhejiang", new String[]{"Hangzhou", "Wenzhou", "Ningbo"});cities.put("Jiangsu", new String[]{"Nanjing", "Suzhou", "Wuxi"});cities.put("Fujian", new String[]{"Fuzhou", "Xiamen", "Sanming"});final String[] ips = new String[] {"192.168.112.240","192.168.112.239","192.168.112.245","192.168.112.246","192.168.112.247","192.168.112.248","192.168.112.249","192.168.112.250","192.168.112.251","192.168.112.252","192.168.112.253","192.168.112.254",};/ Kafka相关的基本配置信息/Properties kafkaConf = new Properties();kafkaConf.put("serializer.class", "kafka.serializer.StringEncoder");kafkaConf.put("metadeta.broker.list", "Master:9092,Worker1:9092,Worker2:9092");ProducerConfig producerConfig = new ProducerConfig(kafkaConf);final Producer<Integer, String> producer = new Producer<Integer, String>(producerConfig);new Thread(new Runnable() {public void run() {while(true) {//在线处理广告点击流的基本数据格式：timestamp、ip、userID、adID、province、cityLong timestamp = new Date().getTime();String ip = ips[random.nextInt(12)]; //可以采用网络上免费提供的ip库int userID = random.nextInt(10000);int adID = random.nextInt(100);String province = provinces[random.nextInt(4)];String city = cities.get(province)[random.nextInt(3)];String clickedAd = timestamp + "\t" + ip + "\t" + userID + "\t" + adID + "\t" + province + "\t" + city;producer.send(new KeyedMessage<Integer, String>("AdClicked", clickedAd));try {Thread.sleep(50);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }} }).start();} } package com.tom.spark.SparkApps.sparkstreaming;import java.sql.Connection;import java.sql.DriverManager;import java.sql.PreparedStatement;import java.sql.ResultSet;import java.sql.SQLException;import java.util.ArrayList;import java.util.Arrays;import java.util.HashMap;import java.util.HashSet;import java.util.Iterator;import java.util.List;import java.util.Map;import java.util.Set;import java.util.concurrent.LinkedBlockingQueue;import kafka.serializer.StringDecoder;import org.apache.spark.SparkConf;import org.apache.spark.api.java.JavaPairRDD;import org.apache.spark.api.java.JavaRDD;import org.apache.spark.api.java.JavaSparkContext;import org.apache.spark.api.java.function.Function;import org.apache.spark.api.java.function.Function2;import org.apache.spark.api.java.function.PairFunction;import org.apache.spark.api.java.function.VoidFunction;import org.apache.spark.sql.DataFrame;import org.apache.spark.sql.Row;import org.apache.spark.sql.RowFactory;import org.apache.spark.sql.hive.HiveContext;import org.apache.spark.sql.types.DataTypes;import org.apache.spark.sql.types.StructType;import org.apache.spark.streaming.Durations;import org.apache.spark.streaming.api.java.JavaDStream;import org.apache.spark.streaming.api.java.JavaPairDStream;import org.apache.spark.streaming.api.java.JavaPairInputDStream;import org.apache.spark.streaming.api.java.JavaStreamingContext;import org.apache.spark.streaming.api.java.JavaStreamingContextFactory;import org.apache.spark.streaming.kafka.KafkaUtils;import com.google.common.base.Optional;import scala.Tuple2;/ 数据处理，Kafka消费者/public class AdClickedStreamingStats {/ @param args/public static void main(String[] args) {// TODO Auto-generated method stub//好处：1、checkpoint 2、工厂final SparkConf conf = new SparkConf().setAppName("SparkStreamingOnKafkaDirect").setMaster("hdfs://Master:7077/");final String checkpointDirectory = "hdfs://Master:9000/library/SparkStreaming/CheckPoint_Data";JavaStreamingContextFactory factory = new JavaStreamingContextFactory() {public JavaStreamingContext create() {// TODO Auto-generated method stubreturn createContext(checkpointDirectory, conf);} };/ 可以从失败中恢复Driver，不过还需要指定Driver这个进程运行在Cluster，并且在提交应用程序的时候制定--supervise;/JavaStreamingContext javassc = JavaStreamingContext.getOrCreate(checkpointDirectory, factory);/ 第三步：创建Spark Streaming输入数据来源input Stream: 1、数据输入来源可以基于File、HDFS、Flume、Kafka、Socket等 2、在这里我们指定数据来源于网络Socket端口，Spark Streaming连接上该端口并在运行的时候一直监听该端口的数据 (当然该端口服务首先必须存在），并且在后续会根据业务需要不断有数据产生（当然对于Spark Streaming 应用程序的运行而言，有无数据其处理流程都是一样的） 3、如果经常在每间隔5秒钟没有数据的话不断启动空的Job其实会造成调度资源的浪费，因为并没有数据需要发生计算；所以 实际的企业级生成环境的代码在具体提交Job前会判断是否有数据，如果没有的话就不再提交Job；///创建Kafka元数据来让Spark Streaming这个Kafka Consumer利用Map<String, String> kafkaParameters = new HashMap<String, String>();kafkaParameters.put("metadata.broker.list", "Master:9092,Worker1:9092,Worker2:9092");Set<String> topics = new HashSet<String>();topics.add("SparkStreamingDirected");JavaPairInputDStream<String, String> adClickedStreaming = KafkaUtils.createDirectStream(javassc, String.class, String.class, StringDecoder.class, StringDecoder.class,kafkaParameters, topics);/因为要对黑名单进行过滤，而数据是在RDD中的，所以必然使用transform这个函数； 但是在这里我们必须使用transformToPair，原因是读取进来的Kafka的数据是Pair<String,String>类型, 另一个原因是过滤后的数据要进行进一步处理，所以必须是读进的Kafka数据的原始类型 在此再次说明，每个Batch Duration中实际上讲输入的数据就是被一个且仅被一个RDD封装的，你可以有多个 InputDStream，但其实在产生job的时候，这些不同的InputDStream在Batch Duration中就相当于Spark基于HDFS 数据操作的不同文件来源而已罢了。/JavaPairDStream<String, String> filteredadClickedStreaming = adClickedStreaming.transformToPair(new Function<JavaPairRDD<String,String>, JavaPairRDD<String,String>>() {public JavaPairRDD<String, String> call(JavaPairRDD<String, String> rdd) throws Exception {/ 在线黑名单过滤思路步骤： 1、从数据库中获取黑名单转换成RDD，即新的RDD实例封装黑名单数据； 2、然后把代表黑名单的RDD的实例和Batch Duration产生的RDD进行Join操作， 准确的说是进行leftOuterJoin操作，也就是说使用Batch Duration产生的RDD和代表黑名单的RDD实例进行 leftOuterJoin操作，如果两者都有内容的话，就会是true，否则的话就是false 我们要留下的是leftOuterJoin结果为false； /final List<String> blackListNames = new ArrayList<String>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();jdbcWrapper.doQuery("SELECT FROM blacklisttable", null, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {while(result.next()){blackListNames.add(result.getString(1));} }});List<Tuple2<String, Boolean>> blackListTuple = new ArrayList<Tuple2<String,Boolean>>();for(String name : blackListNames) {blackListTuple.add(new Tuple2<String, Boolean>(name, true));}List<Tuple2<String, Boolean>> blacklistFromListDB = blackListTuple; //数据来自于查询的黑名单表并且映射成为<String, Boolean>JavaSparkContext jsc = new JavaSparkContext(rdd.context());/ 黑名单的表中只有userID，但是如果要进行join操作的话就必须是Key-Value，所以在这里我们需要 基于数据表中的数据产生Key-Value类型的数据集合/JavaPairRDD<String, Boolean> blackListRDD = jsc.parallelizePairs(blacklistFromListDB);/ 进行操作的时候肯定是基于userID进行join，所以必须把传入的rdd进行mapToPair操作转化成为符合格式的RDD/JavaPairRDD<String, Tuple2<String, String>> rdd2Pair = rdd.mapToPair(new PairFunction<Tuple2<String,String>, String, Tuple2<String, String>>() {public Tuple2<String, Tuple2<String, String>> call(Tuple2<String, String> t) throws Exception {// TODO Auto-generated method stubString userID = t._2.split("\t")[2];return new Tuple2<String, Tuple2<String,String>>(userID, t);} });JavaPairRDD<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> joined = rdd2Pair.leftOuterJoin(blackListRDD);JavaPairRDD<String, String> result = joined.filter(new Function<Tuple2<String,Tuple2<Tuple2<String,String>,Optional<Boolean>>>, Boolean>() {public Boolean call(Tuple2<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> tuple)throws Exception {// TODO Auto-generated method stubOptional<Boolean> optional = tuple._2._2;if(optional.isPresent() && optional.get()){return false;} else {return true;} }}).mapToPair(new PairFunction<Tuple2<String,Tuple2<Tuple2<String,String>,Optional<Boolean>>>, String, String>() {public Tuple2<String, String> call(Tuple2<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> t)throws Exception {// TODO Auto-generated method stubreturn t._2._1;} });return result;} });//广告点击的基本数据格式：timestamp、ip、userID、adID、province、cityJavaPairDStream<String, Long> pairs = filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, String> t) throws Exception {String[] splited=t._2.split("\t");String timestamp = splited[0]; //YYYY-MM-DDString ip = splited[1];String userID = splited[2];String adID = splited[3];String province = splited[4];String city = splited[5]; String clickedRecord = timestamp + "_" +ip + "_"+userID+"_"+adID+"_"+province +"_"+city;return new Tuple2<String, Long>(clickedRecord, 1L);} });/ 第4.3步：在单词实例计数为1基础上，统计每个单词在文件中出现的总次数/JavaPairDStream<String, Long> adClickedUsers= pairs.reduceByKey(new Function2<Long, Long, Long>() {public Long call(Long i1, Long i2) throws Exception{return i1 + i2;} });/判断有效的点击，复杂化的采用机器学习训练模型进行在线过滤 简单的根据ip判断1天不超过100次；也可以通过一个batch duration的点击次数判断是否非法广告点击，通过一个batch来判断是不完整的，还需要一天的数据也可以每一个小时来判断。/JavaPairDStream<String, Long> filterClickedBatch = adClickedUsers.filter(new Function<Tuple2<String,Long>, Boolean>() {public Boolean call(Tuple2<String, Long> v1) throws Exception {if (1 < v1._2){//更新一些黑名单的数据库表return false;} else { return true;} }});//filterClickedBatch.print();//写入数据库filterClickedBatch.foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {public Void call(JavaPairRDD<String, Long> rdd) throws Exception {rdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {public void call(Iterator<Tuple2<String, Long>> partition) throws Exception {//使用数据库连接池的高效读写数据库的方式将数据写入数据库mysql//例如一次插入 1000条 records，使用insertBatch 或 updateBatch//插入的用户数据信息：userID,adID,clickedCount,time//这里面有一个问题，可能出现两条记录的key是一样的，此时需要更新累加操作List<UserAdClicked> userAdClickedList = new ArrayList<UserAdClicked>();while(partition.hasNext()) {Tuple2<String, Long> record = partition.next();String[] splited = record._1.split("\t");UserAdClicked userClicked = new UserAdClicked();userClicked.setTimestamp(splited[0]);userClicked.setIp(splited[1]);userClicked.setUserID(splited[2]);userClicked.setAdID(splited[3]);userClicked.setProvince(splited[4]);userClicked.setCity(splited[5]);userAdClickedList.add(userClicked);}final List<UserAdClicked> inserting = new ArrayList<UserAdClicked>();final List<UserAdClicked> updating = new ArrayList<UserAdClicked>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();//表的字段timestamp、ip、userID、adID、province、city、clickedCountfor(final UserAdClicked clicked : userAdClickedList) {jdbcWrapper.doQuery("SELECT clickedCount FROM adclicked WHERE"+ " timestamp =? AND userID = ? AND adID = ?",new Object[]{clicked.getTimestamp(), clicked.getUserID(),clicked.getAdID()}, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {// TODO Auto-generated method stubif(result.next()) {long count = result.getLong(1);clicked.setClickedCount(count);updating.add(clicked);} else {inserting.add(clicked);clicked.setClickedCount(1L);} }});}//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(UserAdClicked insertRecord : inserting) {insertParametersList.add(new Object[] {insertRecord.getTimestamp(),insertRecord.getIp(),insertRecord.getUserID(),insertRecord.getAdID(),insertRecord.getProvince(),insertRecord.getCity(),insertRecord.getClickedCount()});}jdbcWrapper.doBatch("INSERT INTO adclicked VALUES(?, ?, ?, ?, ?, ?, ?)", insertParametersList);//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> updateParametersList = new ArrayList<Object[]>();for(UserAdClicked updateRecord : updating) {updateParametersList.add(new Object[] {updateRecord.getTimestamp(),updateRecord.getIp(),updateRecord.getUserID(),updateRecord.getAdID(),updateRecord.getProvince(),updateRecord.getCity(),updateRecord.getClickedCount() + 1});}jdbcWrapper.doBatch("UPDATE adclicked SET clickedCount = ? WHERE"+ " timestamp =? AND ip = ? AND userID = ? AND adID = ? "+ "AND province = ? AND city = ?", updateParametersList);} });return null;} });//再次过滤，从数据库中读取数据过滤黑名单JavaPairDStream<String, Long> blackListBasedOnHistory = filterClickedBatch.filter(new Function<Tuple2<String,Long>, Boolean>() {public Boolean call(Tuple2<String, Long> v1) throws Exception {//广告点击的基本数据格式：timestamp,ip,userID,adID,province,cityString[] splited = v1._1.split("\t"); //提取key值String date =splited[0];String userID =splited[2];String adID =splited[3];//查询一下数据库同一个用户同一个广告id点击量超过50次列入黑名单//接下来 根据date、userID、adID条件去查询用户点击广告的数据表，获得总的点击次数//这个时候基于点击次数判断是否属于黑名单点击int clickedCountTotalToday = 81 ;if (clickedCountTotalToday > 50) {return true;}else {return false ;} }});//map操作，找出用户的idJavaDStream<String> blackListuserIDBasedInBatchOnhistroy =blackListBasedOnHistory.map(new Function<Tuple2<String,Long>, String>() {public String call(Tuple2<String, Long> v1) throws Exception {// TODO Auto-generated method stubreturn v1._1.split("\t")[2];} });//有一个问题，数据可能重复，在一个partition里面重复，这个好办；//但多个partition不能保证一个用户重复，需要对黑名单的整个rdd进行去重操作。//rdd去重了，partition也就去重了，一石二鸟，一箭双雕// 找出了黑名单，下一步就写入黑名单数据库表中JavaDStream<String> blackListUniqueuserBasedInBatchOnhistroy = blackListuserIDBasedInBatchOnhistroy.transform(new Function<JavaRDD<String>, JavaRDD<String>>() {public JavaRDD<String> call(JavaRDD<String> rdd) throws Exception {// TODO Auto-generated method stubreturn rdd.distinct();} });// 下一步写入到数据表中blackListUniqueuserBasedInBatchOnhistroy.foreachRDD(new Function<JavaRDD<String>, Void>() {public Void call(JavaRDD<String> rdd) throws Exception {rdd.foreachPartition(new VoidFunction<Iterator<String>>() {public void call(Iterator<String> t) throws Exception {// TODO Auto-generated method stub//插入的用户信息可以只包含：useID//此时直接插入黑名单数据表即可。//写入数据库List<Object[]> blackList = new ArrayList<Object[]>();while(t.hasNext()) {blackList.add(new Object[]{t.next()});}JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();jdbcWrapper.doBatch("INSERT INTO blacklisttable values (?)", blackList);} });return null;} });/广告点击累计动态更新,每个updateStateByKey都会在Batch Duration的时间间隔的基础上进行广告点击次数的更新， 更新之后我们一般都会持久化到外部存储设备上，在这里我们存储到MySQL数据库中/JavaPairDStream<String, Long> updateStateByKeyDSteam = filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, String> t)throws Exception {String[] splited=t._2.split("\t");String timestamp = splited[0]; //YYYY-MM-DDString ip = splited[1];String userID = splited[2];String adID = splited[3];String province = splited[4];String city = splited[5]; String clickedRecord = timestamp + "_" +ip + "_"+userID+"_"+adID+"_"+province +"_"+city;return new Tuple2<String, Long>(clickedRecord, 1L);} }).updateStateByKey(new Function2<List<Long>, Optional<Long>, Optional<Long>>() {public Optional<Long> call(List<Long> v1, Optional<Long> v2)throws Exception {// v1:当前的Key在当前的Batch Duration中出现的次数的集合，例如{1，1，1，。。。，1}// v2:当前的Key在以前的Batch Duration中积累下来的结果；Long clickedTotalHistory = 0L; if(v2.isPresent()){clickedTotalHistory = v2.get();}for(Long one : v1) {clickedTotalHistory += one;}return Optional.of(clickedTotalHistory);} });updateStateByKeyDSteam.foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {public Void call(JavaPairRDD<String, Long> rdd) throws Exception {rdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {public void call(Iterator<Tuple2<String, Long>> partition) throws Exception {//使用数据库连接池的高效读写数据库的方式将数据写入数据库mysql//例如一次插入 1000条 records，使用insertBatch 或 updateBatch//插入的用户数据信息：timestamp、adID、province、city//这里面有一个问题，可能出现两条记录的key是一样的，此时需要更新累加操作List<AdClicked> AdClickedList = new ArrayList<AdClicked>();while(partition.hasNext()) {Tuple2<String, Long> record = partition.next();String[] splited = record._1.split("\t");AdClicked adClicked = new AdClicked();adClicked.setTimestamp(splited[0]);adClicked.setAdID(splited[1]);adClicked.setProvince(splited[2]);adClicked.setCity(splited[3]);adClicked.setClickedCount(record._2);AdClickedList.add(adClicked);}final List<AdClicked> inserting = new ArrayList<AdClicked>();final List<AdClicked> updating = new ArrayList<AdClicked>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();//表的字段timestamp、ip、userID、adID、province、city、clickedCountfor(final AdClicked clicked : AdClickedList) {jdbcWrapper.doQuery("SELECT clickedCount FROM adclickedcount WHERE"+ " timestamp = ? AND adID = ? AND province = ? AND city = ?",new Object[]{clicked.getTimestamp(), clicked.getAdID(),clicked.getProvince(), clicked.getCity()}, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {// TODO Auto-generated method stubif(result.next()) {long count = result.getLong(1);clicked.setClickedCount(count);updating.add(clicked);} else {inserting.add(clicked);clicked.setClickedCount(1L);} }});}//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(AdClicked insertRecord : inserting) {insertParametersList.add(new Object[] {insertRecord.getTimestamp(),insertRecord.getAdID(),insertRecord.getProvince(),insertRecord.getCity(),insertRecord.getClickedCount()});}jdbcWrapper.doBatch("INSERT INTO adclickedcount VALUES(?, ?, ?, ?, ?)", insertParametersList);//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> updateParametersList = new ArrayList<Object[]>();for(AdClicked updateRecord : updating) {updateParametersList.add(new Object[] {updateRecord.getClickedCount(),updateRecord.getTimestamp(),updateRecord.getAdID(),updateRecord.getProvince(),updateRecord.getCity()});}jdbcWrapper.doBatch("UPDATE adclickedcount SET clickedCount = ? WHERE"+ " timestamp =? AND adID = ? AND province = ? AND city = ?", updateParametersList);} });return null;} });/ 对广告点击进行TopN计算，计算出每天每个省份Top5排名的广告 因为我们直接对RDD进行操作，所以使用了transfomr算子；/updateStateByKeyDSteam.transform(new Function<JavaPairRDD<String,Long>, JavaRDD<Row>>() {public JavaRDD<Row> call(JavaPairRDD<String, Long> rdd) throws Exception {JavaRDD<Row> rowRDD = rdd.mapToPair(new PairFunction<Tuple2<String,Long>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, Long> t)throws Exception {// TODO Auto-generated method stubString[] splited=t._1.split("_");String timestamp = splited[0]; //YYYY-MM-DDString adID = splited[3];String province = splited[4];String clickedRecord = timestamp + "_" + adID + "_" + province;return new Tuple2<String, Long>(clickedRecord, t._2);} }).reduceByKey(new Function2<Long, Long, Long>() {public Long call(Long v1, Long v2) throws Exception {// TODO Auto-generated method stubreturn v1 + v2;} }).map(new Function<Tuple2<String,Long>, Row>() {public Row call(Tuple2<String, Long> v1) throws Exception {// TODO Auto-generated method stubString[] splited=v1._1.split("_");String timestamp = splited[0]; //YYYY-MM-DDString adID = splited[3];String province = splited[4];return RowFactory.create(timestamp, adID, province, v1._2);} });StructType structType = DataTypes.createStructType(Arrays.asList(DataTypes.createStructField("timestamp", DataTypes.StringType, true),DataTypes.createStructField("adID", DataTypes.StringType, true),DataTypes.createStructField("province", DataTypes.StringType, true),DataTypes.createStructField("clickedCount", DataTypes.LongType, true)));HiveContext hiveContext = new HiveContext(rdd.context());DataFrame df = hiveContext.createDataFrame(rowRDD, structType);df.registerTempTable("topNTableSource");DataFrame result = hiveContext.sql("SELECT timestamp, adID, province, clickedCount, FROM"+ " (SELECT timestamp, adID, province,clickedCount, "+ "ROW_NUMBER() OVER(PARTITION BY province ORDER BY clickeCount DESC) rank "+ "FROM topNTableSource) subquery "+ "WHERE rank <= 5");return result.toJavaRDD();} }).foreachRDD(new Function<JavaRDD<Row>, Void>() {public Void call(JavaRDD<Row> rdd) throws Exception {// TODO Auto-generated method stubrdd.foreachPartition(new VoidFunction<Iterator<Row>>() {public void call(Iterator<Row> t) throws Exception {// TODO Auto-generated method stubList<AdProvinceTopN> adProvinceTopN = new ArrayList<AdProvinceTopN>();while(t.hasNext()) {Row row = t.next();AdProvinceTopN item = new AdProvinceTopN();item.setTimestamp(row.getString(0));item.setAdID(row.getString(1));item.setProvince(row.getString(2));item.setClickedCount(row.getLong(3));adProvinceTopN.add(item);}// final List<AdProvinceTopN> inserting = new ArrayList<AdProvinceTopN>();// final List<AdProvinceTopN> updating = new ArrayList<AdProvinceTopN>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();Set<String> set = new HashSet<String>();for(AdProvinceTopN item: adProvinceTopN){set.add(item.getTimestamp() + "_" + item.getProvince());}//表的字段timestamp、adID、province、clickedCountArrayList<Object[]> deleteParametersList = new ArrayList<Object[]>();for(String deleteRecord : set) {String[] splited = deleteRecord.split("_");deleteParametersList.add(new Object[]{splited[0],splited[1]});}jdbcWrapper.doBatch("DELETE FROM adprovincetopn WHERE timestamp = ? AND province = ?", deleteParametersList);//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(AdProvinceTopN insertRecord : adProvinceTopN) {insertParametersList.add(new Object[] {insertRecord.getClickedCount(),insertRecord.getTimestamp(),insertRecord.getAdID(),insertRecord.getProvince()});}jdbcWrapper.doBatch("INSERT INTO adprovincetopn VALUES (?, ?, ?, ?)", insertParametersList);} });return null;} });/ 计算过去半个小时内广告点击的趋势 广告点击的基本数据格式：timestamp、ip、userID、adID、province、city/filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, String> t)throws Exception {String splited[] = t._2.split("\t");String adID = splited[3];String time = splited[0]; //Todo:后续需要重构代码实现时间戳和分钟的转换提取。此处需要提取出该广告的点击分钟单位return new Tuple2<String, Long>(time + "_" + adID, 1L);} }).reduceByKeyAndWindow(new Function2<Long, Long, Long>() {public Long call(Long v1, Long v2) throws Exception {// TODO Auto-generated method stubreturn v1 + v2;} }, new Function2<Long, Long, Long>() {public Long call(Long v1, Long v2) throws Exception {// TODO Auto-generated method stubreturn v1 - v2;} }, Durations.minutes(30), Durations.milliseconds(5)).foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {public Void call(JavaPairRDD<String, Long> rdd) throws Exception {// TODO Auto-generated method stubrdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {public void call(Iterator<Tuple2<String, Long>> partition)throws Exception {List<AdTrendStat> adTrend = new ArrayList<AdTrendStat>();// TODO Auto-generated method stubwhile(partition.hasNext()) {Tuple2<String, Long> record = partition.next();String[] splited = record._1.split("_");String time = splited[0];String adID = splited[1];Long clickedCount = record._2;/ 在插入数据到数据库的时候具体需要哪些字段？time、adID、clickedCount; 而我们通过J2EE技术进行趋势绘图的时候肯定是需要年、月、日、时、分这个维度的，所以我们在这里需要 年月日、小时、分钟这些时间维度；/AdTrendStat adTrendStat = new AdTrendStat();adTrendStat.setAdID(adID);adTrendStat.setClickedCount(clickedCount);adTrendStat.set_date(time); //Todo:获取年月日adTrendStat.set_hour(time); //Todo:获取小时adTrendStat.set_minute(time);//Todo:获取分钟adTrend.add(adTrendStat);}final List<AdTrendStat> inserting = new ArrayList<AdTrendStat>();final List<AdTrendStat> updating = new ArrayList<AdTrendStat>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();//表的字段timestamp、ip、userID、adID、province、city、clickedCountfor(final AdTrendStat trend : adTrend) {final AdTrendCountHistory adTrendhistory = new AdTrendCountHistory();jdbcWrapper.doQuery("SELECT clickedCount FROM adclickedtrend WHERE"+ " date =? AND hour = ? AND minute = ? AND AdID = ?",new Object[]{trend.get_date(), trend.get_hour(), trend.get_minute(),trend.getAdID()}, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {// TODO Auto-generated method stubif(result.next()) {long count = result.getLong(1);adTrendhistory.setClickedCountHistoryLong(count);updating.add(trend);} else { inserting.add(trend);} }});}//表的字段date、hour、minute、adID、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(AdTrendStat insertRecord : inserting) {insertParametersList.add(new Object[] {insertRecord.get_date(),insertRecord.get_hour(),insertRecord.get_minute(),insertRecord.getAdID(),insertRecord.getClickedCount()});}jdbcWrapper.doBatch("INSERT INTO adclickedtrend VALUES(?, ?, ?, ?, ?)", insertParametersList);//表的字段date、hour、minute、adID、clickedCountList<Object[]> updateParametersList = new ArrayList<Object[]>();for(AdTrendStat updateRecord : updating) {updateParametersList.add(new Object[] {updateRecord.getClickedCount(),updateRecord.get_date(),updateRecord.get_hour(),updateRecord.get_minute(),updateRecord.getAdID()});}jdbcWrapper.doBatch("UPDATE adclickedtrend SET clickedCount = ? WHERE"+ " date =? AND hour = ? AND minute = ? AND AdID = ?", updateParametersList);} });return null;} });;/ Spark Streaming 执行引擎也就是Driver开始运行，Driver启动的时候是位于一条新的线程中的，当然其内部有消息循环体，用于 接收应用程序本身或者Executor中的消息，/javassc.start();javassc.awaitTermination();javassc.close();}private static JavaStreamingContext createContext(String checkpointDirectory, SparkConf conf) {// If you do not see this printed, that means the StreamingContext has been loaded// from the new checkpointSystem.out.println("Creating new context");// Create the context with a 5 second batch sizeJavaStreamingContext ssc = new JavaStreamingContext(conf, Durations.seconds(10));ssc.checkpoint(checkpointDirectory);return ssc;} }class JDBCWrapper {private static JDBCWrapper jdbcInstance = null;private static LinkedBlockingQueue<Connection> dbConnectionPool = new LinkedBlockingQueue<Connection>();static {try {Class.forName("com.mysql.jdbc.Driver");} catch (ClassNotFoundException e) {// TODO Auto-generated catch blocke.printStackTrace();} }public static JDBCWrapper getJDBCInstance() {if(jdbcInstance == null) {synchronized (JDBCWrapper.class) {if(jdbcInstance == null) {jdbcInstance = new JDBCWrapper();} }}return jdbcInstance; }private JDBCWrapper() {for(int i = 0; i < 10; i++){try {Connection conn = DriverManager.getConnection("jdbc:mysql://Master:3306/sparkstreaming","root", "root");dbConnectionPool.put(conn);} catch (Exception e) {// TODO Auto-generated catch blocke.printStackTrace();} } }public synchronized Connection getConnection() {while(0 == dbConnectionPool.size()){try {Thread.sleep(20);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }return dbConnectionPool.poll();}public int[] doBatch(String sqlText, List<Object[]> paramsList){Connection conn = getConnection();PreparedStatement preparedStatement = null;int[] result = null;try {conn.setAutoCommit(false);preparedStatement = conn.prepareStatement(sqlText);for(Object[] parameters: paramsList) {for(int i = 0; i < parameters.length; i++){preparedStatement.setObject(i + 1, parameters[i]);} preparedStatement.addBatch();}result = preparedStatement.executeBatch();conn.commit();} catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} finally {if(preparedStatement != null) {try {preparedStatement.close();} catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} }if(conn != null) {try {dbConnectionPool.put(conn);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }}return result; }public void doQuery(String sqlText, Object[] paramsList, ExecuteCallBack callback){Connection conn = getConnection();PreparedStatement preparedStatement = null;ResultSet result = null;try {preparedStatement = conn.prepareStatement(sqlText);for(int i = 0; i < paramsList.length; i++){preparedStatement.setObject(i + 1, paramsList[i]);} result = preparedStatement.executeQuery();try {callback.resultCallBack(result);} catch (Exception e) {// TODO Auto-generated catch blocke.printStackTrace();} } catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} finally {if(preparedStatement != null) {try {preparedStatement.close();} catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} }if(conn != null) {try {dbConnectionPool.put(conn);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }} }}interface ExecuteCallBack {void resultCallBack(ResultSet result) throws Exception;}class UserAdClicked {private String timestamp;private String ip;private String userID;private String adID;private String province;private String city;private Long clickedCount;public String getTimestamp() {return timestamp;}public void setTimestamp(String timestamp) {this.timestamp = timestamp;}public String getIp() {return ip;}public void setIp(String ip) {this.ip = ip;}public String getUserID() {return userID;}public void setUserID(String userID) {this.userID = userID;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public String getProvince() {return province;}public void setProvince(String province) {this.province = province;}public String getCity() {return city;}public void setCity(String city) {this.city = city;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdClicked {private String timestamp;private String adID;private String province;private String city;private Long clickedCount;public String getTimestamp() {return timestamp;}public void setTimestamp(String timestamp) {this.timestamp = timestamp;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public String getProvince() {return province;}public void setProvince(String province) {this.province = province;}public String getCity() {return city;}public void setCity(String city) {this.city = city;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdProvinceTopN {private String timestamp;private String adID;private String province;private Long clickedCount;public String getTimestamp() {return timestamp;}public void setTimestamp(String timestamp) {this.timestamp = timestamp;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public String getProvince() {return province;}public void setProvince(String province) {this.province = province;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdTrendStat {private String _date;private String _hour;private String _minute;private String adID;private Long clickedCount;public String get_date() {return _date;}public void set_date(String _date) {this._date = _date;}public String get_hour() {return _hour;}public void set_hour(String _hour) {this._hour = _hour;}public String get_minute() {return _minute;}public void set_minute(String _minute) {this._minute = _minute;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdTrendCountHistory{private Long clickedCountHistoryLong;public Long getClickedCountHistoryLong() {return clickedCountHistoryLong;}public void setClickedCountHistoryLong(Long clickedCountHistoryLong) {this.clickedCountHistoryLong = clickedCountHistoryLong;} } 本篇文章为转载内容。原文链接：https://blog.csdn.net/tom_8899_li/article/details/71194434。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

...媒资信息 需求分析 数据模型 Dao Service 测试 0x03 Logstash：扫描课程计划媒资 创建索引 创建模板文件 配置 mysql.conf 启动 logstash.bat Logstash多实例运行 0x04 搜素服务：查询课程媒资接口 需求分析 Api接口定义 Service Controller 测试 三、在线学习：接口开发 0x01 需求分析 0x02 搭建开发环境 0x03 Api接口 0x04 服务端开发 需求分析 搜索服务注册Eureka 搜索服务客户端 自定义错误代码 Service Controller 测试 0x05 前端开发 需求分析 api方法 配置代理 视频播放页面 简单的测试 完整的测试 1、上传文件 一些问题 ~~方案1：删除本地分块文件重新尝试上传~~ 方案2：检查前端提交的MD5值是否正确 2、为课程计划选择媒资信息 3、前端门户测试 四、待完善的一些功能 😁 认识作者 一、学习页面：查询课程计划 0x01 需求分析 到目前为止，我们已可以编辑课程计划信息并上传课程视频，下一步我们要实现在线学习页面动态读取章节对应的视频并进行播放。在线学习页面所需要的信息有两类： 课程计划信息 课程学习信息（视频地址、学习进度等） 如下图： 在线学习集成媒资管理的需求如下： 1、在线学习页面显示课程计划 2、点击课程计划播放该课程计划对应的视频 本章节实现学习页面动态显示课程计划，进入不同课程的学习页面右侧动态显示当前课程的课程计划。 0x02 Api接口 课程计划信息从哪里获取？ 在课程发布完成后会自动发布到一个 course_pub 的表中，logstash 会自动将课程发布后的信息自动采集到 ES 索引库中，这些信息也包含课程计划信息。 所以考虑性能要求，课程发布后对课程的查询统一从 ES 索引库中查询。 前端通过请求 搜索服务 获取课程信息，需要单独在 搜索服务 中定义课程信息查询接口。 本接口接收课程id，查询课程所有信息返回给前端。 我们在搜素服务 API 下添加以下方法 @ApiOperation("根据id搜索课程发布信息")public Map<String,CoursePub> getdetail(String id); 返回的课程信息为 json 结构：key 为课程id，value 为课程内容。 0x03 服务端开发 在搜索服务中开发查询课程信息接口。 Controller 在搜素服务下添加以下方法 / 根据id搜索课程发布信息 @param id 课程id @return JSON数据/@Override@GetMapping("/getdetail/{id}")public Map<String, CoursePub> getdetail(@PathVariable("id")String id) {return esCourseService.getdetail(id);} Service / 根据id搜索课程发布信息 @param id 课程id @return JSON数据/public Map<String, CoursePub> getdetail(String id) {//设置索引SearchRequest searchRequest = new SearchRequest(es_index);//设置类型searchRequest.types(es_type);//创建搜索源对象SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();//设置查询条件,根据id进行查询searchSourceBuilder.query(QueryBuilders.termQuery("id",id));//这里不使用source的原字段过滤,查询所有字段// searchSourceBuilder.fetchSource(new String[]{"name", "grade", "charge","pic"}, newString[]{});//设置搜索源对象searchRequest.source(searchSourceBuilder);//执行搜索SearchResponse searchResponse = null;try {searchResponse = restHighLevelClient.search(searchRequest);} catch (IOException e) {e.printStackTrace();}//获取搜索结果SearchHits hits = searchResponse.getHits();SearchHit[] searchHits = hits.getHits(); //获取最优结果Map<String,CoursePub> map = new HashMap<>();for (SearchHit hit: searchHits) {//从搜索结果中取值并添加到coursePub对象Map<String, Object> sourceAsMap = hit.getSourceAsMap();String courseId = (String) sourceAsMap.get("id");String name = (String) sourceAsMap.get("name");String grade = (String) sourceAsMap.get("grade");String charge = (String) sourceAsMap.get("charge");String pic = (String) sourceAsMap.get("pic");String description = (String) sourceAsMap.get("description");String teachplan = (String) sourceAsMap.get("teachplan");CoursePub coursePub = new CoursePub();coursePub.setId(courseId);coursePub.setName(name);coursePub.setPic(pic);coursePub.setGrade(grade);coursePub.setTeachplan(teachplan);coursePub.setDescription(description);//设置map对象map.put(courseId,coursePub);}return map;} 测试 使用 swagger-ui 或 postman 测试查询课程信息接口。 0x04 前端开发 配置NGINX虚拟主机 学习中心的二级域名为 ucenter.xuecheng.com ，我们在 nginx 中配置 ucenter 虚拟主机。 学成网用户中心server {listen 80;server_name ucenter.xuecheng.com;个人中心location / {proxy_pass http://ucenter_server_pool;} } 前端ucenterupstream ucenter_server_pool{server 127.0.0.1:7081 weight=10;server 127.0.0.1:13000 weight=10;} 在学习中心要调用搜索的 API，使用 Nginx 解决代理，如下图： 在 ucenter 虚拟主机下配置搜索 Api 代理路径 后台搜索（公开api）upstream search_server_pool{server 127.0.0.1:40100 weight=10;} 学成网用户中心server {listen 80;server_name ucenter.xuecheng.com;个人中心location / {proxy_pass http://ucenter_server_pool;}后端搜索服务location /openapi/search/ {proxy_pass http://search_server_pool/search/;} } 前端 API 方法 在学习中心 xc-ui-pc-leanring 对课程信息的查询属于基础常用功能，所以我们将课程查询的 api 方法定义在base 模块下，如下图： 在system.js 中定义课程查询方法： import http from './public'export const course_view = id => {return http.requestGet('/openapi/search/course/getdetail/'+id);} 前端 API 方法调用 在 learning_video.vue 页面中调用课程信息查询接口得到课程计划，将课程计划json 串转成对象。 xc-ui-pc-leanring/src/module/course/page/learning_video.vue 1、定义视图 课程计划 <!--课程计划部分代码--><div class="navCont"><div class="course-weeklist"><div class="nav nav-stacked" v-for="(teachplan_first, index) in teachplanList"><div class="tit nav-justified text-center"><i class="pull-left glyphicon glyphicon-th-list"></i>{ {teachplan_first.pname} }<i class="pull-right"></i></div><li v-if="teachplan_first.children!=null" v-for="(teachplan_second, index) in teachplan_first.children"><i class="glyphicon glyphicon-check"></i><a :href="url" @click="study(teachplan_second.id)">{ {teachplan_second.pname} }</a></li><!-- <div class="tit nav-justified text-center"><i class="pull-left glyphicon glyphicon-th-list"></i>第一章<i class="pull-right"></i></div><li ><i class="glyphicon glyphicon-check"></i><a :href="url" >第一节</a></li>--><!--<li><i class="glyphicon glyphicon-unchecked"></i>为什么分为A、B、C部分</li>--></div></div></div> 课程名称 <div class="top text-center">{ {coursename} }</div> 定义数据对象 data() {return {url:'',//当前urlcourseId:'',//课程idchapter:'',//章节Idcoursename:'',//课程名称coursepic:'',//课程图片teachplanList:[],//课程计划playerOptions: {//播放参数autoplay: false,controls: true,sources: [{type: "application/x-mpegURL",src: ''}]},} } 在 created 钩子方法中获取课程信息 created(){//当前请求的urlthis.url = window.location//课程idthis.courseId = this.$route.params.courseId//章节idthis.chapter = this.$route.params.chapter//查询课程信息systemApi.course_view(this.courseId).then((view_course)=>{if(!view_course || !view_course[this.courseId]){this.$message.error("获取课程信息失败，请重新进入此页面！")return ;} let courseInfo = view_course[this.courseId]console.log(courseInfo)this.coursename = courseInfo.nameif(courseInfo.teachplan){let teachplan = JSON.parse(courseInfo.teachplan);this.teachplanList = teachplan.children;} })}, 测试 在浏览器请求：http://ucenter.xuecheng.com//learning/4028e581617f945f01617f9dabc40000/0 4028e581617f945f01617f9dabc40000：第一个参数为课程 id，测试时从 ES索引库找一个课程 id 0：第二个参数为课程计划 id，此参数用于点击课程计划播放视频。 如果出现跨域问题，但是确定已经配置了跨域，请尝试结束所以 nginx.exe 的进程 和 清空浏览器缓存。 如果还没有解决？重启电脑试试。 二、学习页面：获取视频播放地址 0x01 需求分析 用户进入在线学习页面，点击课程计划将播放该课程计划对应的教学视频。 业务流程如下： 业务流程说明： 1、用户进入在线学习页面，页面请求搜索服务获取课程信息（包括课程计划信息）并且在页面展示。 2、在线学习请求学习服务获取视频播放地址。 3、学习服务校验当前用户是否有权限学习，如果没有权限学习则提示用户。 4、学习服务校验通过，请求搜索服务获取课程媒资信息。 5、搜索服务请求ElasticSearch获取课程媒资信息。 为什么要请求 ElasticSearch 查询课程媒资信息？ 出于性能的考虑，公开查询课程信息从搜索服务查询，分摊 mysql 数据库的访问压力。 什么时候将课程媒资信息存储到 ElasticSearch 中？ 课程媒资信息是在课程发布的时候存入 ElasticSearch，因为课程发布后课程信息将基本不再修改。 0x02 课程发布：储存媒资信息 需求分析 课程媒资信息是在课程发布的时候存入 ElasticSearch 索引库，因为课程发布后课程信息将基本不再修改，具体的业务流程如下。 1、课程发布，向课程媒资信息表写入数据。 1）根据课程 id 删除 teachplanMediaPub 中的数据 2）根据课程 id 查询 teachplanMedia 数据 3）将查询到的 teachplanMedia 数据插入到 teachplanMediaPub 中 2、Logstash 定时扫描课程媒资信息表，并将课程媒资信息写入索引库。 数据模型 在 xc_course 数据库创建课程计划媒资发布表： CREATE TABLE teachplan_media_pub (teachplan_id varchar(32) NOT NULL COMMENT '课程计划id',media_id varchar(32) NOT NULL COMMENT '媒资文件id',media_fileoriginalname varchar(128) NOT NULL COMMENT '媒资文件的原始名称',media_url varchar(256) NOT NULL COMMENT '媒资文件访问地址',courseid varchar(32) NOT NULL COMMENT '课程Id',timestamp timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT'logstash使用',PRIMARY KEY (teachplan_id)) ENGINE=InnoDB DEFAULT CHARSET=utf8 数据模型类如下： package com.xuecheng.framework.domain.course;import lombok.Data;import lombok.ToString;import org.hibernate.annotations.GenericGenerator;import javax.persistence.;import java.io.Serializable;import java.util.Date;@Data@ToString@Entity@Table(name="teachplan_media_pub")@GenericGenerator(name = "jpa-assigned", strategy = "assigned")public class TeachplanMediaPub implements Serializable {private static final long serialVersionUID = -916357110051689485L;@Id@GeneratedValue(generator = "jpa-assigned")@Column(name="teachplan_id")private String teachplanId;@Column(name="media_id")private String mediaId;@Column(name="media_fileoriginalname")private String mediaFileOriginalName;@Column(name="media_url")private String mediaUrl;@Column(name="courseid")private String courseId;@Column(name="timestamp")private Date timestamp;//时间戳} Dao 创建 TeachplanMediaPub 表的 Dao，向 TeachplanMediaPub 存储信息采用先删除该课程的媒资信息，再添加该课程的媒资信息，所以这里定义根据课程 id 删除课程计划媒资方法： public interface TeachplanMediaPubRepository extends JpaRepository<TeachplanMediaPub, String> {//根据课程id删除课程计划媒资信息long deleteByCourseId(String courseId);} 从TeachplanMedia查询课程计划媒资信息 //从TeachplanMedia查询课程计划媒资信息public interface TeachplanMediaRepository extends JpaRepository<TeachplanMedia, String> {List<TeachplanMedia> findByCourseId(String courseId);} Service 编写保存课程计划媒资信息方法，并在课程发布时调用此方法。 1、保存课程计划媒资信息方法 本方法采用先删除该课程的媒资信息，再添加该课程的媒资信息，在 CourseService 下定义该方法 //保存课程计划媒资信息private void saveTeachplanMediaPub(String courseId){//查询课程媒资信息List<TeachplanMedia> byCourseId = teachplanMediaRepository.findByCourseId(courseId);if(byCourseId == null) return; //没有查询到媒资数据则直接结束该方法//将课程计划媒资信息储存到待索引表//删除原有的索引信息teachplanMediaPubRepository.deleteByCourseId(courseId);//一个课程可能会有多个媒资信息,遍历并使用list进行储存List<TeachplanMediaPub> teachplanMediaPubList = new ArrayList<>();for (TeachplanMedia teachplanMedia: byCourseId) {TeachplanMediaPub teachplanMediaPub = new TeachplanMediaPub();BeanUtils.copyProperties(teachplanMedia, teachplanMediaPub);teachplanMediaPubList.add(teachplanMediaPub);}//保存所有信息teachplanMediaPubRepository.saveAll(teachplanMediaPubList);} 2、课程发布时调用此方法 修改课程发布的 coursePublish 方法： ....//保存课程计划媒资信息到待索引表saveTeachplanMediaPub(courseId);//页面urlString pageUrl = cmsPostPageResult.getPageUrl();return new CoursePublishResult(CommonCode.SUCCESS,pageUrl);..... 测试 测试课程发布后是否成功将课程媒资信息存储到 teachplan_media_pub 中，测试流程如下： 1、指定一个课程 2、为课程计划添加课程媒资 3、执行课程发布 4、观察课程计划媒资信息是否存储至 teachplan_media_pub 中 注意：由于此测试仅用于测试发布课程计划媒资信息的功能，可暂时将 cms页面发布的功能暂时屏蔽，提高测试效率。 测试结果如下 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Vrzs5589-1595567273126)(https://qnoss.codeyee.com/20200704_15/image7)] 0x03 Logstash：扫描课程计划媒资 Logstash 定时扫描课程媒资信息表，并将课程媒资信息写入索引库。 创建索引 1、创建 xc_course_media 索引 2、并向此索引创建如下映射 POST: http://localhost:9200/xc_course_media/doc/_mapping {"properties" : {"courseid" : {"type" : "keyword"},"teachplan_id" : {"type" : "keyword"},"media_id" : {"type" : "keyword"},"media_url" : {"index" : false,"type" : "text"},"media_fileoriginalname" : {"index" : false,"type" : "text"} }} 索引创建成功 创建模板文件 在 logstach 的 config 目录文件 xc_course_media_template.json 文件路径为 %ES_ROOT_DIR%/logstash6.8.8/config/xc_course_media_template.json %ES_ROOT_DIR% 为 ElasticSearch 和 logstash 的安装目录 内容如下： {"mappings" : {"doc" : {"properties" : {"courseid" : {"type" : "keyword"},"teachplan_id" : {"type" : "keyword"},"media_id" : {"type" : "keyword"},"media_url" : {"index" : false,"type" : "text"},"media_fileoriginalname" : {"index" : false,"type" : "text"} }},"template" : "xc_course_media"} } 配置 mysql.conf 在logstash的 config 目录下配置 mysql_course_media.conf 文件供 logstash 使用，logstash 会根据 mysql_course_media.conf 文件的配置的地址从 MySQL 中读取数据向 ES 中写入索引。 参考https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html 配置输入数据源和输出数据源。 input {stdin {} jdbc {jdbc_connection_string => "jdbc:mysql://localhost:3306/xc_course?useUnicode=true&characterEncoding=utf-8&useSSL=true&serverTimezone=UTC" 数据库信息jdbc_user => "root"jdbc_password => "123123" MYSQL 驱动地址,修改为maven仓库对应的位置jdbc_driver_library => "D:/soft/apache-maven-3.5.4/repository/mysql/mysql-connector-java/5.1.40/mysql-connector-java-5.1.40.jar" the name of the driver class for mysqljdbc_driver_class => "com.mysql.jdbc.Driver"jdbc_paging_enabled => "true"jdbc_page_size => "50000"要执行的sql文件statement_filepath => "/conf/course.sql"statement => "select from teachplan_media_pub where timestamp > date_add(:sql_last_value,INTERVAL 8 HOUR)"定时配置schedule => " "record_last_run => truelast_run_metadata_path => "D:/soft/elasticsearch/logstash-6.8.8/config/xc_course_media_metadata"} } output {elasticsearch {ES的ip地址和端口hosts => "localhost:9200"hosts => ["localhost:9200","localhost:9202","localhost:9203"]ES索引库名称index => "xc_course_media"document_id => "%{teachplan_id}"document_type => "doc"template => "D:/soft/elasticsearch/logstash-6.8.8/config/xc_course_media_template.json"template_name =>"xc_course_media"template_overwrite =>"true"} stdout {日志输出codec => json_lines} } 启动 logstash.bat 启动 logstash.bat 采集 teachplan_media_pub 中的数据，向 ES 写入索引。 logstash.bat -f ../config/mysql_course_media.conf 课程发布成功后，Logstash 会自动参加 teachplan_media_pub 表中新增的数据，效果如下 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ILPBxfXi-1595567273134)(https://qnoss.codeyee.com/20200704_15/image10)] Logstash多实例运行 由于之前我们还启动了一个 Logstash 对课程的发布信息进行采集，所以如果想两个 logstash 实例同时运行，因为每个实例都有一个.lock文件，所以不能使用同一个目录来存放数据，所以我们需要使用 --path.data= 为每个实例指定单独的数据目录，具体的代码如下： 该配置是在windows下进行的 课程发布实例 logstash_start_course_pub.bat @title logstash in course_publogstash.bat -f ..\config\mysql.conf --path.data=../data/course_pub 课程计划媒体发布实例 logstash_start_teachplan_media.bat @title logstash i n teachplan_media_publogstash.bat -f ../config/mysql_course_media.conf --path.data=../data/teachplan_media/ 同时运行效果如下 0x04 搜素服务：查询课程媒资接口 需求分析 搜索服务 提供查询课程媒资接口，此接口供学习服务调用。 Api接口定义 @ApiOperation("根据课程计划查询媒资信息")public TeachplanMediaPub getmedia(String teachplanId); Service 1、配置课程计划媒资索引库等信息 在 application.yml 中配置 xuecheng:elasticsearch:hostlist: ${eshostlist:127.0.0.1:9200} 多个结点中间用逗号分隔course:index: xc_coursetype: docsource_field: id,name,grade,mt,st,charge,valid,pic,qq,price,price_old,status,studymodel,teachmode,expires,pub_time,start_time,end_timemedia:index: xc_course_mediatype: docsource_field: courseid,media_id,media_url,teachplan_id,media_fileoriginalname 2、service 方法开发 在 课程搜索服务 中定义课程媒资查询接口，为了适应后续需求，service 参数定义为数组，可一次查询多个课程计划的媒资信息。 / 根据一个或者多个课程计划id查询媒资信息 @param teachplanIds 课程id @return QueryResponseResult/public QueryResponseResult<TeachplanMediaPub> getmedia(String [] teachplanIds){//设置索引SearchRequest searchRequest = new SearchRequest(media_index);//设置类型searchRequest.types(media_type);//创建搜索源对象SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();//源字段过滤String[] media_index_arr = media_field.split(",");searchSourceBuilder.fetchSource(media_index_arr, new String[]{});//查询条件,根据课程计划id查询(可以传入多个课程计划id)searchSourceBuilder.query(QueryBuilders.termsQuery("teachplan_id", teachplanIds));searchRequest.source(searchSourceBuilder);SearchResponse searchResponse = null;try {searchResponse = restHighLevelClient.search(searchRequest);} catch (IOException e) {e.printStackTrace();}//获取结果SearchHits hits = searchResponse.getHits();long totalHits = hits.getTotalHits();SearchHit[] searchHits = hits.getHits();//数据列表List<TeachplanMediaPub> teachplanMediaPubList = new ArrayList<>();for(SearchHit hit:searchHits){TeachplanMediaPub teachplanMediaPub =new TeachplanMediaPub();Map<String, Object> sourceAsMap = hit.getSourceAsMap();//取出课程计划媒资信息String courseid = (String) sourceAsMap.get("courseid");String media_id = (String) sourceAsMap.get("media_id");String media_url = (String) sourceAsMap.get("media_url");String teachplan_id = (String) sourceAsMap.get("teachplan_id");String media_fileoriginalname = (String) sourceAsMap.get("media_fileoriginalname");teachplanMediaPub.setCourseId(courseid);teachplanMediaPub.setMediaUrl(media_url);teachplanMediaPub.setMediaFileOriginalName(media_fileoriginalname);teachplanMediaPub.setMediaId(media_id);teachplanMediaPub.setTeachplanId(teachplan_id);//将对象加入到列表中teachplanMediaPubList.add(teachplanMediaPub);}//构建返回课程媒资信息对象QueryResult<TeachplanMediaPub> queryResult = new QueryResult<>();queryResult.setList(teachplanMediaPubList);queryResult.setTotal(totalHits);return new QueryResponseResult<TeachplanMediaPub>(CommonCode.SUCCESS,queryResult);} Controller / 根据课程计划id搜索发布后的媒资信息 @param teachplanId @return/@GetMapping(value="/getmedia/{teachplanId}")@Overridepublic TeachplanMediaPub getmedia(@PathVariable("teachplanId") String teachplanId) {//为了service的拓展性,所以我们service接收的是数组作为参数,以便后续开发查询多个ID的接口String[] teachplanIds = new String[]{teachplanId};//通过service查询ES获取课程媒资信息QueryResponseResult<TeachplanMediaPub> mediaPubQueryResponseResult = esCourseService.getmedia(teachplanIds);QueryResult<TeachplanMediaPub> queryResult = mediaPubQueryResponseResult.getQueryResult();if(queryResult!=null&& queryResult.getList()!=null&& queryResult.getList().size()>0){//返回课程计划对应课程媒资return queryResult.getList().get(0);} return new TeachplanMediaPub();} 测试 使用 swagger-ui 和 postman 测试课程媒资查询接口。 三、在线学习：接口开发 0x01 需求分析 根据下边的业务流程，本章节完成前端学习页面请求学习服务获取课程视频地址，并自动播放视频。 0x02 搭建开发环境 1、创建数据库 创建 xc_learning 数据库，学习数据库将记录学生的选课信息、学习信息。 导入：资料/xc_learning.sql 2、创建学习服务工程 参考课程管理服务工程结构，创建学习服务工程： 导入：资料/xc-service-learning.zip 项目工程结构如下 0x03 Api接口 此 api 接口是课程学习页面请求学习服务获取课程学习地址。 定义返回值类型： package com.xuecheng.framework.domain.learning.response;import com.xuecheng.framework.model.response.ResponseResult;import com.xuecheng.framework.model.response.ResultCode;import lombok.Data;import lombok.NoArgsConstructor;import lombok.ToString;@Data@ToString@NoArgsConstructorpublic class GetMediaResult extends ResponseResult {public GetMediaResult(ResultCode resultCode, String fileUrl) {super(resultCode);this.fileUrl = fileUrl;}//媒资文件播放地址private String fileUrl;} 定义接口，学习服务根据传入课程 ID、章节 Id(课程计划 ID)来取学习地址。 @Api(value = "录播课程学习管理",description = "录播课程学习管理")public interface CourseLearningControllerApi {@ApiOperation("获取课程学习地址")public GetMediaResult getMediaPlayUrl(String courseId,String teachplanId);} 0x04 服务端开发 需求分析 学习服务根据传入课程ID、章节Id(课程计划ID)请求搜索服务获取学习地址。 搜索服务注册Eureka 学习服务要调用搜索服务查询课程媒资信息，所以需要将搜索服务注册到 eureka 中。 1、查看服务名称是否为 xc-service-search 注意修改application.xml中的服务名称：spring:application:name: xc‐service‐search 2、配置搜索服务的配置文件 application.yml，加入 Eureka 配置 如下： eureka:client:registerWithEureka: true 服务注册开关fetchRegistry: true 服务发现开关serviceUrl: Eureka客户端与Eureka服务端进行交互的地址，多个中间用逗号分隔defaultZone: ${EUREKA_SERVER:http://localhost:50101/eureka/,http://localhost:50102/eureka/}instance:prefer-ip-address: true 将自己的ip地址注册到Eureka服务中ip-address: ${IP_ADDRESS:127.0.0.1}instance-id: ${spring.application.name}:${server.port} 指定实例idribbon:MaxAutoRetries: 2 最大重试次数，当Eureka中可以找到服务，但是服务连不上时将会重试，如果eureka中找不到服务则直接走断路器MaxAutoRetriesNextServer: 3 切换实例的重试次数OkToRetryOnAllOperations: false 对所有操作请求都进行重试，如果是get则可以，如果是post，put等操作没有实现幂等的情况下是很危险的,所以设置为falseConnectTimeout: 5000 请求连接的超时时间ReadTimeout: 6000 请求处理的超时时间 3、添加 eureka 依赖 <dependency><groupId>org.springframework.cloud</groupId><artifactId>spring‐cloud‐starter‐netflix‐eureka‐client</artifactId></dependency> 4、修改启动类，在class上添加如下注解： @EnableDiscoveryClient 搜索服务客户端 在 学习服务 创建搜索服务的客户端接口，此接口会生成代理对象，调用搜索服务： package com.xuecheng.learning.client;import com.xuecheng.framework.domain.course.TeachplanMediaPub;import org.springframework.cloud.openfeign.FeignClient;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.PathVariable;@FeignClient(value = "xc‐service‐search")public interface CourseSearchClient {@GetMapping(value="/getmedia/{teachplanId}")public TeachplanMediaPub getmedia(@PathVariable("teachplanId") String teachplanId);} 自定义错误代码 我们在 com.xuecheng.framework.domain.learning.response 包下自定义一个错误消息模型 package com.xuecheng.framework.domain.learning.response;import com.xuecheng.framework.model.response.ResultCode;import lombok.ToString;@ToStringpublic enum LearningCode implements ResultCode {LEARNING_GET_MEDIA_ERROR(false,23001,"学习中心获取媒资信息错误！");//操作代码boolean success;//操作代码int code;//提示信息String message;private LearningCode(boolean success, int code, String message){this.success = success;this.code = code;this.message = message;}@Overridepublic boolean success() {return success;}@Overridepublic int code() {return code;}@Overridepublic String message() {return message;} } 该消息模型基于 ResultCode 来实现，代码如下 package com.xuecheng.framework.model.response;/ Created by mrt on 2018/3/5. 10000-- 通用错误代码 22000-- 媒资错误代码 23000-- 用户中心错误代码 24000-- cms错误代码 25000-- 文件系统/public interface ResultCode {//操作是否成功,true为成功，false操作失败boolean success();//操作代码int code();//提示信息String message(); 从 ResultCode 中我们可以看出，我们约定了用户中心的错误代码使用 23000，所以我们定义的一些错误信息的代码就从 23000 开始计数。 Service 在学习服务中定义 service 方法，此方法远程请求课程管理服务、媒资管理服务获取课程学习地址。 package com.xuecheng.learning.service.impl;import com.netflix.discovery.converters.Auto;import com.xuecheng.framework.domain.course.TeachplanMediaPub;import com.xuecheng.framework.domain.learning.response.GetMediaResult;import com.xuecheng.framework.exception.ExceptionCast;import com.xuecheng.framework.model.response.CommonCode;import com.xuecheng.learning.client.CourseSearchClient;import com.xuecheng.learning.service.LearningService;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Service;@Servicepublic class LearningServiceImpl implements LearningService {@AutowiredCourseSearchClient courseSearchClient;/ 远程调用搜索服务获取已发布媒体信息中的url @param courseId 课程id @param teachplanId 媒体信息id @return/@Overridepublic GetMediaResult getMediaPlayUrl(String courseId, String teachplanId) {//校验学生权限,是否已付费等//远程调用搜索服务进行查询媒体信息TeachplanMediaPub mediaPub = courseSearchClient.getmedia(teachplanId);if(mediaPub == null) ExceptionCast.cast(CommonCode.FAIL);return new GetMediaResult(CommonCode.SUCCESS, mediaPub.getMediaUrl());} } Controller 调用 service 根据课程计划 id 查询视频播放地址： @RestController@RequestMapping("/learning/course")public class CourseLearningController implements CourseLearningControllerApi {@AutowiredLearningService learningService;@Override@GetMapping("/getmedia/{courseId}/{teachplanId}")public GetMediaResult getMediaPlayUrl(@PathVariable String courseId, @PathVariable String teachplanId) {//获取课程学习地址return learningService.getMedia(courseId, teachplanId);} } 测试 使用 swagger-ui 或postman 测试学习服务查询课程视频地址接口。 0x05 前端开发 需求分析 需要在学习中心前端页面需要完成如下功能： 1、进入课程学习页面需要带上 课程 Id参数及课程计划Id的参数，其中 课程 Id 参数必带，课程计划 Id 可以为空。 2、进入页面根据 课程 Id 取出该课程的课程计划显示在右侧。 3、进入页面后判断如果请求参数中有课程计划 Id 则播放该章节的视频。 4、进入页面后判断如果 课程计划id 为0则需要取出本课程第一个 课程计划的Id，并播放第一个课程计划的视频。 进入到模块 xc-ui-pc-leanring/src/module/course api方法 let sysConfig = require('@/../config/sysConfig')let apiUrl = sysConfig.xcApiUrlPre;/获取播放地址/export const get_media = (courseId,chapter) => {return http.requestGet(apiUrl+'/api/learning/course/getmedia/'+courseId+'/'+chapter);} 配置代理 在 Nginx 中的 ucenter.xuecheng.com 虚拟主机中配置 /api/learning/ 的路径转发，此url 请转发到学习服务。 学习服务upstream learning_server_pool{server 127.0.0.1:40600 weight=10;}学成网用户中心server {listen 80;server_name ucenter.xuecheng.com;个人中心location / {proxy_pass http://ucenter_server_pool;}后端搜索服务location /openapi/search/ {proxy_pass http://search_server_pool/search/; }学习服务location ^~ /api/learning/ {proxy_pass http://learning_server_pool/learning/;} } 视频播放页面 1、如果传入的课程计划id为0则取出第一个课程计划id 在 created 钩子方法中完成 created(){//当前请求的urlthis.url = window.location//课程idthis.courseId = this.$route.params.courseId//章节idthis.chapter = this.$route.params.chapter//查询课程信息systemApi.course_view(this.courseId).then((view_course)=>{if(!view_course || !view_course[this.courseId]){this.$message.error("获取课程信息失败，请重新进入此页面！")return ;}let courseInfo = view_course[this.courseId]console.log(courseInfo)this.coursename = courseInfo.nameif(courseInfo.teachplan){console.log("准备开始播放视频")let teachplan = JSON.parse(courseInfo.teachplan);this.teachplanList = teachplan.children;//开始学习if(this.chapter == "0" || !this.chapter){//取出第一个教学计划this.chapter = this.getFirstTeachplan();console.log("第一个教学计划id为 ",this.chapter);this.study(this.chapter);}else{this.study(this.chapter);} }})}, 取出第一个章节 id，用户未输入课程计划 id 或者输入为 0 时，播放第一个。 //取出第一个章节getFirstTeachplan(){for(var i=0;i<this.teachplanList.length;i++){let firstTeachplan = this.teachplanList[i];//如果当前children存在，则取出第一个返回if(firstTeachplan.children && firstTeachplan.children.length>0){let secondTeachplan = firstTeachplan.children[0];return secondTeachplan.id;} }return ;}, 开始学习： //开始学习study(chapter){// 获取播放地址courseApi.get_media(this.courseId,chapter).then((res)=>{if(res.success){let fileUrl = sysConfig.videoUrl + res.fileUrl//播放视频this.playvideo(fileUrl)}else if(res.message){this.$message.error(res.message)}else{this.$message.error("播放视频失败，请刷新页面重试")} }).catch(res=>{this.$message.error("播放视频失败，请刷新页面重试")});}, 2、点击右侧课程章节切换播放 在原有代码基础上添加 click 事件，点击调用开始学习方法（study）。 <li v‐if="teachplan_first.children!=null" v‐for="(teachplan_second, index) inteachplan_first.children"><i class="glyphicon glyphicon‐check"></i><a :href="url" @click="study(teachplan_second.id)">{ {teachplan_second.pname} }</a></li> 3、地址栏路由url变更 这里需要注意一个问题，在用户点击课程章节切换播放时，地址栏的 url 也应该同步改变为当前所选择的课程计划 id 4、在线学习按钮 将 learnstatus 默认更改为 1，这样就能显示出马上学习的按钮，方便我们后续的集成测试。 文件路径为 xc-ui-pc-static-portal/include/course_detail_dynamic.html 部分代码块如下 <script>var body= new Vue({ //创建一个Vue的实例el: "body", //挂载点是id="app"的地方data: {editLoading: false,title:'测试',courseId:'',charge:'',//203001免费,203002收费learnstatus: 1 ,//课程状态，1：马上学习，2：立即报名、3：立即购买course:{},companyId:'template',company_stat:[],course_stat:{"s601001":"","s601002":"","s601003":""} }, 简单的测试 访问在线学习页面：http://ucenter.xuecheng.com//learning/课程id/课程计划id 通过 url 传入两个参数：课程id 和 课程计划id 如果没有课程计划则传入0 测试项目如下： 1、传入正确的课程id、课程计划id，自动播放本章节的视频 2、传入正确的课程id、课程计划id传入0，自动播放第一个视频 3、传入错误的课程id 或 课程计划id，提示错误信息。 4、通过右侧章节目录切换章节及播放视频。 访问： http://ucenter.xuecheng.com//learning/4028e58161bcf7f40161bcf8b77c0000/4028e58161bd18ea0161bd1f73190008 传入正确的课程id、课程计划id，自动播放本章节的视频 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Ef0xxym7-1595567273153)(https://qnoss.codeyee.com/20200704_15/image17)] 传入正确的课程id、课程计划id传入0，自动播放第一个视频 访问 http://ucenter.xuecheng.com//learning/4028e58161bcf7f40161bcf8b77c0000/0 识别出第一个课程计划的 id 需要注意的是这里的 chapter 参数是我自己在 study 函数里加上去的，可以忽略。 传入错误的课程id或课程计划id，提示错误信息。 通过右侧章节目录切换章节及播放视频。 点击章节即可播放，但是点击制定章节后 url 没有发生改变，这个问题暂时还没有解决，关注笔记后面的内容。 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-TOGdxwb4-1595567273158)(https://qnoss.codeyee.com/20200704_15/image20)] 完整的测试 准备工作 启动 RabbitMQ，启动 Logstash、ElasticSearch 建议把所有后端服务都开起来 启动 前端静态门户、启动 nginx 、启动课程管理前端 我们整理一下测试的流程 上传两个媒资视频文件，用于测试 进入到课程管理，为课程计划选择媒资信息 发布课程，等待 logstash 将数据采集到 ElasticSearch 的索引库中 进入学成网主页，点击课程，进入到搜索门户页面 搜索课程，进入到课程详情页面 点击开始学习，进入到课程学习页面，选择课程计划中的一个章节进行学习。 1、上传文件 首先我们使用之前开发的媒资管理模块，上传两个视频文件用于测试。 第一个文件上传成功 一些问题 在上传第二个文件时，发生了错误，我们来检查一下问题出在了哪里 在媒体服务的控制台中可以看到，在 mergeChunks 方法在校验文件 md5 时候抛出了异常 我们在 MD5 校验这里打个断点，重新上传文件，分析一下问题所在。 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-OpEMZGI8-1595567273166)(https://qnoss.codeyee.com/20200704_15/image23)] 单步调试后发现，合并文件后的MD5值与用户上传的源文件值不相等 方案1：删除本地分块文件重新尝试上传 考虑到可能是在用户上传完 视频的分块文件时发生了一些问题，导致合并文件后与源文件的大小不等，导致MD5也不相同，这里我们把这个视频上传到本地的文件全部删除，在媒资上传页面重新上传文件。 对比所有分块文件的字节大小和本地源文件的大小，完全是相等的 删除所有文件后重新上传，md5值还是不等，考虑从调试一下文件合并的代码。 方案2：检查前端提交的MD5值是否正确 在查阅是否有其他的MD5值获取方案时，发现了一个使用 windows 本地命令获取文件MD5值的方法 certutil -hashfile .\19-在线学习接口-集成测试.avi md5 惊奇的发现，TM的原来是前端那边转换的MD5值不正确，后端这边是没有问题的。 从前面的图可以看出，本地和后端转换的都是以一个 f6f0 开头的MD5值 那么问题就出现在前端了，还需要花一些时间去分析一下，这里暂时就先告一段落，因为上传了几个文件测试中只有这一个文件出现了问题。 2、为课程计划选择媒资信息 进入到一个课程的管理页面 http://localhost:12000//course/manage/baseinfo/4028e58161bcf7f40161bcf8b77c0000 将刚才我们上传的媒资文件的信息和课程计划绑定 选择效果如下 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-epKaqzCD-1595567273178)(https://qnoss.codeyee.com/20200704_15/image29)] 2、发布课程，等待 logstash 从 course_pub 以及 teachplan_media_pub 表中采集数据到 ElasticSearch 当中 发布成功后，我们可以从 teachplan_media_pub 表中看到刚才我们发布的媒资信息 再观察 Logstash 的控制台，发现两个 Logstash 的实例都对更新的课程发布信息进行了采集 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-hTUve2ik-1595567273183)(https://qnoss.codeyee.com/20200704_15/image32)] 3、前端门户测试 打开我们的门户主站 http://www.xuecheng.com/ [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4wZe9R84-1595567273185)(https://qnoss.codeyee.com/20200704_15/image33)] 点击导航栏的课程，进入到我们的搜索门户页面 如果无法进入到搜索门户，请检查你的 xc-ui-pc-portal 前端工程是否已经启动 进入到搜索门户后，可以看到一些初始化时搜索的课程数据，默认是搜索第一页的数据，每页2个课程。 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-BJ1AKoJb-1595567273187)(https://qnoss.codeyee.com/20200704_15/image34)] 我们可以测试搜索一下前面我们选择媒资信息时所用的课程 点击课程，进入到课程详情页面，然后再点击开始学习。 点击马上学习后，会进入到该课程的在线学习页面，默认自动播放我们第一个课程计划中的视频。 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-tcuLWnf2-1595567273193)(https://qnoss.codeyee.com/20200704_15/image37)] 我们可以在右侧的目录中选择第二个课程计划，会自动播放所选的课程计划所对应的媒资视频播放地址，该 播放地址正是我们刚才通过 Logstash 自动采集到 ElasticSearch 的索引信息，效果图如下 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Cvi9Dr0Y-1595567273195)(https://qnoss.codeyee.com/20200704_15/image38)] 四、待完善的一些功能 课程发布前，校验课程计划里面是否包含二级课程计划 课程发布前，校验课程计划信息里面是否全部包含媒资信息 删除媒资信息，并且同步删除ES中的索引 在获取该课程的播放地址时校验用户的合法、 在线学习页面，点击右侧目录中的课程计划同时改变url中的课程计划地址 视频文件 19-在线学习接口-集成测试.avi 前端上传时提交的MD5值不正确 😁 认识作者 作者：👦 LCyee ，全干型代码🐕 自建博客：https://www.codeyee.com 记录学习以及项目开发过程中的笔记与心得，记录认知迭代的过程，分享想法与观点。 CSDN 博客：https://blog.csdn.net/codeyee 记录和分享一些开发过程中遇到的问题以及解决的思路。 欢迎加入微服务练习生的队伍，一起交流项目学习过程中的一些问题、分享学习心得等，不定期组织一起刷题、刷项目，共同见证成长。 本篇文章为转载内容。原文链接：https://blog.csdn.net/codeyee/article/details/107558901。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

...次攻击，拿到用户隐私数据。 攻击者需要诱骗点击 反馈率低，所以较难发现和响应修复 盗取用户敏感保密信息 为了防止出现非持久型 XSS 漏洞，需要确保这么几件事情： Web 页面渲染的所有内容或者渲染的数据都必须来自于服务端。 尽量不要从 URL，document.referrer，document.forms 等这种 DOM API 中获取数据直接渲染。 尽量不要使用 eval, new Function()，document.write()，document.writeln()，window.setInterval()，window.setTimeout()，innerHTML，document.creteElement() 等可执行字符串的方法。 如果做不到以上几点，也必须对涉及 DOM 渲染的方法传入的字符串参数做 escape 转义。 前端渲染的时候对任何的字段都需要做 escape 转义编码。 escape 转义的目的是将一些构成 HTML 标签的元素转义，比如 <，>，空格 等，转义成 <，>， 等显示转义字符。有很多开源的工具可以协助我们做 escape 转义。 持久型 XSS 持久型 XSS 漏洞，也被称为存储型 XSS 漏洞，一般存在于 Form 表单提交等交互功能，如发帖留言，提交文本信息等，黑客利用的 XSS 漏洞，将内容经正常功能提交进入数据库持久保存，当前端页面获得后端从数据库中读出的注入代码时，恰好将其渲染执行。 主要注入页面方式和非持久型 XSS 漏洞类似，只不过持久型的不是来源于 URL，refferer，forms 等，而是来源于后端从数据库中读出来的数据。持久型 XSS 攻击不需要诱骗点击，黑客只需要在提交表单的地方完成注入即可，但是这种 XSS 攻击的成本相对还是很高。攻击成功需要同时满足以下几个条件： POST 请求提交表单后端没做转义直接入库。 后端从数据库中取出数据没做转义直接输出给前端。 前端拿到后端数据没做转义直接渲染成 DOM。 持久型 XSS 有以下几个特点： 持久性，植入在数据库中 危害面广，甚至可以让用户机器变成 DDoS 攻击的肉鸡。 盗取用户敏感私密信息 为了防止持久型 XSS 漏洞，需要前后端共同努力： 后端在入库前应该选择不相信任何前端数据，将所有的字段统一进行转义处理。 后端在输出给前端数据统一进行转义处理。 前端在渲染页面 DOM 的时候应该选择不相信任何后端数据，任何字段都需要做转义处理。 基于字符集的 XSS 其实现在很多的浏览器以及各种开源的库都专门针对了 XSS 进行转义处理，尽量默认抵御绝大多数 XSS 攻击，但是还是有很多方式可以绕过转义规则，让人防不胜防。比如「基于字符集的 XSS 攻击」就是绕过这些转义处理的一种攻击方式，比如有些 Web 页面字符集不固定，用户输入非期望字符集的字符，有时会绕过转义过滤规则。 以基于 utf-7 的 XSS 为例 utf-7 是可以将所有的 unicode 通过 7bit 来表示的一种字符集 (但现在已经从 Unicode 规格中移除)。 这个字符集为了通过 7bit 来表示所有的文字, 除去数字和一部分的符号,其它的部分将都以 base64 编码为基础的方式呈现。 <script>alert("xss")</script>可以被解释为：+ADw-script+AD4-alert(+ACI-xss+ACI-)+ADw-/script+AD4- 可以形成「基于字符集的 XSS 攻击」的原因是由于浏览器在 meta 没有指定 charset 的时候有自动识别编码的机制，所以这类攻击通常就是发生在没有指定或者没来得及指定 meta 标签的 charset 的情况下。 所以我们有什么办法避免这种 XSS 呢？ 记住指定 XML 中不仅要指定字符集为 utf-8，而且标签要闭合 牛文推荐：http://drops.wooyun.org/papers/1327 （这个讲的很详细） 基于 Flash 的跨站 XSS 基于 Flash 的跨站 XSS 也是属于反射型 XSS 的一种，虽然现在开发 ActionScript 的产品线几乎没有了，但还是提一句吧，AS 脚本可以接受用户输入并操作 cookie，攻击者可以配合其他 XSS（持久型或者非持久型）方法将恶意 swf 文件嵌入页面中。主要是因为 AS 有时候需要和 JS 传参交互，攻击者会通过恶意的 XSS 注入篡改参数，窃取并操作cookie。 避免方法： 严格管理 cookie 的读写权限 对 Flash 能接受用户输入的参数进行过滤 escape 转义处理 未经验证的跳转 XSS 有一些场景是后端需要对一个传进来的待跳转的 URL 参数进行一个 302 跳转，可能其中会带有一些用户的敏感（cookie）信息。如果服务器端做302 跳转，跳转的地址来自用户的输入，攻击者可以输入一个恶意的跳转地址来执行脚本。 这时候需要通过以下方式来防止这类漏洞： 对待跳转的 URL 参数做白名单或者某种规则过滤 后端注意对敏感信息的保护, 比如 cookie 使用来源验证。 CSRF CSRF（Cross-Site Request Forgery），中文名称：跨站请求伪造攻击 那么 CSRF 到底能够干嘛呢？你可以这样简单的理解：攻击者可以盗用你的登陆信息，以你的身份模拟发送各种请求。攻击者只要借助少许的社会工程学的诡计，例如通过 QQ 等聊天软件发送的链接(有些还伪装成短域名，用户无法分辨)，攻击者就能迫使 Web 应用的用户去执行攻击者预设的操作。例如，当用户登录网络银行去查看其存款余额，在他没有退出时，就点击了一个 QQ 好友发来的链接，那么该用户银行帐户中的资金就有可能被转移到攻击者指定的帐户中。 所以遇到 CSRF 攻击时，将对终端用户的数据和操作指令构成严重的威胁。当受攻击的终端用户具有管理员帐户的时候，CSRF 攻击将危及整个 Web 应用程序。 CSRF 原理 下图大概描述了 CSRF 攻击的原理，可以理解为有一个小偷在你配钥匙的地方得到了你家的钥匙，然后拿着要是去你家想偷什么偷什么。 csrf原理 完成 CSRF 攻击必须要有三个条件： 用户已经登录了站点 A，并在本地记录了 cookie 在用户没有登出站点 A 的情况下（也就是 cookie 生效的情况下），访问了恶意攻击者提供的引诱危险站点 B (B 站点要求访问站点A)。 站点 A 没有做任何 CSRF 防御 你也许会问：「如果我不满足以上三个条件中的任意一个，就不会受到 CSRF 的攻击」。其实可以这么说的，但你不能保证以下情况不会发生： 你不能保证你登录了一个网站后，不再打开一个 tab 页面并访问另外的网站，特别现在浏览器都是支持多 tab 的。 你不能保证你关闭浏览器了后，你本地的 cookie 立刻过期，你上次的会话已经结束。 上图中所谓的攻击网站 B，可能是一个存在其他漏洞的可信任的经常被人访问的网站。 预防 CSRF CSRF 的防御可以从服务端和客户端两方面着手，防御效果是从服务端着手效果比较好，现在一般的 CSRF 防御也都在服务端进行。服务端的预防 CSRF 攻击的方式方法有多种，但思路上都是差不多的，主要从以下两个方面入手： 正确使用 GET，POST 请求和 cookie 在非 GET 请求中增加 token 一般而言，普通的 Web 应用都是以 GET、POST 请求为主，还有一种请求是 cookie 方式。我们一般都是按照如下规则设计应用的请求： GET 请求常用在查看，列举，展示等不需要改变资源属性的时候（数据库 query 查询的时候） POST 请求常用在 From 表单提交，改变一个资源的属性或者做其他一些事情的时候（数据库有 insert、update、delete 的时候） 当正确的使用了 GET 和 POST 请求之后，剩下的就是在非 GET 方式的请求中增加随机数，这个大概有三种方式来进行： 为每个用户生成一个唯一的 cookie token，所有表单都包含同一个伪随机值，这种方案最简单，因为攻击者不能获得第三方的 cookie(理论上)，所以表单中的数据也就构造失败，但是由于用户的 cookie 很容易由于网站的 XSS 漏洞而被盗取，所以这个方案必须要在没有 XSS 的情况下才安全。 每个 POST 请求使用验证码，这个方案算是比较完美的，但是需要用户多次输入验证码，用户体验比较差，所以不适合在业务中大量运用。 渲染表单的时候，为每一个表单包含一个 csrfToken，提交表单的时候，带上 csrfToken，然后在后端做 csrfToken 验证。 CSRF 的防御可以根据应用场景的不同自行选择。CSRF 的防御工作确实会在正常业务逻辑的基础上带来很多额外的开发量，但是这种工作量是值得的，毕竟用户隐私以及财产安全是产品最基础的根本。 SQL 注入 SQL 注入漏洞（SQL Injection）是 Web 开发中最常见的一种安全漏洞。可以用它来从数据库获取敏感信息，或者利用数据库的特性执行添加用户，导出文件等一系列恶意操作，甚至有可能获取数据库乃至系统用户最高权限。 而造成 SQL 注入的原因是因为程序没有有效的转义过滤用户的输入，使攻击者成功的向服务器提交恶意的 SQL 查询代码，程序在接收后错误的将攻击者的输入作为查询语句的一部分执行，导致原始的查询逻辑被改变，额外的执行了攻击者精心构造的恶意代码。 很多 Web 开发者没有意识到 SQL 查询是可以被篡改的，从而把 SQL 查询当作可信任的命令。殊不知，SQL 查询是可以绕开访问控制，从而绕过身份验证和权限检查的。更有甚者，有可能通过 SQL 查询去运行主机系统级的命令。 SQL 注入原理 下面将通过一些真实的例子来详细讲解 SQL 注入的方式的原理。 考虑以下简单的管理员登录表单： <form action="/login" method="POST"><p>Username: <input type="text" name="username" /></p><p>Password: <input type="password" name="password" /></p><p><input type="submit" value="登陆" /></p></form> 后端的 SQL 语句可能是如下这样的： let querySQL = SELECT FROM userWHERE username='${username}'AND psw='${password}'; // 接下来就是执行 sql 语句… 目的就是来验证用户名和密码是不是正确，按理说乍一看上面的 SQL 语句也没什么毛病，确实是能够达到我们的目的，可是你只是站在用户会老老实实按照你的设计来输入的角度来看问题，如果有一个恶意攻击者输入的用户名是 zoumiaojiang’ OR 1 = 1 --，密码随意输入，就可以直接登入系统了。WFT! 冷静下来思考一下，我们之前预想的真实 SQL 语句是: SELECT FROM user WHERE username='zoumiaojiang' AND psw='mypassword' 可以恶意攻击者的奇怪用户名将你的 SQL 语句变成了如下形式： SELECT FROM user WHERE username='zoumiaojiang' OR 1 = 1 --' AND psw='xxxx' 在 SQL 中，-- 是注释后面的内容的意思，所以查询语句就变成了： SELECT FROM user WHERE username='zoumiaojiang' OR 1 = 1 这条 SQL 语句的查询条件永远为真，所以意思就是恶意攻击者不用我的密码，就可以登录进我的账号，然后可以在里面为所欲为，然而这还只是最简单的注入，牛逼的 SQL 注入高手甚至可以通过 SQL 查询去运行主机系统级的命令，将你主机里的内容一览无余，这里我也没有这个能力讲解的太深入，毕竟不是专业研究这类攻击的，但是通过以上的例子，已经了解了 SQL 注入的原理，我们基本已经能找到防御 SQL 注入的方案了。 如何预防 SQL 注入 防止 SQL 注入主要是不能允许用户输入的内容影响正常的 SQL 语句的逻辑，当用户的输入的信息将要用来拼接 SQL 语句的话，我们应该永远选择不相信，任何内容都必须进行转义过滤，当然做到这个还是不够的，下面列出防御 SQL 注入的几点注意事项： 严格限制Web应用的数据库的操作权限，给此用户提供仅仅能够满足其工作的最低权限，从而最大限度的减少注入攻击对数据库的危害 后端代码检查输入的数据是否符合预期，严格限制变量的类型，例如使用正则表达式进行一些匹配处理。 对进入数据库的特殊字符（’，"，\，<，>，&，，; 等）进行转义处理，或编码转换。基本上所有的后端语言都有对字符串进行转义处理的方法，比如 lodash 的 lodash._escapehtmlchar 库。 所有的查询语句建议使用数据库提供的参数化查询接口，参数化的语句使用参数而不是将用户输入变量嵌入到 SQL 语句中，即不要直接拼接 SQL 语句。例如 Node.js 中的 mysqljs 库的 query 方法中的 ? 占位参数。 mysql.query(SELECT FROM user WHERE username = ? AND psw = ?, [username, psw]); 在应用发布之前建议使用专业的 SQL 注入检测工具进行检测，以及时修补被发现的 SQL 注入漏洞。网上有很多这方面的开源工具，例如 sqlmap、SQLninja 等。 避免网站打印出 SQL 错误信息，比如类型错误、字段不匹配等，把代码里的 SQL 语句暴露出来，以防止攻击者利用这些错误信息进行 SQL 注入。 不要过于细化返回的错误信息，如果目的是方便调试，就去使用后端日志，不要在接口上过多的暴露出错信息，毕竟真正的用户不关心太多的技术细节，只要话术合理就行。 碰到要操作的数据库的代码，一定要慎重，小心使得万年船，多找几个人多来几次 code review，将问题都暴露出来，而且要善于利用工具，操作数据库相关的代码属于机密，没事不要去各种论坛晒自家站点的 SQL 语句，万一被人盯上了呢？ 命令行注入 命令行注入漏洞，指的是攻击者能够通过 HTTP 请求直接侵入主机，执行攻击者预设的 shell 命令，听起来好像匪夷所思，这往往是 Web 开发者最容易忽视但是却是最危险的一个漏洞之一，看一个实例： 假如现在需要实现一个需求：用户提交一些内容到服务器，然后在服务器执行一些系统命令去产出一个结果返回给用户，接口的部分实现如下： // 以 Node.js 为例，假如在接口中需要从 github 下载用户指定的 repoconst exec = require('mz/child_process').exec;let params = {/ 用户输入的参数 /};exec(git clone ${params.repo} /some/path); 这段代码确实能够满足业务需求，正常的用户也确实能从指定的 git repo 上下载到想要的代码，可是和 SQL 注入一样，这段代码在恶意攻击者眼中，简直就是香饽饽。 如果 params.repo 传入的是 https://github.com/zoumiaojiang/zoumiaojiang.github.io.git 当然没问题了。 可是如果 params.repo 传入的是 https://github.com/xx/xx.git && rm -rf / && 恰好你的服务是用 root 权限起的就惨了。 具体恶意攻击者能用命令行注入干什么也像 SQL 注入一样，手法是千变万化的，比如「反弹 shell 注入」等，但原理都是一样的，我们绝对有能力防止命令行注入发生。防止命令行注入需要做到以下几件事情： 后端对前端提交内容需要完全选择不相信，并且对其进行规则限制（比如正则表达式）。 在调用系统命令前对所有传入参数进行命令行参数转义过滤。 不要直接拼接命令语句，借助一些工具做拼接、转义预处理，例如 Node.js 的 shell-escape npm 包。 还是前面的例子，我们可以做到如下： const exec = require('mz/child_process').exec;// 借助 shell-escape npm 包解决参数转义过滤问题const shellescape = require('shell-escape');let params = {/ 用户输入的参数 /};// 先过滤一下参数，让参数符合预期if (!/正确的表达式/.test(params.repo)) {return;}let cmd = shellescape(['git','clone',params.repo,'/some/path']);// cmd 的值: git clone 'https://github.com/xx/xx.git && rm -rf / &&' /some/path// 这样就不会被注入成功了。exec(cmd); DDoS 攻击 DDoS 又叫分布式拒绝服务，全称 Distributed Denial of Service，其原理就是利用大量的请求造成资源过载，导致服务不可用，这个攻击应该不能算是安全问题，这应该算是一个另类的存在，因为这种攻击根本就是耍流氓的存在，「伤敌一千，自损八百」的行为。出于保护 Web App 不受攻击的攻防角度，还是介绍一下 DDoS 攻击吧，毕竟也是挺常见的。 DDoS 攻击可以理解为：「你开了一家店，隔壁家点看不惯，就雇了一大堆黑社会人员进你店里干坐着，也不消费，其他客人也进不来，导致你营业惨淡」。为啥说 DDoS 是个「伤敌一千，自损八百」的行为呢？毕竟隔壁店还是花了不少钱雇黑社会但是啥也没得到不是？DDoS 攻击的目的基本上就以下几个： 深仇大恨，就是要干死你 敲诈你，不给钱就干你 忽悠你，不买我防火墙服务就会有“人”继续干你 也许你的站点遭受过 DDoS 攻击，具体什么原因怎么解读见仁见智。DDos 攻击从层次上可分为网络层攻击与应用层攻击，从攻击手法上可分为快型流量攻击与慢型流量攻击，但其原理都是造成资源过载，导致服务不可用。 网络层 DDoS 网络层 DDos 攻击包括 SYN Flood、ACK Flood、UDP Flood、ICMP Flood 等。 SYN Flood 攻击 SYN flood 攻击主要利用了 TCP 三次握手过程中的 Bug，我们都知道 TCP 三次握手过程是要建立连接的双方发送 SYN，SYN + ACK，ACK 数据包，而当攻击方随意构造源 IP 去发送 SYN 包时，服务器返回的 SYN + ACK 就不能得到应答（因为 IP 是随意构造的），此时服务器就会尝试重新发送，并且会有至少 30s 的等待时间，导致资源饱和服务不可用，此攻击属于慢型 DDoS 攻击。 ACK Flood 攻击 ACK Flood 攻击是在 TCP 连接建立之后，所有的数据传输 TCP 报文都是带有 ACK 标志位的，主机在接收到一个带有 ACK 标志位的数据包的时候，需要检查该数据包所表示的连接四元组是否存在，如果存在则检查该数据包所表示的状态是否合法，然后再向应用层传递该数据包。如果在检查中发现该数据包不合法，例如该数据包所指向的目的端口在本机并未开放，则主机操作系统协议栈会回应 RST 包告诉对方此端口不存在。 UDP Flood 攻击 UDP flood 攻击是由于 UDP 是一种无连接的协议，因此攻击者可以伪造大量的源 IP 地址去发送 UDP 包，此种攻击属于大流量攻击。正常应用情况下，UDP 包双向流量会基本相等，因此发起这种攻击的攻击者在消耗对方资源的时候也在消耗自己的资源。 ICMP Flood 攻击 ICMP Flood 攻击属于大流量攻击，其原理就是不断发送不正常的 ICMP 包（所谓不正常就是 ICMP 包内容很大），导致目标带宽被占用，但其本身资源也会被消耗。目前很多服务器都是禁 ping 的（在防火墙在可以屏蔽 ICMP 包），因此这种攻击方式已经落伍。 网络层 DDoS 防御 网络层的 DDoS 攻击究其本质其实是无法防御的，我们能做得就是不断优化服务本身部署的网络架构，以及提升网络带宽。当然，还是做好以下几件事也是有助于缓解网络层 DDoS 攻击的冲击： 网络架构上做好优化，采用负载均衡分流。 确保服务器的系统文件是最新的版本，并及时更新系统补丁。 添加抗 DDos 设备，进行流量清洗。 限制同时打开的 SYN 半连接数目，缩短 SYN 半连接的 Timeout 时间。 限制单 IP 请求频率。 防火墙等防护设置禁止 ICMP 包等。 严格限制对外开放的服务器的向外访问。 运行端口映射程序或端口扫描程序，要认真检查特权端口和非特权端口。 关闭不必要的服务。 认真检查网络设备和主机/服务器系统的日志。只要日志出现漏洞或是时间变更,那这台机器就可能遭到了攻击。 限制在防火墙外与网络文件共享。这样会给黑客截取系统文件的机会，主机的信息暴露给黑客，无疑是给了对方入侵的机会。 加钱堆机器。。 报警。。 应用层 DDoS 应用层 DDoS 攻击不是发生在网络层，是发生在 TCP 建立握手成功之后，应用程序处理请求的时候，现在很多常见的 DDoS 攻击都是应用层攻击。应用层攻击千变万化，目的就是在网络应用层耗尽你的带宽，下面列出集中典型的攻击类型。 CC 攻击 当时绿盟为了防御 DDoS 攻击研发了一款叫做 Collapasar 的产品，能够有效的防御 SYN Flood 攻击。黑客为了挑衅，研发了一款 Challenge Collapasar 攻击工具（简称 CC）。 CC 攻击的原理，就是针对消耗资源比较大的页面不断发起不正常的请求，导致资源耗尽。因此在发送 CC 攻击前，我们需要寻找加载比较慢，消耗资源比较多的网页，比如需要查询数据库的页面、读写硬盘文件的等。通过 CC 攻击，使用爬虫对某些加载需要消耗大量资源的页面发起 HTTP 请求。 DNS Flood DNS Flood 攻击采用的方法是向被攻击的服务器发送大量的域名解析请求，通常请求解析的域名是随机生成或者是网络世界上根本不存在的域名，被攻击的DNS 服务器在接收到域名解析请求的时候首先会在服务器上查找是否有对应的缓存，如果查找不到并且该域名无法直接由服务器解析的时候，DNS 服务器会向其上层 DNS 服务器递归查询域名信息。域名解析的过程给服务器带来了很大的负载，每秒钟域名解析请求超过一定的数量就会造成 DNS 服务器解析域名超时。 根据微软的统计数据，一台 DNS 服务器所能承受的动态域名查询的上限是每秒钟 9000 个请求。而我们知道，在一台 P3 的 PC 机上可以轻易地构造出每秒钟几万个域名解析请求，足以使一台硬件配置极高的 DNS 服务器瘫痪，由此可见 DNS 服务器的脆弱性。 HTTP 慢速连接攻击 针对 HTTP 协议，先建立起 HTTP 连接，设置一个较大的 Conetnt-Length，每次只发送很少的字节，让服务器一直以为 HTTP 头部没有传输完成，这样连接一多就很快会出现连接耗尽。 应用层 DDoS 防御 判断 User-Agent 字段（不可靠，因为可以随意构造） 针对 IP + cookie，限制访问频率（由于 cookie 可以更改，IP 可以使用代理，或者肉鸡，也不可靠) 关闭服务器最大连接数等，合理配置中间件，缓解 DDoS 攻击。 请求中添加验证码，比如请求中有数据库操作的时候。 编写代码时，尽量实现优化，并合理使用缓存技术，减少数据库的读取操作。 加钱堆机器。。 报警。。 应用层的防御有时比网络层的更难，因为导致应用层被 DDoS 攻击的因素非常多，有时往往是因为程序员的失误，导致某个页面加载需要消耗大量资源，有时是因为中间件配置不当等等。而应用层 DDoS 防御的核心就是区分人与机器（爬虫），因为大量的请求不可能是人为的，肯定是机器构造的。因此如果能有效的区分人与爬虫行为，则可以很好地防御此攻击。 其他 DDoS 攻击 发起 DDoS 也是需要大量的带宽资源的，但是互联网就像森林，林子大了什么鸟都有，DDoS 攻击者也能找到其他的方式发起廉价并且极具杀伤力的 DDoS 攻击。 利用 XSS 举个例子，如果 12306 页面有一个 XSS 持久型漏洞被恶意攻击者发现，只需在春节抢票期间在这个漏洞中执行脚本使得往某一个小站点随便发点什么请求，然后随着用户访问的增多，感染用户增多，被攻击的站点自然就会迅速瘫痪了。这种 DDoS 简直就是无本万利，不用惊讶，现在大站有 XSS 漏洞的不要太多。 来自 P2P 网络攻击 大家都知道，互联网上的 P2P 用户和流量都是一个极为庞大的数字。如果他们都去一个指定的地方下载数据，成千上万的真实 IP 地址连接过来，没有哪个设备能够支撑住。拿 BT 下载来说，伪造一些热门视频的种子，发布到搜索引擎，就足以骗到许多用户和流量了，但是这只是基础攻击。 高级的 P2P 攻击，是直接欺骗资源管理服务器。如迅雷客户端会把自己发现的资源上传到资源管理服务器，然后推送给其它需要下载相同资源的用户，这样，一个链接就发布出去。通过协议逆向，攻击者伪造出大批量的热门资源信息通过资源管理中心分发出去，瞬间就可以传遍整个 P2P 网络。更为恐怖的是，这种攻击是无法停止的，即使是攻击者自身也无法停止，攻击一直持续到 P2P 官方发现问题更新服务器且下载用户重启下载软件为止。 最后总结下，DDoS 不可能防的住，就好比你的店只能容纳 50 人，黑社会有 100 人，你就换一家大店，能容纳 500 人，然后黑社会又找来了 1000 人，这种堆人头的做法就是 DDoS 本质上的攻防之道，「道高一尺，魔高一丈，魔高一尺，道高一丈」，讲真，必要的时候就答应勒索你的人的条件吧，实在不行就报警吧。 流量劫持 流量劫持应该算是黑产行业的一大经济支柱了吧？简直是让人恶心到吐，不吐槽了，还是继续谈干货吧，流量劫持基本分两种：DNS 劫持 和 HTTP 劫持，目的都是一样的，就是当用户访问 zoumiaojiang.com 的时候，给你展示的并不是或者不完全是 zoumiaojiang.com 提供的 “内容”。 DNS 劫持 DNS 劫持，也叫做域名劫持，可以这么理解，「你打了一辆车想去商场吃饭，结果你打的车是小作坊派来的，直接给你拉到小作坊去了」，DNS 的作用是把网络地址域名对应到真实的计算机能够识别的 IP 地址，以便计算机能够进一步通信，传递网址和内容等。如果当用户通过某一个域名访问一个站点的时候，被篡改的 DNS 服务器返回的是一个恶意的钓鱼站点的 IP，用户就被劫持到了恶意钓鱼站点，然后继而会被钓鱼输入各种账号密码信息，泄漏隐私。 dns劫持 这类劫持，要不就是网络运营商搞的鬼，一般小的网络运营商与黑产勾结会劫持 DNS，要不就是电脑中毒，被恶意篡改了路由器的 DNS 配置，基本上做为开发者或站长却是很难察觉的，除非有用户反馈，现在升级版的 DNS 劫持还可以对特定用户、特定区域等使用了用户画像进行筛选用户劫持的办法，另外这类广告显示更加随机更小，一般站长除非用户投诉否则很难觉察到，就算觉察到了取证举报更难。无论如何，如果接到有 DNS 劫持的反馈，一定要做好以下几件事： 取证很重要，时间、地点、IP、拨号账户、截屏、URL 地址等一定要有。 可以跟劫持区域的电信运营商进行投诉反馈。 如果投诉反馈无效，直接去工信部投诉，一般来说会加白你的域名。 HTTP 劫持 HTTP 劫持您可以这么理解，「你打了一辆车想去商场吃饭，结果司机跟你一路给你递小作坊的广告」，HTTP 劫持主要是当用户访问某个站点的时候会经过运营商网络，而不法运营商和黑产勾结能够截获 HTTP 请求返回内容，并且能够篡改内容，然后再返回给用户，从而实现劫持页面，轻则插入小广告，重则直接篡改成钓鱼网站页面骗用户隐私。能够实施流量劫持的根本原因，是 HTTP 协议没有办法对通信对方的身份进行校验以及对数据完整性进行校验。如果能解决这个问题，则流量劫持将无法轻易发生。所以防止 HTTP 劫持的方法只有将内容加密，让劫持者无法破解篡改，这样就可以防止 HTTP 劫持了。 HTTPS 协议就是一种基于 SSL 协议的安全加密网络应用层协议，可以很好的防止 HTTP 劫持。这里有篇 文章 讲的不错。HTTPS 在这就不深讲了，后面有机会我会单独好好讲讲 HTTPS。如果不想站点被 HTTP 劫持，赶紧将你的站点全站改造成 HTTPS 吧。 服务器漏洞 服务器除了以上提到的那些大名鼎鼎的漏洞和臭名昭著的攻击以外，其实还有很多其他的漏洞，往往也很容易被忽视，在这个小节也稍微介绍几种。 越权操作漏洞 如果你的系统是有登录控制的，那就要格外小心了，因为很有可能你的系统越权操作漏洞，越权操作漏洞可以简单的总结为 「A 用户能看到或者操作 B 用户的隐私内容」，如果你的系统中还有权限控制就更加需要小心了。所以每一个请求都需要做 userid 的判断 以下是一段有漏洞的后端示意代码： // ctx 为请求的 context 上下文let msgId = ctx.params.msgId;mysql.query('SELECT FROM msg_table WHERE msg_id = ?',[msgId]); 以上代码是任何人都可以查询到任何用户的消息，只要有 msg_id 就可以，这就是比较典型的越权漏洞，需要如下这么改进一下： // ctx 为请求的 context 上下文let msgId = ctx.params.msgId;let userId = ctx.session.userId; // 从会话中取出当前登陆的 userIdmysql.query('SELECT FROM msg_table WHERE msg_id = ? AND user_id = ?',[msgId, userId]); 嗯，大概就是这个意思，如果有更严格的权限控制，那在每个请求中凡是涉及到数据库的操作都需要先进行严格的验证，并且在设计数据库表的时候需要考虑进 userId 的账号关联以及权限关联。 目录遍历漏洞 目录遍历漏洞指通过在 URL 或参数中构造 …/，./ 和类似的跨父目录字符串的 ASCII 编码、unicode 编码等，完成目录跳转，读取操作系统各个目录下的敏感文件，也可以称作「任意文件读取漏洞」。 目录遍历漏洞原理：程序没有充分过滤用户输入的 …/ 之类的目录跳转符，导致用户可以通过提交目录跳转来遍历服务器上的任意文件。使用多个… 符号，不断向上跳转，最终停留在根 /，通过绝对路径去读取任意文件。 目录遍历漏洞几个示例和测试，一般构造 URL 然后使用浏览器直接访问，或者使用 Web 漏洞扫描工具检测，当然也可以自写程序测试。 http://somehost.com/../../../../../../../../../etc/passwdhttp://somehost.com/some/path?file=../../Windows/system.ini 借助 %00 空字符截断是一个比较经典的攻击手法http://somehost.com/some/path?file=../../Windows/system.ini%00.js 使用了 IIS 的脚本目录来移动目录并执行指令http://somehost.com/scripts/..%5c../Windows/System32/cmd.exe?/c+dir+c:\ 防御 方法就是需要对 URL 或者参数进行 …/，./ 等字符的转义过滤。 物理路径泄漏 物理路径泄露属于低风险等级缺陷，它的危害一般被描述为「攻击者可以利用此漏洞得到信息，来对系统进一步地攻击」，通常都是系统报错 500 的错误信息直接返回到页面可见导致的漏洞。得到物理路径有些时候它能给攻击者带来一些有用的信息，比如说：可以大致了解系统的文件目录结构；可以看出系统所使用的第三方软件；也说不定会得到一个合法的用户名（因为很多人把自己的用户名作为网站的目录名）。 防止这种泄漏的方法就是做好后端程序的出错处理，定制特殊的 500 报错页面。 源码暴露漏洞 和物理路径泄露类似，就是攻击者可以通过请求直接获取到你站点的后端源代码，然后就可以对系统进一步研究攻击。那么导致源代码暴露的原因是什么呢？基本上就是发生在服务器配置上了，服务器可以设置哪些路径的文件才可以被直接访问的，这里给一个 koa 服务起的例子，正常的 koa 服务器可以通过 koa-static 中间件去指定静态资源的目录，好让静态资源可以通过路径的路由访问。比如你的系统源代码目录是这样的： |- project|- src|- static|- ...|- server.js 你想要将 static 的文件夹配成静态资源目录，你应该会在 server.js 做如下配置： const Koa = require('koa');const serve = require('koa-static');const app = new Koa();app.use(serve(__dirname + '/project/static')); 但是如果配错了静态资源的目录，可能就出大事了，比如： // ...app.use(serve(__dirname + '/project')); 这样所有的源代码都可以通过路由访问到了，所有的服务器都提供了静态资源机制，所以在通过服务器配置静态资源目录和路径的时候，一定要注意检验，不然很可能产生漏洞。 最后，希望 Web 开发者们能够管理好自己的代码隐私，注意代码安全问题，比如不要将产品的含有敏感信息的代码放到第三方外部站点或者暴露给外部用户，尤其是前端代码，私钥类似的保密性的东西不要直接输出在代码里或者页面中。也许还有很多值得注意的点，但是归根结底还是绷住安全那根弦，对待每一行代码都要多多推敲。 请关注我的订阅号 本篇文章为转载内容。原文链接：https://blog.csdn.net/MrCoderStack/article/details/88547919。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

...PI接口设计、前后端数据交互、实时通信等多个核心环节。近期，随着前端框架Vue3和React hooks等现代技术的发展，JSON数组的动态处理与渲染效率得到了显著提升。例如，开发者可以通过Vue3的reactive特性或React Hooks中的useState和useEffect对JSON数组进行高效的状态管理，并实时反映在用户界面上。 同时，随着大数据和云计算技术的普及，JSON数组在处理大规模、非结构化数据时的作用也日益凸显。如Apache Spark等分布式计算框架已支持原生JSON数据类型，使得JSON数组能在海量数据场景下实现快速解析与处理。 此外，在安全性和隐私保护方面，业界正针对JSON数组的数据传输安全推出一系列新标准和解决方案。例如，通过JSON Web Tokens（JWT）进行身份验证时，如何安全地封装和解码包含敏感信息的JSON数组成为了研究热点。 综上所述，JSON数组在现代Web开发中不仅扮演着数据交换的关键角色，而且随着技术发展不断拓展其应用场景。从提高性能优化到强化数据安全性，JSON数组的相关实践和研究都在与时俱进，为构建高效、安全的Web应用提供有力支撑。

...现代前端开发中的表单处理实践。随着Angular框架的不断迭代更新，Angular（即Angular 2+）对表单处理提供了更为强大且灵活的支持。例如，Angular采用了响应式表单和模板驱动表单两种模式，允许开发者根据项目需求选择最适合的方案。 近期，Angular团队发布了Angular 13版本，其中对表单控件和验证机制进行了优化升级，新增了对可访问性规范的严格遵循以及更细致的错误提示，从而帮助开发者更好地处理表单提交行为异常，提高用户体验。同时，Angular Material库也同步更新了一系列UI组件，为表单设计与交互提供了丰富的、符合Material Design规范的选择。 此外，在实际项目中，如何结合最新的前端安全策略来防止XSS攻击和CSRF攻击也是表单提交时不可忽视的一环。开发者应确保在表单数据提交前后进行有效的验证与清理，并合理利用Angular提供的依赖注入和HTTP服务模块来进行安全的数据交互。 综上所述，掌握Angular（包括AngularJS及后续版本）中表单处理的最佳实践，不仅能够有效避免类似ngsubmit异常这样的问题，更能助力开发者构建出高效稳定、安全易用的现代Web应用。

...r容器中的文件管理、数据持久化以及多容器间的共享存储更为便捷和安全。 同时，随着微服务架构的广泛应用，Docker Compose工具因其对多容器应用程序定义和部署的简化而备受瞩目。通过Compose文件，开发人员可以轻松配置多个容器间的数据卷挂载策略，从而确保服务间数据的可靠传输与同步。 另外，对于数据敏感型应用，诸如数据库容器等，Docker持续优化其对存储驱动的支持，如支持本地存储、网络存储（NFS、iSCSI）以及云服务商提供的块存储服务，这无疑提升了容器环境中数据的安全性和可用性。 此外，业界也在积极研究和发展基于Docker的新型文件系统解决方案，例如结合分布式存储系统以满足大规模集群环境下容器对高性能、高可用文件读写的诉求。这些前沿技术和实践为Docker在企业级应用场景中提供了更强大的支撑，也体现了容器技术在持续演进中不断解决实际问题的决心与创新力。 总之，深入掌握Docker容器中的文件读写机制，并关注其在云原生领域的发展动态和技术革新，将有助于我们在构建现代化、可扩展的应用架构时，更好地利用Docker的优势，提升开发运维效率，保障业务系统的稳定运行。

...Python编程语言处理正数求和问题后，我们可以进一步探索其在实际应用场景中的价值。近期，数据分析领域的一项实时研究引起了广泛关注：科研人员利用Python进行大规模气象数据处理时，面临了类似的问题。由于原始数据中包含正负数值，研究人员需要快速准确地计算特定参数（如温度增量）的正向变化总和。通过借鉴文中提到的Python正数筛选与累加方法，并结合abs()函数确保结果正确性，成功实现了对复杂数据集的有效分析。 此外，在金融风控领域，Python同样扮演着关键角色。在评估投资组合收益时，分析师需要精确计算正收益部分的累积和，以排除亏损交易的影响。运用文中介绍的条件判断循环结构，结合Python强大的pandas库进行数据清洗和计算，使得复杂的财务数据分析变得更为高效且精准。 更进一步，Python内置函数的强大性和灵活性，不仅体现在abs()这样的数学运算上，还表现在众多其他场景中。例如，Python 3.9版本引入了新特性——":= walrus operator"，它可以简化if条件语句内部的赋值操作，使代码更加简洁易读。这一更新对于解决类似本文所述问题的程序编写具有重要意义，让开发者能够更好地应对实际编程挑战，提升代码质量及执行效率。 综上所述，无论是基础的正数求和问题，还是前沿的数据科学、金融分析等领域，Python以其丰富全面的功能和不断优化的语法设计，持续赋能广大开发者实现高效、准确的数据处理与业务逻辑构建。

...L是一种开源的关系型数据库管理系统，广泛应用于Web应用开发中，尤其在LAMP（Linux、Apache、MySQL、PHP/Perl/Python）架构中扮演核心角色。用户可以使用SQL语言对MySQL进行数据查询、更新、管理和控制。在本文中，MySQL是被检测和安装的目标程序，用于满足用户在工作中处理和存储数据的需求。 sudo apt-get , 这是基于Debian和Ubuntu等Linux操作系统的包管理器命令，用于自动从软件仓库获取、安装、升级或卸载软件包及其依赖项。在本文语境下，当需要在Ubuntu系统上安装MySQL时，用户会运行\ sudo apt-get update\ 来更新软件源列表信息，接着执行\ sudo apt-get install mysql-server\ 命令以下载并安装MySQL服务器。 Windows命令提示符/终端 , Windows命令提示符（对于Windows操作系统）和终端（对于macOS和Linux操作系统）是操作系统提供的命令行界面工具，允许用户通过输入文本指令与系统交互，执行各种任务，包括文件管理、系统配置以及软件安装与管理等。在本文中，用户需在命令提示符或终端中输入特定命令来检测MySQL是否已安装，以及在必要时安装MySQL。

...uery中对JSON数据的操作之外，开发者们还可以关注当下更为现代化的JavaScript库和框架对于JSON处理的最新实践。 例如，Vue.js、React和Angular等主流前端框架均提供了强大且便捷的JSON数据绑定与处理机制。Vue.js利用其响应式的数据绑定特性，可以直接将JSON对象赋值给组件的状态（state），实现视图的自动更新；React通过setState方法更新状态，并结合JSX语法实现JSON数据到UI的渲染；Angular则凭借其强大的模板表达式和变更检测系统，让JSON数据操作变得直观高效。 此外，在Node.js后端环境中，诸如Express框架支持直接将JSON传递给路由处理器，并内建了中间件来解析JSON请求体。同时，使用诸如axios或fetch这类现代HTTP客户端库，可以更加优雅地发起异步请求并处理返回的JSON数据。 近期，ECMAScript标准也在JSON支持上进行了优化，如引入JSON.stringify()的第三个参数用于定制化序列化过程，以及JSON.parse()可选的reviver函数对反序列化结果进行深度处理。这些新特性的运用能够帮助开发者更精细地控制JSON数据在程序中的流转和表现形式。 总的来说，理解并熟练掌握JSON数据处理已经成为现代全栈开发者的必备技能，持续关注相关技术和最佳实践的发展，能更好地适应快速变化的Web开发环境，提升开发效率和代码质量。

...深入了解了MySQL数据库中表基本信息的基础操作后，进一步掌握更高级的SQL查询语句和优化策略将有助于提升数据管理与分析效率。近期，MySQL 8.0版本推出了一系列新特性，如窗口函数、JSON字段支持全文检索等，使得复杂查询与大数据处理更为便捷（来源：MySQL官网，2022年更新公告）。同时，随着云服务的普及，AWS RDS for MySQL、阿里云RDS等托管数据库服务提供了自动备份、性能监控、一键扩展等功能，极大地简化了MySQL的运维工作。 此外，对于表结构设计及索引优化的理解至关重要。一篇来自DBA Stack Exchange社区的热门讨论帖（发布日期：2022年5月）深入剖析了如何根据业务场景合理设计表关系，以及何时应创建唯一索引、复合索引以提高查询性能。而一篇发表于InfoQ的技术文章《MySQL性能调优实战》则从实战角度出发，详细解读了如何通过EXPLAIN分析查询执行计划、利用慢查询日志定位瓶颈，并结合实例探讨了分区表、分库分表策略在高并发场景下的应用。 综上所述，无论是紧跟MySQL最新技术动态，还是深化对数据库内部机制和性能优化的理解，都将为您的数据库管理工作带来显著提升。持续学习并实践这些进阶知识，能够帮助您更好地应对日益增长的数据管理和分析挑战。

在易语言中处理JSON数据是现代Web开发和API交互的重要技能。随着易语言功能的不断丰富和完善，越来越多的开发者选择易语言进行项目开发，尤其是对于需要与服务器端频繁交换JSON格式数据的应用场景。近期，易语言官方发布了全新的JSON支持库，进一步优化了JSON数据的解析效率与内存占用，使得开发者能够更加便捷高效地操作JSON对象。 实际案例方面，某电商团队利用易语言开发了一款移动端管理工具，通过内建的HTTP客户端发送请求获取服务器返回的大量JSON数据，并利用易语言的JSON模块成功实现了复杂嵌套结构的数据提取与展示，极大地提高了业务处理速度与用户体验。 深入解读上，易语言对JSON的支持不仅体现在基础的读取、解析能力，更在于它如何将JSON数据映射为易语言中的数据结构，以及错误处理机制的设计。例如，当遇到无效或缺失的JSON键值时，易语言可以通过异常捕获机制确保程序稳定运行，同时提供详细的错误信息供开发者定位问题。 此外，为了帮助开发者更好地掌握JSON处理技术，易语言社区定期举办线上教程和实战训练营，邀请行业专家分享JSON在实际项目中的最佳实践，以及易语言与其他主流Web框架整合的最新方案。无论是初学者还是资深开发者，都能从中获得提升JSON数据处理能力的关键知识与技巧，紧跟时代步伐，适应日益增长的数据交换需求。

...进一步探讨MySQL数据库的管理与优化。近日，MySQL 8.0版本推出了一系列新特性，如窗口函数的增强、JSON功能的升级以及性能改进等，这为数据库管理员提供了更高效便捷的操作手段。例如，基于新的窗口函数，可以更轻松地进行复杂的数据分析和统计计算；而JSON字段类型的增强则顺应了现代应用中大量非结构化数据处理的需求。 同时，对于MySQL实例的运维管理，安全性和稳定性至关重要。定期检查并更新MySQL服务器的配置文件、确保数据目录的安全权限设置，并合理利用缓存机制以提升查询效率，是每一位数据库管理人员应熟练掌握的基本功。此外，针对线上大规模并发访问场景，深入理解并运用MySQL的InnoDB存储引擎的事务处理机制、锁机制及索引策略，有助于提升系统整体性能和用户体验。 另外，在云服务日益普及的今天，各大云服务商（如AWS RDS、阿里云RDS等）提供了托管型MySQL服务，用户无需关心底层MySQL实例的具体安装位置，即可享受到便捷的数据库创建、备份恢复及监控告警等功能。但这也要求DBA们熟悉云环境下的MySQL管理工具和服务接口，以便更好地适应云计算时代的新挑战。 总之，无论是对MySQL实例进行精细的本地部署维护，还是依托于云平台实现高效便捷的数据库管理，都需要不断跟进MySQL技术的发展动态，深入理解其核心原理，并结合实际业务场景灵活运用各种优化策略，从而确保数据库系统的稳定、安全、高效运行。

...念、操作方法以及其在数据交换中的重要角色后，进一步探索相关技术领域的发展和实践应用显得尤为重要。近期，随着Web服务和API接口设计的不断优化，JSON格式的数据传输愈发普遍，其中JSON数组的高效处理成为众多开发者关注的焦点。 例如，在2022年发布的JavaScript新版标准ECMAScript 2022（ES13）中，对JSON.stringify()和JSON.parse()方法进行了性能提升和错误处理机制的增强，让开发者在处理包含大量数据或复杂嵌套结构的JSON数组时更为得心应手。同时，许多现代前端框架如React、Vue.js等也提供了与JSON数组紧密相关的高级特性，如状态管理工具Redux利用JSON序列化进行状态持久化，Vue3更是通过Composition API简化了JSON数据到组件状态的映射过程。 另外，针对不同场景下的数据类型兼容性问题，一些跨平台开发库如axios、fetch等在发起HTTP请求时，会自动处理JSON数组和其他数据类型的转换，确保前后端数据交互的无缝衔接。而在大数据处理和云计算领域，诸如AWS Lambda、Azure Functions等无服务器计算服务也广泛支持JSON数组作为输入输出参数，极大地提高了数据集成和处理的灵活性。 因此，对于任何涉及数据处理和交换的现代编程项目而言，掌握并熟练运用JSON数组不仅是一种基础技能，更是在实际开发中实现高效、稳定运行的关键要素。了解和紧跟行业发展趋势，结合前沿技术动态来深化对JSON数组的理解和实践，无疑将助力开发者不断提升工作效率和代码质量。

...题后，我们进一步关注分布式系统管理和协调的最新进展。近日，Apache ZooKeeper 3.7.0版本发布，带来了更稳定、高效的集群管理能力，包括优化的读写性能和增强的容错机制，对于Solr等依赖于ZooKeeper进行服务协调的应用来说，升级至新版本有望提升整体系统的健壮性和稳定性。 同时，Solr社区也在不断推进其与ZooKeeper集成的深度优化，例如改进连接池管理策略，减少无效的ZooKeeper会话创建，以及针对大规模集群环境下的动态节点发现与负载均衡策略的研发。这些更新使得Solr能够更好地适应云原生架构下复杂多变的部署场景，降低运维难度，并有效防止因节点失效导致的服务中断。 此外，在实际生产环境中，如何根据业务需求合理配置ZooKeeper和Solr，以实现最优性能，是每个开发者和运维人员都需要深入研究和实践的主题。建议读者可以参考《ZooKeeper实战》、《Solr权威指南》等专业书籍，结合线上教程和官方文档，了解如何在不同规模和业务场景下对这两个组件进行调优和故障排查，从而构建出既稳定又高效的搜索与数据分析平台。

在处理大数据搜索与索引构建的实际场景中，Apache Lucene的稳定性和灵活性备受开发者青睐。然而，正如文中所述，遇到NoSuchDirectoryException这样的文件系统异常时，需要对文件目录管理有精细的掌控。实际上，随着云存储和分布式系统的普及，如何在复杂环境下确保Lucene能够正确访问和管理索引目录成为了一个新的挑战。 近期，Apache Lucene在其最新版本中引入了对云存储服务如Amazon S3、Google Cloud Storage等的支持，使得开发者可以直接在云存储上创建和维护索引目录，极大地增强了其在分布式环境下的适应能力。这意味着即使本地磁盘不存在预期目录，只要配置得当，Lucene也能自动在云端创建并使用相应的目录，从而有效避免了NoSuchDirectoryException。 此外，对于更深入的文件系统交互问题，开发团队建议遵循Java的文件I/O最佳实践，例如采用try-with-resources语句确保文件资源的释放，以及适时监控文件系统状态以预防潜在的权限或空间不足等问题。同时，结合现代化运维工具进行日志分析和异常预警，能够在出现问题时快速定位并解决诸如目录缺失等故障，进一步保障基于Apache Lucene的应用服务稳定性与可靠性。

...到更复杂的场景，例如处理不同目录层级的模块导入、设置PYTHONPATH环境变量以包含自定义模块路径、以及利用importlib等内置库动态加载模块等高级技术。 近期，Python 3.9版本引入了对“位置无关模块”（即“namespace packages”）更好的支持，使得大型项目中的模块组织结构更加灵活和可扩展。这一改进让开发者能够更容易地管理跨多个目录或包的代码，并实现无缝导入。 另外，随着微服务架构和容器化部署的普及，Python的虚拟环境（如venv和conda）以及pipenv等工具在管理项目依赖和模块导入方面也发挥了关键作用。这些工具确保了各项目间模块版本的隔离，避免了因版本冲突导致的运行错误。 深入探究Python模块系统，还可参考Guido van Rossum（Python之父）在PEP 328中提出的相对导入概念，以及他在PEP 420中关于隐式namespace包的设计理念。通过研读官方文档和社区最佳实践，开发者可以更好地掌握模块导入的艺术，从而提高代码复用率和整体开发效率。 实际上，在开源社区和各大企业级项目中，模块化设计与管理已成为软件工程的基础要求之一。例如，Django、Flask等流行Web框架的核心设计理念就离不开合理的模块划分与导入机制。因此，理解和熟练运用Python模块，不仅有助于日常开发，也是提升个人技术深度和广度的重要途径。

...种现代Web应用开发架构，它将应用程序的前端界面展示逻辑与后端业务逻辑、数据处理分离开来。在前后端分离的应用中，前端（Vue.js等框架实现）主要负责用户交互和页面渲染，通过API接口向后端请求数据；后端则专注于数据处理、业务逻辑运算以及为前端提供API服务。这样做的好处在于提升开发效率、降低耦合度，并有利于团队分工协作及项目维护。 Vue-Router , Vue Router是Vue.js官方提供的路由管理库，用于构建单页面应用（SPA）。它实现了组件级别的路由导航，允许开发者根据不同的URL地址映射到不同的Vue组件，从而实现页面间的跳转和视图切换。Vue Router还提供了丰富的导航守卫钩子函数（如beforeEach），使得开发者可以在路由切换的过程中执行预加载数据、权限验证等各种操作。 Vuex , Vuex是Vue.js生态中的一款状态管理库，用于在大型应用中集中管理组件的状态和共享数据。Vuex通过定义全局状态仓库，统一管理组件内部状态的变化，并通过Action、Mutation和Getter等方式进行状态的异步更新、同步提交和获取。在Vue应用中结合Vue-Router使用时，Vuex能够确保在路由切换过程中数据的一致性和高效性，比如实现预加载功能，即在进入新路由之前预先加载并存储所需的数据至Vuex状态树中。