ElasticSearch入门篇 前言本章将介绍：ElasticSearch的作用，搭建elasticsearch的环境（Windows/Linux），ElasticSearch集群的搭建，可视化客户端插件elasticsearch-head的安装及使用，对IK分词器的安装及使用；本章介绍的ElasticSearch操作基于Restful形式（使用http请求的形式）ElasticSearch简介Elaticsearch，简称为es， es是一个开源的高扩展的分布式全文检索引擎，它可以近乎实时的存储、检索数据；本身扩展性很好，可以扩展到上百台服务器，处理PB级别的数据。es也使用Java开发并使用Lucene作为其核心来实现所有索引和搜索的功能，但是它的目的是通过简单的RESTful API来隐藏Lucene的复杂性，从而让全文搜索变得简单。ElasticSearch的使用案例2013年初，GitHub抛弃了Solr，采取ElasticSearch 来做PB级的搜索。 “GitHub使用ElasticSearch搜索20TB的数据，包括13亿文件和1300亿行代码”维基百科：启动以elasticsearch为基础的核心搜索架构SoundCloud：“SoundCloud使用ElasticSearch为1.8亿用户提供即时而精准的音乐搜索服务”百度：百度目前广泛使用ElasticSearch作为文本数据分析，采集百度所有服务器上的各类指标数据及用户自定义数据，通过对各种数据进行多维分析展示，辅助定位分析实例异常或业务层面异常。目前覆盖百度内部20多个业务线（包括casio、云分析、网盟、预测、文库、直达号、钱包、风控等），单集群最大100台机器，200个ES节点，每天导入30TB+数据新浪使用ES 分析处理32亿条实时日志阿里使用ES 构建挖财自己的日志采集和分析体系ElasticSearch与solr的对比Solr 利用 Zookeeper 进行分布式管理，而 Elasticsearch 自身带有分布式协调管理功能;Solr 支持更多格式的数据，而 Elasticsearch 仅支持json文件格式；Solr 官方提供的功能更多，而 Elasticsearch 本身更注重于核心功能，高级功能多有第三方插件提供；Solr 在传统的搜索应用中表现好于 Elasticsearch，但在处理实时搜索应用时效率明显低于 ElasticsearchElasticSearch安装windows安装下载压缩包： ElasticSearch的官方地址： https://www.elastic.co/products/elasticsearch安装 注意：es使用java开发，使用lucene作为核心，需要配置好java环境！（jdk1.8以上）类似与tomcat，直接解压即可。其目录结构如下：修改配置文件 修改 conf\jvm.option 文件#-Xmx2g修改成为： -Xms340m -Xmx340m 否则因为虚拟机内存不够无法启动修改 conf\elasticsearch.yml 文件elasticsearch-x.x.x\config\elasticsearch.yml中末尾加入： http.cors.enabled: true http.cors.allow-origin: "*" network.host: 127.0.0.1 目的是使ES支持跨域请求启动 点击 ElasticSearch 下的 bin 目录下的 elasticsearch.bat 启动，控制台显示的日志信息如下：注意：9300 是tcp通信端口，es集群之间使用tcp进行通信，9200是http协议端口。我们在浏览器可以访问：安装图形化插件 上述可以发现， ElasticSearch 不同于 Solr 自带图形化界面，我们可以通过安装ElasticSearch的head插件，完成图形化界面的效果，完成索引数据的查看。安装插件的方式有两种，在线安装和本地安装。本文档采用本地安装方式进行head插件的安装。 elasticsearch-5-x 以上版本安装head需要安装node和grunt。下载 head 插件：https://github.com/mobz/elasticsearch-head 下载压缩包后解压即可。下载 node.js：https://nodejs.org/en/download/双击安装，通过 cmd 输入 node -v 查看版本号将 grunt 安装为全局命令 ，Grunt是基于Node.js的项目构建工具在 cmd 中输入：npm install ‐g grunt‐cli由于访问的是国外的服务器，如果下载速度较慢，可以切换淘宝镜像npm install -g cnpm –registry=https://registry.npm.taobao.org后续使用的时候，只需要把npm xxx 换成 cnpm xxx 即可检测是否安装成功npm config get registry注意：后续使用时需要将 npm 替换为 cnpm。启动head 进入 head 插件目录，打开 cmd ，输入：npm install grunt server打开浏览器，输入 http://localhost:9100 即可Linux安装保证你已经安装了java环境 可以直接把windows中的copy过去，不用安装，解压即可。tar -zxf elasticsearch-6.3.2.tar.gz修改配置文件 修改 conf\jvm.option 文件#-Xmx2g修改成为： -Xms340m -Xmx340m 否则因为虚拟机内存不够无法启动修改 conf\elasticsearch.yml 文件#elasticsearch-5.6.8\config\elasticsearch.yml中末尾加入： http.cors.enabled: true http.cors.allow-origin: "*" network.host: 127.0.0.1 目的是使ES支持跨域请求启动 注意：在linux环境下不支持root用户直接启动（理由是安全问题）添加用户：[root@coderxz bin]# useradd rxz -p rongxianzhao [root@coderxz bin]# chown -R rxz:rxz /usr/local/elasticsearch/* [root@coderxz bin]# su rxz执行：#注意：切换为非root用户执行哦 [rxz@coderxz bin]$ ./elasticsearch查看运行状态：[root@coderxz ~]# ps aux|grep elasticsearch [root@coderxz ~]# curl -X GET 'http://localhost:9200'配置外网访问 9200 端口 需要开放服务器的端口修改配置文件：config/elasticsearch.yml network.host: 0.0.0.0后台启动如果你在服务器上安装Elasticsearch，而你想在本地机器上进行开发，这时候，你很可能需要在关闭终端的时候，让Elasticsearch继续保持运行。最简单的方法就是使用nohup。先按Ctrl + C，停止当前运行的Elasticsearch，改用下面的命令运行Elasticsearch。nohup ./bin/elasticsearch &ES相关概念概述（重要）Elasticsearch是面向文档(document oriented)的，这意味着它可以存储整个对象或文档(document)。然而它不仅仅是存储，还会索引(index)每个文档的内容使之可以被搜索。在Elasticsearch中，你可以对文档（而非成行成列的数据）进行索引、搜索、排序、过滤。Elasticsearch比传统关系型数据库如下：Relational DB ‐> Databases ‐> Tables ‐> Rows ‐> ColumnsElasticsearch ‐> Indices ‐> Types ‐> Documents ‐> Fields核心概念index索引 一个索引就是一个拥有几分相似特征的文档的集合。比如说，你可以有一个客户数据的索引，另一个产品目录的索引，还有一个订单数据的索引。一个索引由一个名字来标识（必须全部是小写字母的），并且当我们要对对应于这个索引中的文档进行索引、搜索、更新和删除的时候，都要使用到这个名字。在一个集群中，可以定义任意多的索引。可类比mysql中的数据库type类型 在一个索引中，你可以定义一种或多种类型。一个类型是你的索引的一个逻辑上的分类/分区，其语义完全由你来定。通常，会为具有一组共同字段的文档定义一个类型。比如说，我们假设你运营一个博客平台并且将你所有的数据存储到一个索引中。在这个索引中，你可以为用户数据定义一个类型，为博客数据定义另一个类型，当然，也可以为评论数据定义另一个类型。可类比mysql中的表Filed字段 相当于是数据表的字段，对文档数据根据不同属性进行的分类标识 。映射mapping mapping是处理数据的方式和规则方面做一些限制，如某个字段的数据类型、默认值、分析器、是否被索引等等，这些都是映射里面可以设置的，其它就是处理es里面数据的一些使用规则设置也叫做映射，按着最优规则处理数据对性能提高很大，因此才需要建立映射，并且需要思考如何建立映射才能对性能更好。相当于mysql中的创建表的过程，设置主键外键等等document文档 一个文档是一个可被索引的基础信息单元。比如，你可以拥有某一个客户的文档，某一个产品的一个文档，当然，也可以拥有某个订单的一个文档。文档以JSON（Javascript Object Notation）格式来表示，而JSON是一个到处存在的互联网数据交互格式。在一个index/type里面，你可以存储任意多的文档。注意，尽管一个文档，物理上存在于一个索引之中，文档必须被索引/赋予一个索引的type。 插入索引库以文档为单位，类比与数据库中的一行数据集群cluster 一个集群就是由一个或多个节点组织在一起，它们共同持有整个的数据，并一起提供索引和搜索功能。一个集群由 一个唯一的名字标识，这个名字默认就是“elasticsearch”。这个名字是重要的，因为一个节点只能通过指定某个集 群的名字，来加入这个集群。节点node 一个节点是集群中的一个服务器，作为集群的一部分，它存储数据，参与集群的索引和搜索功能。和集群类似，一 个节点也是由一个名字来标识的，默认情况下，这个名字是一个随机的漫威漫画角色的名字，这个名字会在启动的 时候赋予节点。这个名字对于管理工作来说挺重要的，因为在这个管理过程中，你会去确定网络中的哪些服务器对 应于Elasticsearch集群中的哪些节点。一个节点可以通过配置集群名称的方式来加入一个指定的集群。默认情况下，每个节点都会被安排加入到一个叫 做“elasticsearch”的集群中，这意味着，如果你在你的网络中启动了若干个节点，并假定它们能够相互发现彼此， 它们将会自动地形成并加入到一个叫做“elasticsearch”的集群中。在一个集群里，只要你想，可以拥有任意多个节点。而且，如果当前你的网络中没有运行任何Elasticsearch节点， 这时启动一个节点，会默认创建并加入一个叫做“elasticsearch”的集群。分片和复制 shards&replicas 一个索引可以存储超出单个结点硬件限制的大量数据。比如，一个具有10亿文档的索引占据1TB的磁盘空间，而任一节点都没有这样大的磁盘空间；或者单个节点处理搜索请求，响应太慢。为了解决这个问题，Elasticsearch提供了将索引划分成多份的能力，这些份就叫做分片。当你创建一个索引的时候，你可以指定你想要的分片的数量。每个分片本身也是一个功能完善并且独立的“索引”，这个“索引”可以被放置到集群中的任何节点上。分片很重要，主要有两方面的原因： 1）允许你水平分割/扩展你的内容容量。 2）允许你在分片（潜在地，位于多个节点上）之上进行分布式的、并行的操作，进而提高性能/吞吐量。至于一个分片怎样分布，它的文档怎样聚合回搜索请求，是完全由Elasticsearch管理的，对于作为用户的你来说，这些都是透明的。在一个网络/云的环境里，失败随时都可能发生，在某个分片/节点不知怎么的就处于离线状态，或者由于任何原因消失了，这种情况下，有一个故障转移机制是非常有用并且是强烈推荐的。为此目的，Elasticsearch允许你创建分片的一份或多份拷贝，这些拷贝叫做复制分片，或者直接叫复制。复制之所以重要，有两个主要原因： 在分片/节点失败的情况下，提供了高可用性。因为这个原因，注意到复制分片从不与原/主要（original/primary）分片置于同一节点上是非常重要的。扩展你的搜索量/吞吐量，因为搜索可以在所有的复制上并行运行。总之，每个索引可以被分成多个分片。一个索引也可以被复制0次（意思是没有复制）或多次。一旦复制了，每个索引就有了主分片（作为复制源的原来的分片）和复制分片（主分片的拷贝）之别。分片和复制的数量可以在索引创建的时候指定。在索引创建之后，你可以在任何时候动态地改变复制的数量，但是你事后不能改变分片的数量。默认情况下，Elasticsearch中的每个索引被分片5个主分片和1个复制，这意味着，如果你的集群中至少有两个节点，你的索引将会有5个主分片和另外5个复制分片（1个完全拷贝），这样的话每个索引总共就有10个分片。ElasticSearch客户端操作上述部分为理论部分，实际开发中，主要有三种方式可以作为es服务的客户端：使用elasticsearch-head插件使用elasticsearch提供的Restful接口直接访问使用elasticsearch提供的API进行访问使用Restful接口直接访问我们需要使用http请求，介绍两款接口测试工具： postman 和 Talend API tester 。Talend API tester安装： 这是一款chrome插件，无需下载。Postman安装： Postman官网：https://www.getpostman.com使用Talend API tester进行es客户端操作。Elasticsearch的接口语法curl ‐X<VERB> '<PROTOCOL>://<HOST>:<PORT>/<PATH>?<QUERY_STRING>' ‐d '<BODY>'其中：创建索引库index并添加映射mapping------PUT 请求体：article：type类型；相当于这个索引库中有张表叫做article下面定义的这张表中的字段的定义，字段默认为不索引的；analyzer：分词器使用标准分词器{ "mappings": { "article": { "properties": { "id": { "type": "long", "store": true, "index": "not_analyzed" }, "title": { "type": "text", "store": true, "index": "analyzed", "analyzer": "standard" }, "content": { "type": "text", "store": true, "index": "analyzed", "analyzer": "standard" } } } } }在可视化工具elasticsearch-head中查看：先创建索引index，再添加mapping ----PUT 我们可以在创建索引时设置mapping信息，当然也可以先创建索引然后再设置mapping。在上一个步骤中不设置maping信息，直接使用put方法创建一个索引，然后设置mapping信息。请求的url：PUT http://127.0.0.1:9200/hello2/article/_mapping请求体：{ "article": { "properties": { "id": { "type": "long", "store": true, "index": "not_analyzed" }, "title": { "type": "text", "store": true, "index": "analyzed", "analyzer": "standard" }, "content": { "type": "text", "store": true, "index": "analyzed", "analyzer": "standard" } } } }删除索引index ----DELETE 请求URL:DELETE http://127.0.0.1:9200/hello2创建文档document（向索引库中添加内容）---POST 请求URL:POST http://127.0.0.1:9200/hello/article/1请求体：{ "id": 1, "title": "ElasticSearch是一个基于Lucene的搜索服务器", "content": "它提供了一个分布式多用户能力的全文搜索引擎，基于RESTful web接口。Elasticsearch是用Java开发的，并作为Apache许可条款下的开放源码发布，是当前流行的企业级搜索引擎。设计用于云计算中，能够达到实时搜索，稳定，可靠，快速，安装使用方便。" }在elasticsearch-head中查看：注意，一般我们将_id与id赋相同值。修改document内容----POST 请求URL:POST http://127.0.0.1:9200/hello/article/1在elasticsearch-head中查看：删除文档document---DELETE 请求URL:DELETE http://127.0.0.1:9200/hello/article/2查询文档document-----GET 查询文档有三种方式：根据id查询；根据关键词查询根据输入的内容先分词，再查询i.根据id查询请求URL：GET http://127.0.0.1:9200/hello/article/1ii.根据关键字查询-term查询请求URL：POST http://127.0.0.1:9200/hello/article/_search请求体：{ "query": { "term": { "title": "搜" } } }iii.查询文档-querystring查询请求URL：POST http://127.0.0.1:9200/hello/article/_search请求体：{ "query": { "query_string": { "default_field": "title", "query": "搜索服务器" } } }指定：在哪个字段上进行查询；要查询的内容是什么；它会把查询内容先进行分词，再进行查询使用elasticsearch-head进行es客户端操作在elasticsearch-head中集成了http请求的工具，可以提供复查查询：IK分词器和Elasticsearch集成使用上述分词器使用的是标准分词器，其对中文分词不是很友好，例如对我是程序员进行分词得到：GET http://127.0.0.1:9200/_analyze?analyzer=standard&pretty=true&text=我是程序员"tokens":[ {"token": "我", "start_offset": 0, "end_offset": 1, "type": "<IDEOGRAPHIC>",…}, {"token": "是", "start_offset": 1, "end_offset": 2, "type": "<IDEOGRAPHIC>",…}, {"token": "程", "start_offset": 2, "end_offset": 3, "type": "<IDEOGRAPHIC>",…}, {"token": "序", "start_offset": 3, "end_offset": 4, "type": "<IDEOGRAPHIC>",…}, {"token": "员", "start_offset": 4, "end_offset": 5, "type": "<IDEOGRAPHIC>",…} ]我们希望达到的分词是：我、是、程序、程序员。支持中文的分词器有很多，word分词器，庖丁解牛，Ansj分词器，下面注意说IK分词器的使用。IK分词器的安装1）下载地址：https://github.com/medcl/elasticsearch-analysis-ik/releases2）解压，将解压后的elasticsearch文件夹拷贝到elasticsearch-5.6.8\plugins下，并重命名文件夹为analysis-ik （其他名字也可以，目的是不要重名）3）重新启动ElasticSearch，即可加载IK分词器IK分词器测试IK提供两种分词ik_smart和ik_max_word其中ik_smart为最少切分，ik_max_word为最细粒度划分。下面测试一下：最小切分：在浏览器输入地址：GET http://127.0.0.1:9200/_analyze?analyzer=ik_smart&pretty=true&text=我是程序员返回结果："tokens":[ {"token": "我", "start_offset": 0, "end_offset": 1, "type": "CN_CHAR",…}, {"token": "是", "start_offset": 1, "end_offset": 2, "type": "CN_CHAR",…}, {"token": "程序员", "start_offset": 2, "end_offset": 5, "type": "CN_WORD",…} ]最新切分：在浏览器输入地址：GET http://127.0.0.1:9200/_analyze?analyzer=ik_max_word&pretty=true&text=我是程序员返回结果："tokens":[ {"token": "我", "start_offset": 0, "end_offset": 1, "type": "CN_CHAR",…}, {"token": "是", "start_offset": 1, "end_offset": 2, "type": "CN_CHAR",…}, {"token": "程序员", "start_offset": 2, "end_offset": 5, "type": "CN_WORD",…}, {"token": "程序", "start_offset": 2, "end_offset": 4, "type": "CN_WORD",…}, {"token": "员", "start_offset": 4, "end_offset": 5, "type": "CN_CHAR",…} ]ElasticSearch集群ES集群是一个 P2P类型(使用 gossip 协议)的分布式系统，除了集群状态管理以外，其他所有的请求都可以发送到集群内任意一台节点上，这个节点可以自己找到需要转发给哪些节点，并且直接跟这些节点通信。所以，从网络架构及服务配置上来说，构建集群所需要的配置极其简单。在 Elasticsearch 2.0 之前，无阻碍的网络下，所有配置了相同 cluster.name 的节点都自动归属到一个集群中。2.0 版本之后，基于安全的考虑避免开发环境过于随便造成的麻烦，从 2.0 版本开始，默认的自动发现方式改为了单播(unicast)方式。配置里提供几台节点的地址，ES 将其视作gossip router 角色，借以完成集群的发现。由于这只是 ES 内一个很小的功能，所以 gossip router 角色并不需要单独配置，每个 ES 节点都可以担任。所以，采用单播方式的集群，各节点都配置相同的几个节点列表作为 router即可。集群中节点数量没有限制，一般大于等于2个节点就可以看做是集群了。一般处于高性能及高可用方面来考虑，一般集群中的节点数量都是3个及3个以上 .集群的搭建Windows 准备三台elasticsearch服务器： 修改每台服务器的配置 修改 \comf\elasticsearch.yml 配置文件：#Node节点1： http.cors.enabled: true http.cors.allow-origin: "*" #节点1的配置信息： #集群名称，保证唯一 cluster.name: my-elasticsearch #节点名称，必须不一样 node.name: node-1 #必须为本机的ip地址 network.host: 127.0.0.1 #服务端口号，在同一机器下必须不一样 http.port: 9201 #集群间通信端口号，在同一机器下必须不一样 transport.tcp.port: 9301 #设置集群自动发现机器ip集合 discovery.zen.ping.unicast.hosts: ["127.0.0.1:9301","127.0.0.1:9302","127.0.0.1:9303"] #Node节点2： http.cors.enabled: true http.cors.allow-origin: "*" #节点1的配置信息： #集群名称，保证唯一 cluster.name: my-elasticsearch #节点名称，必须不一样 node.name: node-2 #必须为本机的ip地址 network.host: 127.0.0.1 #服务端口号，在同一机器下必须不一样 http.port: 9202 #集群间通信端口号，在同一机器下必须不一样 transport.tcp.port: 9302 #设置集群自动发现机器ip集合 discovery.zen.ping.unicast.hosts: ["127.0.0.1:9301","127.0.0.1:9302","127.0.0.1:9303"] #Node节点3： http.cors.enabled: true http.cors.allow-origin: "*" #节点1的配置信息： #集群名称，保证唯一 cluster.name: my-elasticsearch #节点名称，必须不一样 node.name: node-3 #必须为本机的ip地址 network.host: 127.0.0.1 #服务端口号，在同一机器下必须不一样 http.port: 9203 #集群间通信端口号，在同一机器下必须不一样 transport.tcp.port: 9303 #设置集群自动发现机器ip集合 discovery.zen.ping.unicast.hosts: ["127.0.0.1:9301","127.0.0.1:9302","127.0.0.1:9303"]启动各个节点服务器 可以分别启动每个服务器下的 elasticsearch.bat ，我这里使用的是 windows 下的批处理文件：新建一个 elasticsearch_cluster_start.bat 文件，然后添加下面内容：格式为：start "需要启动的文件名" "文件的路径" &表示启动A后继续执行。start "elasticsearch.bat" "F:\Soft\ES-cluster\cluster01\bin\elasticsearch.bat" & start "elasticsearch.bat" "F:\Soft\ES-cluster\cluster02\bin\elasticsearch.bat" & start "elasticsearch.bat" "F:\Soft\ES-cluster\cluster03\bin\elasticsearch.bat" 关于Windows的批处理在本章就不细说了。集群测试 只要连接集群中的任意节点，其操作方式与单机版本基本相同，改变的仅仅是存储的结构。添加索引和映射PUT http://127.0.0.1:9201/hello请求体：{ "mappings": { "article": { "properties": { "id": { "type": "long", "store": true, "index": "not_analyzed" }, "title": { "type": "text", "store": true, "index": true, "analyzer": "ik_smart" }, "content": { "type": "text", "store": true, "index": true, "analyzer": "ik_smart" } } } } }返回结果：{ "acknowledged": true, "shards_acknowledged": true, "index": "hello" }添加文档POST http://127.0.0.1:9201/hello/article/1请求体：{ "id":1, "title":"ElasticSearch是一个基于Lucene的搜索服务器", "content":"它提供了一个分布式多用户能力的全文搜索引擎，基于RESTful web接口。Elasticsearch是用Java开发的，并作为Apache许可条款下的开放源码发布，是当前流行的企业级搜索引擎。设计用于云计算中，能够达到实时搜索，稳定，可靠，快速，安装使用方便。" }返回值：{ "_index": "hello", "_type": "article", "_id": "1", "_version": 1, "result": "created", "_shards":{ "total": 2, "successful": 2, "failed": 0 }, "created": true }在elasticsearch-head中查看：

超越Redis，新一代Redis Plus来了，性能炸裂！ 线程模型链接管理锁机制Active-Replica今天给大家介绍的是 KeyDB ，KeyDB项目是从redis fork出来的分支。众所周知redis是一个单线程的kv内存存储系统，而KeyDB在100%兼容redis API的情况下将redis改造成多线程。上次也跟大家说了，项目地址是：https://github.com/EQ-Alpha/KeyDB线程模型KeyDB 将 redis 原来的主线程拆分成了 主线程 和 worker线程 。每个worker线程都是io线程，负责监听端口，accept请求，读取数据和解析协议。如图所示：KeyDB使用了SO_REUSEPORT特性，多个线程可以绑定监听同个端口。每个worker线程做了cpu绑核，读取数据也使用了SO_INCOMING_CPU特性，指定cpu接收数据。解析协议之后每个线程都会去操作内存中的数据，由一把全局锁来控制多线程访问内存数据。主线程其实也是一个worker线程，包括了worker线程的工作内容，同时也包括只有主线程才可以完成的工作内容。在worker线程数组中下标为0的就是主线程。主线程的主要工作在实现 serverCron ，包括：处理统计客户端链接管理db数据的resize和reshard处理aofreplication主备同步cluster模式下的任务链接管理在redis中所有链接管理都是在一个线程中完成的。在KeyDB的设计中，每个worker线程负责一组链接，所有的链接插入到本线程的链接列表中维护。链接的产生、工作、销毁必须在同个线程中。每个链接新增一个字段int iel; /* the event loop index we're registered with */用来表示链接属于哪个线程接管。KeyDB维护了三个关键的数据结构做链接管理：clients_pending_write：线程专属的链表，维护同步给客户链接发送数据的队列clients_pending_asyncwrite：线程专属的链表，维护异步给客户链接发送数据的队列clients_to_close：全局链表，维护需要异步关闭的客户链接分成同步和异步两个队列，是因为redis有些联动api，比如pub/sub，pub之后需要给sub的客户端发送消息，pub执行的线程和sub的客户端所在线程不是同一个线程，为了处理这种情况，KeyDB将需要给非本线程的客户端发送数据维护在异步队列中。同步发送的逻辑比较简单，都是在本线程中完成，以下图来说明如何同步给客户端发送数据：如上文所提到的，一个链接的创建、接收数据、发送数据、释放链接都必须在同个线程执行。异步发送涉及到两个线程之间的交互。KeyDB通过管道在两个线程中传递消息：int fdCmdWrite; //写管道 int fdCmdRead; //读管道本地线程需要异步发送数据时，先检查client是否属于本地线程，非本地线程获取到client专属的线程ID，之后给专属的线程管到发送AE_ASYNC_OP::CreateFileEvent的操作，要求添加写socket事件。专属线程在处理管道消息时将对应的请求添加到写事件中，如图所示：redis有些关闭客户端的请求并非完全是在链接所在的线程执行关闭，所以在这里维护了一个全局的异步关闭链表。锁机制KeyDB实现了一套类似spinlock的锁机制，称之为fastlock。fastlock的主要数据结构有：int fdCmdWrite; //写管道 int fdCmdRead; //读管道使用原子操作__atomic_load_2，__atomic_fetch_add，__atomic_compare_exchange来通过比较m_active=m_avail判断是否可以获取锁。fastlock提供了两种获取锁的方式：try_lock：一次获取失败，直接返回lock：忙等，每1024 * 1024次忙等后使用sched_yield 主动交出cpu，挪到cpu的任务末尾等待执行。在KeyDB中将try_lock和事件结合起来，来避免忙等的情况发生。每个客户端有一个专属的lock，在读取客户端数据之前会先尝试加锁，如果失败，则退出，因为数据还未读取，所以在下个epoll_wait处理事件循环中可以再次处理。Active-ReplicaKeyDB实现了多活的机制，每个replica可设置成可写非只读，replica之间互相同步数据。主要特性有：每个replica有个uuid标志，用来去除环形复制新增加rreplay API，将增量命令打包成rreplay命令，带上本地的uuidkey，value加上时间戳版本号，作为冲突校验，如果本地有相同的key且时间戳版本号大于同步过来的数据，新写入失败。采用当前时间戳向左移20位，再加上后44位自增的方式来获取key的时间戳版本号。项目地址：https://github.com/EQ-Alpha/KeyDB

号称取代 Elasticsearch，太猛了！ 10倍 提升效率，号称取代 Elasticsearch 的轻量级搜索引擎到底有多强悍？# Manticore Search介绍 Manticore Search 是一个使用 C++ 开发的高性能搜索引擎，创建于 2017 年，其前身是 Sphinx Search 。Manticore Search 充分利用了 Sphinx，显着改进了它的功能，修复了数百个错误，几乎完全重写了代码并保持开源。这一切使 Manticore Search 成为一个现代，快速，轻量级和功能齐全的数据库，具有出色的全文搜索功能。Manticore Search目前在GitHub收获3.7k star，拥有大批忠实用户。同时开源者在GitHub介绍中明确说明了该项目是是Elasticsearch的良好替代品，在不久的将来就会取代ELK中的E。同时，来自 MS 官方的测试表明 Manticore Search 性能比 ElasticSearch 有质的提升：在一定的场景中，Manticore 比 Elasticsearch 快 15 倍！完整的测评结果，可以参考:https://manticoresearch.com/blog/manticore-alternative-to-elasticsearch/优势 它与其他解决方案的区别在于：它非常快，因此比其他替代方案更具成本效益。例如，Manticore:对于小型数据，比MySQL快182倍（可重现）对于日志分析，比Elasticsearch快29倍（可重现）对于小型数据集，比Elasticsearch快15倍（可重现）对于中等大小的数据，比Elasticsearch快5倍（可重现）对于大型数据，比Elasticsearch快4倍（可重现）在单个服务器上进行数据导入时，最大吞吐量比Elasticsearch快最多2倍（可重现）由于其现代的多线程架构和高效的查询并行化能力，Manticore能够充分利用所有CPU核心，以实现最快的响应时间。强大而快速的全文搜索功能能够无缝地处理小型和大型数据集。针对小、中、大型数据集提供逐行存储。对于更大的数据集，Manticore通过Manticore Columnar Library提供列存储支持，可以处理无法适合内存的数据集。自动创建高效的二级索引，节省时间和精力。成本优化的查询优化器可优化搜索查询以实现最佳性能。Manticore是基于SQL的，使用SQL作为其本机语法，并与MySQL协议兼容，使您可以使用首选的MySQL客户端。通过PHP、Python、JavaScript、Java、Elixir和Go等客户端，与Manticore Search的集成变得简单。Manticore还提供了一种编程HTTP JSON协议，用于更多样化的数据和模式管理。Manticore Search使用C++构建，启动快速，内存使用最少，低级别优化有助于其卓越性能。实时插入，新添加的文档立即可访问。提供互动课程，使学习轻松愉快。Manticore还拥有内置的复制和负载均衡功能，增加了可靠性。可以轻松地从MySQL、PostgreSQL、ODBC、xml和csv等来源同步数据。 - 虽然不完全符合ACID，但Manticore仍支持事务和binlog以确保安全写入。内置工具和SQL命令可轻松备份和恢复数据。Craigslist、Socialgist、PubChem、Rozetka和许多其他公司使用 Manticore 进行高效搜索和流过滤。# 使用 具体的安装方法：https://manticoresearch.com/install/Docker镜像可在Docker Hub上获取：https://hub.docker.com/r/manticoresearch/manticore/要在 Docker 中试验 Manticore Search，只需运行：docker run -e EXTRA=1 --name manticore --rm -d manticoresearch/manticore && until docker logs manticore 2>&1 | grep -q "accepting connections"; do sleep 1; done && docker exec -it manticore mysql && docker stop manticore 之后，可以进行其他操作，例如创建表、添加数据并运行搜索： create table movies(title text, year int) morphology='stem_en' html_strip='1' stopwords='en'; insert into movies(title, year) values ('The Seven Samurai', 1954), ('Bonnie and Clyde', 1954), ('Reservoir Dogs', 1992), ('Airplane!', 1980), ('Raging Bull', 1980), ('Groundhog Day', 1993), ('<a href="http://google.com/">Jurassic Park</a>', 1993), ('Ferris Bueller\'s Day Off', 1986); select highlight(), year from movies where match('the dog'); select highlight(), year from movies where match('days') facet year; select * from movies where match('google');完整文档和开源代码，可以移步：https://github.com/manticoresoftware/manticoresearch

小而全的第三方登录开源类库，开箱即用！ JustAuth ，如你所见，它仅仅是一个第三方授权登录的工具类库，它可以让我们脱离繁琐的第三方登录 SDK，让登录变得 So easy!JustAuth 集成了诸如：Github、Gitee、支付宝、新浪微博、微信、Google、Facebook、Twitter、StackOverflow 等国内外数十家第三方平台。功能丰富的 OAuth 平台： 集成国内外数十家第三方平台，实现快速接入。自定义 state： 支持自定义 State 和缓存方式，开发者可根据实际情况选择任意缓存插件。自定义 OAuth： 提供统一接口，支持接入任意 OAuth 网站，快速实现 OAuth 登录功能。更容易适配自有的 OAuth 服务。自定义 Http： 接口 HTTP 工具，开发者可以根据自己项目的实际情况选择相对应的 HTTP 工具。自定义 Scope： 支持自定义 scope，以适配更多的业务场景，而不仅仅是为了登录。代码规范·简单： JustAuth 代码严格遵守阿里巴巴编码规约，结构清晰、逻辑简单。快速使用（以 QQ 为例）申请开发者如果是第一次使用，需要到 “QQ 互联平台” 申请开发者，通过后创建应用并且复制三个信息：App ID、App Key和网站回调域。集成 JustAuth添加依赖<dependency> <groupId>me.zhyd.oauth</groupId> <artifactId>JustAuth</artifactId> <version>{latest-version}</version> </dependency>创建Request，把第一步的三个信息添加进去AuthRequest authRequest = new AuthQqRequest(AuthConfig.builder() .clientId("App ID") .clientSecret("App Key") .redirectUri("网站回调域") .build());生成授权地址//这个链接可以直接在后台重定向跳转，也可以返回到前端跳转 String authorizeUrl = authRequest.authorize(AuthStateUtils.createState());或者生成静态授权页面AuthRequest authRequest = AuthRequestBuilder.builder() .source("github") .authConfig(AuthConfig.builder() .clientId("clientId") .clientSecret("clientSecret") .redirectUri("redirectUri") .build()) .build(); // 生成授权页面 authRequest.authorize("state"); // 授权登录后会返回code（auth_code（仅限支付宝））、state，1.8.0版本后，可以用AuthCallback类作为回调接口的参数 // 注：JustAuth默认保存state的时效为3分钟，3分钟内未使用则会自动清除过期的state authRequest.login(callback);或者生成动态授权页面AuthRequest authRequest = AuthRequestBuilder.builder() .source("gitee") .authConfig((source) -> { // 通过 source 动态获取 AuthConfig // 此处可以灵活的从 sql 中取配置也可以从配置文件中取配置 return AuthConfig.builder() .clientId("clientId") .clientSecret("clientSecret") .redirectUri("redirectUri") .build(); }) .build(); Assert.assertTrue(authRequest instanceof AuthGiteeRequest); System.out.println(authRequest.authorize(AuthStateUtils.createState()));JustAuth 的团队还在持续接入其他平台的授权登录，感兴趣的可以关注一下。开源地址：https://github.com/justauth/JustAuth

记录windows报无法定位程序输入点于XXX动态链接库***.dll上解决办法 解决方法查阅了Windows系统cmd的一些应用后，发现了 sfc sfc 扫描所有受保护的系统文件的完整性，并使用正确的 Microsoft 版本替换不正确的版本问题解决步骤：1) 在搜索中搜索cmd，以管理员身份运行，或者Windows + R打开运行，输入cmd，Enter，进入到命令行窗口；2) 在命令行中输入sfc /SCANNOW，Enter，等待执行结束。到此就完美解决Windows报错提示：无法定位输入点于XXX链接库*.dll上了。补充：关于 “Windows 资源保护找到了损坏文件，但其中有一些文件无法修复” 的问题解决办法：按 “Windows 徽标键 + X”，启动 “Windows PowerShell（管理员）”，依次执行以下命令：Dism /Online /Cleanup-Image /ScanHealthDism /Online /Cleanup-Image /CheckHealthDISM /Online /Cleanup-image /RestoreHealthsfc /SCANNOW执行完毕后重启设备，查看问题是否解决。

JAVA 6 种服务限流方案技术选型，哪个最香？ 服务限流 ，是指通过控制请求的速率或次数来达到保护服务的目的，在微服务中，我们通常会将它和熔断、降级搭配在一起使用，来避免瞬时的大量请求对系统造成负荷，来达到保护服务平稳运行的目的。下面就来看一看常见的6种限流方式，以及它们的实现与使用。## 固定窗口算法固定窗口算法 通过在单位时间内维护一个计数器，能够限制在每个固定的时间段内请求通过的次数，以达到限流的效果。算法实现起来也比较简单，可以通过构造方法中的参数指定时间窗口大小以及允许通过的请求数量，当请求进入时先比较当前时间是否超过窗口上边界，未越界且未超过计数器上限则可以放行请求。@Slf4j public class FixedWindowRateLimiter { // 时间窗口大小，单位毫秒 private long windowSize; // 允许通过请求数 private int maxRequestCount; // 当前窗口通过的请求计数 private AtomicInteger count=new AtomicInteger(0); // 窗口右边界 private long windowBorder; public FixedWindowRateLimiter(long windowSize,int maxRequestCount){ this.windowSize = windowSize; this.maxRequestCount = maxRequestCount; windowBorder = System.currentTimeMillis()+windowSize; } public synchronized boolean tryAcquire(){ long currentTime = System.currentTimeMillis(); if (windowBorder < currentTime){ log.info("window reset"); do { windowBorder += windowSize; }while(windowBorder < currentTime); count=new AtomicInteger(0); } if (count.intValue() < maxRequestCount){ count.incrementAndGet(); log.info("tryAcquire success"); return true; }else { log.info("tryAcquire fail"); return false; } } }进行测试，允许在1000毫秒内通过5个请求：void test() throws InterruptedException { FixedWindowRateLimiter fixedWindowRateLimiter = new FixedWindowRateLimiter(1000, 5); for (int i = 0; i < 10; i++) { if (fixedWindowRateLimiter.tryAcquire()) { System.out.println("执行任务"); }else{ System.out.println("被限流"); TimeUnit.MILLISECONDS.sleep(300); } } }运行结果：固定窗口算法 的优点是实现简单，但是可能无法应对突发流量的情况，比如每秒允许放行100个请求，但是在0.9秒前都没有请求进来，这就造成了在0.9秒到1秒这段时间内要处理100个请求，而在1秒到1.1秒间可能会再进入100个请求，这就造成了要在0.2秒内处理200个请求，这种流量激增就可能导致后端服务出现异常。滑动窗口算法滑动窗口算法 在固定窗口的基础上，进行了一定的升级改造。它的算法的核心在于将时间窗口进行了更精细的分片，将固定窗口分为多个小块，每次仅滑动一小块的时间。并且在每个时间段内都维护了单独的计数器，每次滑动时，都减去前一个时间块内的请求数量，并再添加一个新的时间块到末尾，当时间窗口内所有小时间块的计数器之和超过了请求阈值时，就会触发限流操作。看一下算法的实现，核心就是通过一个int类型的数组循环使用来维护每个时间片内独立的计数器：@Slf4j public class SlidingWindowRateLimiter { // 时间窗口大小，单位毫秒 private long windowSize; // 分片窗口数 private int shardNum; // 允许通过请求数 private int maxRequestCount; // 各个窗口内请求计数 private int[] shardRequestCount; // 请求总数 private int totalCount; // 当前窗口下标 private int shardId; // 每个小窗口大小，毫秒 private long tinyWindowSize; // 窗口右边界 private long windowBorder; public SlidingWindowRateLimiter(long windowSize, int shardNum, int maxRequestCount) { this.windowSize = windowSize; this.shardNum = shardNum; this.maxRequestCount = maxRequestCount; shardRequestCount = new int[shardNum]; tinyWindowSize = windowSize/ shardNum; windowBorder=System.currentTimeMillis(); } public synchronized boolean tryAcquire() { long currentTime = System.currentTimeMillis(); if (currentTime > windowBorder){ do { shardId = (++shardId) % shardNum; totalCount -= shardRequestCount[shardId]; shardRequestCount[shardId]=0; windowBorder += tinyWindowSize; }while (windowBorder < currentTime); } if (totalCount < maxRequestCount){ log.info("tryAcquire success,{}",shardId); shardRequestCount[shardId]++; totalCount++; return true; }else{ log.info("tryAcquire fail,{}",shardId); return false; } } }进行一下测试，对第一个例子中的规则进行修改，每1秒允许100个请求通过不变，在此基础上再把每1秒等分为10个0.1秒的窗口。void test() throws InterruptedException { SlidingWindowRateLimiter slidingWindowRateLimiter = new SlidingWindowRateLimiter(1000, 10, 10); TimeUnit.MILLISECONDS.sleep(800); for (int i = 0; i < 15; i++) { boolean acquire = slidingWindowRateLimiter.tryAcquire(); if (acquire){ System.out.println("执行任务"); }else{ System.out.println("被限流"); } TimeUnit.MILLISECONDS.sleep(10); } }查看运行结果：程序启动后，在先休眠了一段时间后再发起请求，可以看到在0.9秒到1秒的时间窗口内放行了6个请求，在1秒到1.1秒内放行了4个请求，随后就进行了限流，解决了在固定窗口算法中相邻时间窗口内允许通过大量请求的问题。滑动窗口算法通过将时间片进行分片，对流量的控制更加精细化，但是相应的也会浪费一些存储空间，用来维护每一块时间内的单独计数，并且还没有解决固定窗口中可能出现的流量激增问题。漏桶算法为了应对流量激增的问题，后续又衍生出了漏桶算法，用专业一点的词来说，漏桶算法能够进行流量整形和流量控制。漏桶是一个很形象的比喻，外部请求就像是水一样不断注入水桶中，而水桶已经设置好了最大出水速率，漏桶会以这个速率匀速放行请求，而当水超过桶的最大容量后则被丢弃。看一下代码实现：@Slf4j public class LeakyBucketRateLimiter { // 桶的容量 private int capacity; // 桶中现存水量 private AtomicInteger water=new AtomicInteger(0); // 开始漏水时间 private long leakTimeStamp; // 水流出的速率，即每秒允许通过的请求数 private int leakRate; public LeakyBucketRateLimiter(int capacity,int leakRate){ this.capacity=capacity; this.leakRate=leakRate; } public synchronized boolean tryAcquire(){ // 桶中没有水，重新开始计算 if (water.get()==0){ log.info("start leaking"); leakTimeStamp = System.currentTimeMillis(); water.incrementAndGet(); return water.get() < capacity; } // 先漏水，计算剩余水量 long currentTime = System.currentTimeMillis(); int leakedWater= (int) ((currentTime-leakTimeStamp)/1000 * leakRate); log.info("lastTime:{}, currentTime:{}. LeakedWater:{}",leakTimeStamp,currentTime,leakedWater); // 可能时间不足,则先不漏水 if (leakedWater != 0){ int leftWater = water.get() - leakedWater; // 可能水已漏光，设为0 water.set(Math.max(0,leftWater)); leakTimeStamp=System.currentTimeMillis(); } log.info("剩余容量:{}",capacity-water.get()); if (water.get() < capacity){ log.info("tryAcquire success"); water.incrementAndGet(); return true; }else { log.info("tryAcquire fail"); return false; } } }进行一下测试，先初始化一个漏桶，设置桶的容量为3，每秒放行1个请求，在代码中每500毫秒尝试请求1次：void test() throws InterruptedException { LeakyBucketRateLimiter leakyBucketRateLimiter =new LeakyBucketRateLimiter(3,1); for (int i = 0; i < 15; i++) { if (leakyBucketRateLimiter.tryAcquire()) { System.out.println("执行任务"); }else { System.out.println("被限流"); } TimeUnit.MILLISECONDS.sleep(500); } }查看运行结果，按规则进行了放行：但是，漏桶算法同样也有缺点，不管当前系统的负载压力如何，所有请求都得进行排队，即使此时服务器的负载处于相对空闲的状态，这样会造成系统资源的浪费。由于漏桶的缺陷比较明显，所以在实际业务场景中，使用的比较少。令牌桶算法令牌桶算法 是基于 漏桶算法 的一种改进，主要在于令牌桶算法能够在限制服务调用的平均速率的同时，还能够允许一定程度内的突发调用。它的主要思想是系统以 恒定的 速度生成令牌，并将令牌放入令牌桶中，当令牌桶中满了的时候，再向其中放入的令牌就会被丢弃。而每次请求进入时，必须从令牌桶中获取一个令牌，如果没有获取到令牌则被限流拒绝。假设令牌的生成速度是每秒100个，并且第一秒内只使用了70个令牌，那么在第二秒可用的令牌数量就变成了130，在允许的请求范围上限内，扩大了请求的速率。当然，这里要设置桶容量的上限，避免超出系统能够承载的最大请求数量。Guava 中的 RateLimiter 就是基于令牌桶实现的，可以直接拿来使用，先引入依赖：<dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>29.0-jre</version> </dependency>进行测试，设置 每秒产生5个令牌 ：void acquireTest(){ RateLimiter rateLimiter=RateLimiter.create(5); for (int i = 0; i < 10; i++) { double time = rateLimiter.acquire(); log.info("等待时间：{}s",time); } }运行结果：可以看到，每 200ms 左右产生一个令牌并放行请求，也就是1秒放行 5 个请求，使用 RateLimiter 能够很好的实现单机的限流。那么再回到我们前面提到的突发流量情况，令牌桶是怎么解决的呢？RateLimiter中引入了一个预消费的概念。在源码中，有这么一段注释： * <p>It is important to note that the number of permits requested <i>never</i> affects the * throttling of the request itself (an invocation to {@code acquire(1)} and an invocation to {@code * acquire(1000)} will result in exactly the same throttling, if any), but it affects the throttling * of the <i>next</i> request. I.e., if an expensive task arrives at an idle RateLimiter, it will be * granted immediately, but it is the <i>next</i> request that will experience extra throttling, * thus paying for the cost of the expensive task.大意就是，申请令牌的数量不同不会影响这个申请令牌这个动作本身的响应时间， acquire(1) 和 acquire(1000) 这两个请求会消耗同样的时间返回结果，但是会影响下一个请求的响应时间。如果一个消耗大量令牌的任务到达空闲的 RateLimiter ，会被立即批准执行，但是当下一个请求进来时，将会额外等待一段时间，用来支付前一个请求的时间成本。至于为什么要这么做，通过举例来引申一下。当一个系统处于空闲状态时，突然来了1个需要消耗100个令牌的任务，那么白白等待100秒是毫无意义的浪费资源行为，那么可以先允许它执行，并对后续请求进行限流时间上的延长，以此来达到一个应对突发流量的效果。看一下具体的代码示例：void acquireMultiTest(){ RateLimiter rateLimiter=RateLimiter.create(1); for (int i = 0; i <3; i++) { int num = 2 * i + 1; log.info("获取{}个令牌", num); double cost = rateLimiter.acquire(num); log.info("获取{}个令牌结束，耗时{}ms",num,cost); } }运行结果：可以看到，在第二次请求时需要 3 个令牌，但是并没有等 3 秒后才获取成功，而是在等第一次的1个令牌所需要的1秒偿还后，立即获得了3个令牌得到了放行。同样，第三次获取5个令牌时等待的3秒是偿还的第二次获取令牌的时间，偿还完成后立即获取5个新令牌，而并没有等待全部重新生成完成。除此之外 RateLimiter 还具有 平滑预热功能 ，下面的代码就实现了在启动3秒内，平滑提高令牌发放速率到每秒 5 个的功能：void acquireSmoothly(){ RateLimiter rateLimiter=RateLimiter.create(5,3, TimeUnit.SECONDS); long startTimeStamp = System.currentTimeMillis(); for (int i = 0; i < 15; i++) { double time = rateLimiter.acquire(); log.info("等待时间:{}s, 总时间:{}ms" ,time,System.currentTimeMillis()-startTimeStamp); } }查看运行结果：可以看到，令牌发放时间从最开始的 500ms 多逐渐缩短，在 3 秒后达到了 200ms 左右的匀速发放。总的来说，基于令牌桶实现的 RateLimiter 功能还是非常强大的，在限流的基础上还可以把请求平均分散在各个时间段内，因此在单机情况下它是使用比较广泛的限流组件。中间件限流前面讨论的四种方式都是针对单体架构，无法跨 JVM 进行限流，而在分布式、微服务架构下，可以借助一些中间件进行限。 Sentinel 是 Spring Cloud Alibaba 中常用的熔断限流组件，为我们提供了开箱即用的限流方法。使用起来也非常简单，在 service层 的方法上添加 @SentinelResource注解 ，通过 value 指定资源名称， blockHandler 指定一个方法，该方法会在原方法被限流、降级、系统保护时被调用。@Service public class QueryService { public static final String KEY="query"; @SentinelResource(value = KEY, blockHandler ="blockHandlerMethod") public String query(String name){ return "begin query,name="+name; } public String blockHandlerMethod(String name, BlockException e){ e.printStackTrace(); return "blockHandlerMethod for Query : " + name; } }配置限流规则，这里使用直接编码方式配置，指定 QPS 到达 1 时进行限流：@Component public class SentinelConfig { @PostConstruct private void init(){ List<FlowRule> rules = new ArrayList<>(); FlowRule rule = new FlowRule(QueryService.KEY); rule.setCount(1); rule.setGrade(RuleConstant.FLOW_GRADE_QPS); rule.setLimitApp("default"); rules.add(rule); FlowRuleManager.loadRules(rules); } }在 application.yml 中配置 sentinel 的端口及 dashboard 地址：spring: application: name: sentinel-test cloud: sentinel: transport: port: 8719 dashboard: localhost:8088启动项目后，启动 sentinel-dashboard ：java -Dserver.port=8088 -jar sentinel-dashboard-1.8.0.jar在浏览器打开 dashboard 就可以看见我们设置的流控规则：进行接口测试，在超过 QPS 指定的限制后，则会执行 blockHandler() 方法中的逻辑：Sentinel 在微服务架构下得到了广泛的使用，能够提供可靠的集群流量控制、服务断路等功能。在使用中，限流可以结合熔断、降级一起使用，成为有效应对三高系统的三板斧，来保证服务的稳定性。网关限流网关限流也是目前比较流行的一种方式，这里我们介绍采用 Spring Cloud 的gateway组件进行限流的方式。在项目中引入依赖， gateway 的限流实际使用的是 Redis 加 lua 脚本的方式实现的令牌桶，因此还需要引入redis的相关依赖：<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-gateway</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis-reactive</artifactId> </dependency>对 gateway 进行配置，主要就是配一下令牌的生成速率、令牌桶的存储量上限，以及用于限流的键的解析器。这里设置的桶上限为 2 ，每秒填充 1 个令牌：spring: application: name: gateway-test cloud: gateway: routes: - id: limit_route uri: lb://sentinel-test predicates: - Path=/sentinel-test/** filters: - name: RequestRateLimiter args: # 令牌桶每秒填充平均速率 redis-rate-limiter.replenishRate: 1 # 令牌桶上限 redis-rate-limiter.burstCapacity: 2 # 指定解析器，使用spEl表达式按beanName从spring容器中获取 key-resolver: "#{@pathKeyResolver}" - StripPrefix=1 redis: host: 127.0.0.1 port: 6379我们使用请求的路径作为限流的键，编写对应的解析器：@Slf4j @Component public class PathKeyResolver implements KeyResolver { public Mono<String> resolve(ServerWebExchange exchange) { String path = exchange.getRequest().getPath().toString(); log.info("Request path: {}",path); return Mono.just(path); } }启动gateway，使用 jmeter 进行测试，设置请求间隔为 500ms ，因为每秒生成一个令牌，所以后期达到了每两个请求放行1个的限流效果，在被限流的情况下，http请求会返回429状态码。除了上面的根据请求路径限流外，我们还可以灵活设置各种限流的维度，例如根据请求 heade r中携带的用户信息、或是携带的参数等等。当然，如果不想用 gateway 自带的这个 Redis 的限流器的话，我们也可以自己实现 RateLimiter接口 来实现一个自己的限流工具。gateway 实现限流的关键是 spring-cloud-gateway-core 包中的 RedisRateLimiter类 ，以及 META-INF/scripts 中的 request-rate-limiter.lua 这个脚本，如果有兴趣可以看一下具体是如何实现的。{mtitle title="总结"/}总的来说，要保证系统的抗压能力，限流是一个必不可少的环节，虽然可能会造成某些用户的请求被丢弃，但相比于突发流量造成的系统宕机来说，这些损失一般都在可以接受的范围之内。前面也说过，限流可以结合熔断、降级一起使用，多管齐下，保证服务的可用性与健壮性。

JAVA实现订单 30 分钟未支付则自动取消，我有五种方案！ 引言方案分析（1）数据库轮询（2）JDK的延迟队列（3）时间轮算法（4）redis缓存（5）使用消息队列总结引言在开发中，往往会遇到一些关于延时任务的需求。例如：生成订单30分钟未支付，则自动取消；生成订单60秒后,给用户发短信。对上述的任务，我们给一个专业的名字来形容，那就是 延时任务 。那么这里就会产生一个问题，这个 延时任务和定时任务 的区别究竟在哪里呢？一共有如下几点区别：定时任务有明确的触发时间，延时任务没有；定时任务有执行周期，而延时任务在某事件触发后一段时间内执行，没有执行周期；定时任务一般执行的是批处理操作是多个任务，而延时任务一般是单个任务。下面，我们以判断订单是否超时为例，进行方案分析。方案分析(1) 数据库轮询思路 该方案通常是在小型项目中使用，即通过一个线程定时的去扫描数据库，通过订单时间来判断是否有超时的订单，然后进行update或delete等操作。实现 博主当年早期是用 quartz 来实现的(实习那会的事)，简单介绍一下 maven 项目引入一个依赖如下所示<dependency> <groupId>org.quartz-scheduler</groupId> <artifactId>quartz</artifactId> <version>2.2.2</version> </dependency>调用Demo类MyJob如下所示package com.rjzheng.delay1; import org.quartz.JobBuilder; import org.quartz.JobDetail; import org.quartz.Scheduler; import org.quartz.SchedulerException; import org.quartz.SchedulerFactory; import org.quartz.SimpleScheduleBuilder; import org.quartz.Trigger; import org.quartz.TriggerBuilder; import org.quartz.impl.StdSchedulerFactory; import org.quartz.Job; import org.quartz.JobExecutionContext; import org.quartz.JobExecutionException; public class MyJob implements Job { public void execute(JobExecutionContext context) throws JobExecutionException { System.out.println("要去数据库扫描啦。。。"); } public static void main(String[] args) throws Exception { // 创建任务 JobDetail jobDetail = JobBuilder.newJob(MyJob.class) .withIdentity("job1", "group1").build(); // 创建触发器 每3秒钟执行一次 Trigger trigger = TriggerBuilder .newTrigger() .withIdentity("trigger1", "group3") .withSchedule( SimpleScheduleBuilder.simpleSchedule() .withIntervalInSeconds(3).repeatForever()) .build(); Scheduler scheduler = new StdSchedulerFactory().getScheduler(); // 将任务及其触发器放入调度器 scheduler.scheduleJob(jobDetail, trigger); // 调度器开始调度任务 scheduler.start(); } }运行代码，可发现每隔3秒，输出如下要去数据库扫描啦。。。优缺点 优点: 简单易行，支持集群操作。缺点:(1)对服务器内存消耗大；(2)存在延迟，比如你每隔3分钟扫描一次，那最坏的延迟时间就是3分钟；(3)假设你的订单有几千万条，每隔几分钟这样扫描一次，数据库损耗极大。(2) JDK的延迟队列思路 该方案是利用 JDK 自带的 DelayQueue 来实现，这是一个无界阻塞队列，该队列只有在延迟期满的时候才能从中获取元素，放入DelayQueue中的对象，是必须实现Delayed接口的。DelayedQueue实现工作流程如下图所示其中poll(): 获取并移除队列的超时元素，没有则返回空；take(): 获取并移除队列的超时元素，如果没有则wait当前线程，直到有元素满足超时条件，返回结果。实现定义一个类OrderDelay实现Delayed，代码如下：package com.rjzheng.delay2; import java.util.concurrent.Delayed; import java.util.concurrent.TimeUnit; public class OrderDelay implements Delayed { private String orderId; private long timeout; OrderDelay(String orderId, long timeout) { this.orderId = orderId; this.timeout = timeout + System.nanoTime(); } public int compareTo(Delayed other) { if (other == this) return 0; OrderDelay t = (OrderDelay) other; long d = (getDelay(TimeUnit.NANOSECONDS) - t .getDelay(TimeUnit.NANOSECONDS)); return (d == 0) ? 0 : ((d < 0) ? -1 : 1); } // 返回距离你自定义的超时时间还有多少 public long getDelay(TimeUnit unit) { return unit.convert(timeout - System.nanoTime(), TimeUnit.NANOSECONDS); } void print() { System.out.println(orderId+"编号的订单要删除啦。。。。"); } }运行的测试Demo为，我们设定延迟时间为3秒。package com.rjzheng.delay2; import java.util.ArrayList; import java.util.List; import java.util.concurrent.DelayQueue; import java.util.concurrent.TimeUnit; public class DelayQueueDemo { public static void main(String[] args) { // TODO Auto-generated method stub List<String> list = new ArrayList<String>(); list.add("00000001"); list.add("00000002"); list.add("00000003"); list.add("00000004"); list.add("00000005"); DelayQueue<OrderDelay> queue = new DelayQueue<OrderDelay>(); long start = System.currentTimeMillis(); for(int i = 0;i<5;i++){ //延迟三秒取出 queue.put(new OrderDelay(list.get(i), TimeUnit.NANOSECONDS.convert(3, TimeUnit.SECONDS))); try { queue.take().print(); System.out.println("After " + (System.currentTimeMillis()-start) + " MilliSeconds"); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } }输出如下：00000001编号的订单要删除啦。。。。 After 3003 MilliSeconds 00000002编号的订单要删除啦。。。。 After 6006 MilliSeconds 00000003编号的订单要删除啦。。。。 After 9006 MilliSeconds 00000004编号的订单要删除啦。。。。 After 12008 MilliSeconds 00000005编号的订单要删除啦。。。。 After 15009 MilliSeconds可以看到都是延迟3秒，订单被删除。优缺点 优点: 效率高,任务触发时间延迟低。缺点:(1) 服务器重启后，数据全部消失，怕宕机；(2) 集群扩展相当麻烦；(3) 因为内存条件限制的原因，比如下单未付款的订单数太多，那么很容易就出现OOM异常；(4) 代码复杂度较高。(3) 时间轮算法思路 先上一张时间轮的图(这图到处都是啦)时间轮算法可以类比于时钟，如上图箭头（指针）按某一个方向按固定频率轮动，每一次跳动称为一个 tick。这样可以看出定时轮由个3个重要的属性参数，ticksPerWheel（一轮的tick数），tickDuration（一个tick的持续时间）以及 timeUnit（时间单位），例如当ticksPerWheel=60，tickDuration=1，timeUnit=秒，这就和现实中的始终的秒针走动完全类似了。如果当前指针指在1上面，我有一个任务需要4秒以后执行，那么这个执行的线程回调或者消息将会被放在5上。那如果需要在20秒之后执行怎么办，由于这个环形结构槽数只到8，如果要20秒，指针需要多转2圈。位置是在2圈之后的5上面（20 % 8 + 1）。实现 我们用Netty的HashedWheelTimer来实现 给Pom加上下面的依赖：<dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <version>4.1.24.Final</version> </dependency>测试代码HashedWheelTimerTest，如下所示：package com.rjzheng.delay3; import io.netty.util.HashedWheelTimer; import io.netty.util.Timeout; import io.netty.util.Timer; import io.netty.util.TimerTask; import java.util.concurrent.TimeUnit; public class HashedWheelTimerTest { static class MyTimerTask implements TimerTask{ boolean flag; public MyTimerTask(boolean flag){ this.flag = flag; } public void run(Timeout timeout) throws Exception { // TODO Auto-generated method stub System.out.println("要去数据库删除订单了。。。。"); this.flag =false; } } public static void main(String[] argv) { MyTimerTask timerTask = new MyTimerTask(true); Timer timer = new HashedWheelTimer(); timer.newTimeout(timerTask, 5, TimeUnit.SECONDS); int i = 1; while(timerTask.flag){ try { Thread.sleep(1000); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } System.out.println(i+"秒过去了"); i++; } } }输出如下：1秒过去了 2秒过去了 3秒过去了 4秒过去了 5秒过去了 要去数据库删除订单了。。。。 6秒过去了优缺点 优点: 效率高，任务触发时间延迟时间比delayQueue低，代码复杂度比delayQueue低。缺点:(1) 服务器重启后，数据全部消失，怕宕机；(2) 集群扩展相当麻烦；(3) 因为内存条件限制的原因，比如下单未付款的订单数太多，那么很容易就出现OOM异常。(4) redis缓存思路一 利用 redis 的 zset ，zset是一个有序集合，每一个元素(member)都关联了一个score，通过score排序来取集合中的值。zset常用命令添加元素:ZADD key score member [[score member] [score member] ...]按顺序查询元素:ZRANGE key start stop [WITHSCORES]查询元素score:ZSCORE key member移除元素:ZREM key member [member ...]测试如下： # 添加单个元素 redis> ZADD page_rank 10 google.com (integer) 1 # 添加多个元素 redis> ZADD page_rank 9 baidu.com 8 bing.com (integer) 2 redis> ZRANGE page_rank 0 -1 WITHSCORES 1) "bing.com" 2) "8" 3) "baidu.com" 4) "9" 5) "google.com" 6) "10" # 查询元素的score值 redis> ZSCORE page_rank bing.com "8" # 移除单个元素 redis> ZREM page_rank google.com (integer) 1 redis> ZRANGE page_rank 0 -1 WITHSCORES 1) "bing.com" 2) "8" 3) "baidu.com" 4) "9"那么如何实现呢？我们将订单超时时间戳与订单号分别设置为 score 和 member ，系统扫描第一个元素判断是否超时，具体如下图所示：实现一package com.rjzheng.delay4; import java.util.Calendar; import java.util.Set; import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPool; import redis.clients.jedis.Tuple; public class AppTest { private static final String ADDR = "127.0.0.1"; private static final int PORT = 6379; private static JedisPool jedisPool = new JedisPool(ADDR, PORT); public static Jedis getJedis() { return jedisPool.getResource(); } //生产者,生成5个订单放进去 public void productionDelayMessage(){ for(int i=0;i<5;i++){ //延迟3秒 Calendar cal1 = Calendar.getInstance(); cal1.add(Calendar.SECOND, 3); int second3later = (int) (cal1.getTimeInMillis() / 1000); AppTest.getJedis().zadd("OrderId", second3later,"OID0000001"+i); System.out.println(System.currentTimeMillis()+"ms:redis生成了一个订单任务：订单ID为"+"OID0000001"+i); } } //消费者，取订单 public void consumerDelayMessage(){ Jedis jedis = AppTest.getJedis(); while(true){ Set<Tuple> items = jedis.zrangeWithScores("OrderId", 0, 1); if(items == null || items.isEmpty()){ System.out.println("当前没有等待的任务"); try { Thread.sleep(500); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } continue; } int score = (int) ((Tuple)items.toArray()[0]).getScore(); Calendar cal = Calendar.getInstance(); int nowSecond = (int) (cal.getTimeInMillis() / 1000); if(nowSecond >= score){ String orderId = ((Tuple)items.toArray()[0]).getElement(); jedis.zrem("OrderId", orderId); System.out.println(System.currentTimeMillis() +"ms:redis消费了一个任务：消费的订单OrderId为"+orderId); } } } public static void main(String[] args) { AppTest appTest =new AppTest(); appTest.productionDelayMessage(); appTest.consumerDelayMessage(); } }此时对应输出如下：1525086085261ms:redis生成了一个订单任务：订单ID为OID00000010 1525086085263ms:redis生成了一个订单任务：订单ID为OID00000011 1525086085266ms:redis生成了一个订单任务：订单ID为OID00000012 1525086085268ms:redis生成了一个订单任务：订单ID为OID00000013 1525086085270ms:redis生成了一个订单任务：订单ID为OID00000014 1525086088000ms:redis消费了一个任务：消费的订单OrderId为OID00000010 1525086088001ms:redis消费了一个任务：消费的订单OrderId为OID00000011 1525086088002ms:redis消费了一个任务：消费的订单OrderId为OID00000012 1525086088003ms:redis消费了一个任务：消费的订单OrderId为OID00000013 1525086088004ms:redis消费了一个任务：消费的订单OrderId为OID00000014 当前没有等待的任务 当前没有等待的任务 当前没有等待的任务可以看到，几乎都是3秒之后，消费订单。然而，这一版存在一个致命的硬伤，在高并发条件下，多消费者会取到同一个订单号，我们上测试代码 ThreadTest 。package com.rjzheng.delay4; import java.util.concurrent.CountDownLatch; public class ThreadTest { private static final int threadNum = 10; private static CountDownLatch cdl = new CountDownLatch(threadNum); static class DelayMessage implements Runnable{ public void run() { try { cdl.await(); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } AppTest appTest =new AppTest(); appTest.consumerDelayMessage(); } } public static void main(String[] args) { AppTest appTest =new AppTest(); appTest.productionDelayMessage(); for(int i=0;i<threadNum;i++){ new Thread(new DelayMessage()).start(); cdl.countDown(); } } }输出如下所示：1525087157727ms:redis生成了一个订单任务：订单ID为OID00000010 1525087157734ms:redis生成了一个订单任务：订单ID为OID00000011 1525087157738ms:redis生成了一个订单任务：订单ID为OID00000012 1525087157747ms:redis生成了一个订单任务：订单ID为OID00000013 1525087157753ms:redis生成了一个订单任务：订单ID为OID00000014 1525087160009ms:redis消费了一个任务：消费的订单OrderId为OID00000010 1525087160011ms:redis消费了一个任务：消费的订单OrderId为OID00000010 1525087160012ms:redis消费了一个任务：消费的订单OrderId为OID00000010 1525087160022ms:redis消费了一个任务：消费的订单OrderId为OID00000011 1525087160023ms:redis消费了一个任务：消费的订单OrderId为OID00000011 1525087160029ms:redis消费了一个任务：消费的订单OrderId为OID00000011 1525087160038ms:redis消费了一个任务：消费的订单OrderId为OID00000012 1525087160045ms:redis消费了一个任务：消费的订单OrderId为OID00000012 1525087160048ms:redis消费了一个任务：消费的订单OrderId为OID00000012 1525087160053ms:redis消费了一个任务：消费的订单OrderId为OID00000013 1525087160064ms:redis消费了一个任务：消费的订单OrderId为OID00000013 1525087160065ms:redis消费了一个任务：消费的订单OrderId为OID00000014 1525087160069ms:redis消费了一个任务：消费的订单OrderId为OID00000014 当前没有等待的任务 当前没有等待的任务 当前没有等待的任务 当前没有等待的任务显然，出现了多个线程消费同一个资源的情况。解决方案(1) 用分布式锁，但是用分布式锁，性能下降了，该方案不细说；(2) 对ZREM的返回值进行判断，只有大于0的时候，才消费数据，于是将 consumerDelayMessage() 方法里的。if(nowSecond >= score){ String orderId = ((Tuple)items.toArray()[0]).getElement(); jedis.zrem("OrderId", orderId); System.out.println(System.currentTimeMillis()+"ms:redis消费了一个任务：消费的订单OrderId为"+orderId); }修改为：if(nowSecond >= score){ String orderId = ((Tuple)items.toArray()[0]).getElement(); Long num = jedis.zrem("OrderId", orderId); if( num != null && num>0){ System.out.println(System.currentTimeMillis()+"ms:redis消费了一个任务：消费的订单OrderId为"+orderId); } }在这种修改后，重新运行ThreadTest类，发现输出正常了。思路二 该方案使用 redis 的 Keyspace Notifications ，中文翻译就是键空间机制，就是利用该机制可以在key失效之后，提供一个回调，实际上是redis会给客户端发送一个消息。是需要 redis版本2.8以上 。实现二 在 redis.conf 中，加入一条配置：notify-keyspace-events Ex运行代码如下：package com.rjzheng.delay5; import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPool; import redis.clients.jedis.JedisPubSub; public class RedisTest { private static final String ADDR = "127.0.0.1"; private static final int PORT = 6379; private static JedisPool jedis = new JedisPool(ADDR, PORT); private static RedisSub sub = new RedisSub(); public static void init() { new Thread(new Runnable() { public void run() { jedis.getResource().subscribe(sub, "__keyevent@0__:expired"); } }).start(); } public static void main(String[] args) throws InterruptedException { init(); for(int i =0;i<10;i++){ String orderId = "OID000000"+i; jedis.getResource().setex(orderId, 3, orderId); System.out.println(System.currentTimeMillis()+"ms:"+orderId+"订单生成"); } } static class RedisSub extends JedisPubSub { @Override public void onMessage(String channel, String message) { System.out.println(System.currentTimeMillis()+"ms:"+message+"订单取消"); } } }输出如下：1525096202813ms:OID0000000订单生成 1525096202818ms:OID0000001订单生成 1525096202824ms:OID0000002订单生成 1525096202826ms:OID0000003订单生成 1525096202830ms:OID0000004订单生成 1525096202834ms:OID0000005订单生成 1525096202839ms:OID0000006订单生成 1525096205819ms:OID0000000订单取消 1525096205920ms:OID0000005订单取消 1525096205920ms:OID0000004订单取消 1525096205920ms:OID0000001订单取消 1525096205920ms:OID0000003订单取消 1525096205920ms:OID0000006订单取消 1525096205920ms:OID0000002订单取消可以明显看到3秒过后，订单取消了。ps: redis的pub/sub 机制存在一个硬伤，官网内容如下：原: Because Redis Pub/Sub is fire and forget currently there is no way to use this feature if your application demands reliable notification of events, that is, if your Pub/Sub client disconnects, and reconnects later, all the events delivered during the time the client was disconnected are lost.翻: Redis的发布/订阅目前是即发即弃(fire and forget)模式的，因此无法实现事件的可靠通知。也就是说，如果发布/订阅的客户端断链之后又重连，则在客户端断链期间的所有事件都丢失了。因此，方案二不是太推荐。当然，如果你对可靠性要求不高，可以使用。优缺点 优点:(1) 由于使用Redis作为消息通道，消息都存储在Redis中。如果发送程序或者任务处理程序挂了，重启之后，还有重新处理数据的可能性；(2) 做集群扩展相当方便；(3) 时间准确度高。缺点: (1) 需要额外进行redis维护。(5) 使用消息队列我们可以采用 RabbitMQ 的延时队列。RabbitMQ 具有以下两个特性，可以实现延迟队列：（1）RabbitMQ可以针对Queue和Message设置 x-message-tt，来控制消息的生存时间，如果超时，则消息变为dead letter（2）lRabbitMQ的Queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key（可选）两个参数，用来控制队列内出现了deadletter，则按照这两个参数重新路由。结合以上两个特性，就可以模拟出延迟消息的功能,具体的，我改天再写一篇文章，这里再讲下去，篇幅太长。优缺点优点: 高效，可以利用rabbitmq的分布式特性轻易的进行横向扩展，消息支持持久化增加了可靠性。缺点：本身的易用度要依赖于rabbitMq的运维。因为要引用rabbitMq，所以复杂度和成本变高。总结本文总结了目前互联网中，绝大部分的延时任务的实现方案。希望大家在工作中能够有所收获。

【Minecraft开服教程】使用 MCSM 面板一键搭建我的世界服务器，并内网穿透公网远程联机 文章目录Mcsmanager安装创建Minecraft服务器本地测试联机内网穿透前言MCSManager 是一个开源、分布式、轻量级、一键开服、支持大部分游戏服务端和控制台程序的管理工具，我们可以用它来一键部署搭建Minecraft我的世界服务器，跟小伙伴们联机。现在一般家庭局域网宽带没有公网IP，不在同个局域网下的小伙伴没办法直接联机。所以我们在这个教程中同时做内网穿透，将本地端口映射到公网上，使用所生成的公网地址来远程联机，突破局域网的限制，不需要公网IP，不需要设置路由器，也不用购买云服务器。1.Mcsmanager安装点击下载 MCSManager ，我们下载windwos版本 【不支持windwos10以下系统安装】 。下载好后解压打开,然后点击启动器,启动后台程序启动后,会自动跳转到浏览器打开,如没有跳转,可直接在浏览器输入localhost:23333,首次访问需要我们创建一个账号然后即可进入到面板界面2.创建Minecraft服务器在面板中,我们点击快速开始,创建一个Minecraft 服务器选择一键开服这里可以自由选择相应的版本,这里选择1.19.2原版,点击安装然后输入服务器的名称,输入后等待安装完成安装完成后,点击前往实例控制台然后出现联机方式，我们点击下面跳过设置联机方式的选择接下来修改配置文件,取消正版服务器的验证,点击文件管理找到 server.properties 文件,点击编辑把 online-mode 的值改为 false ,然后保存,关闭然后开启实例,这里的实例表示服务的意思启动成功,我们可以看到端口信息,端口号为:255653.本地测试联机打开我的世界启动器,选择和服务器一样的版本1.19.2,启动然后点击多人游戏正常来说会自动搜索到这个本地服务器,可以直接点击即可连接如果没有,我们可以点击下面的直接连接,然后输入地址:127.0.0.1:25565,点击加入服务器然后我们可以看到成功进入了游戏4. 内网穿透测试本地联机正常后，我们接着做内网穿透，实现在外不同网络环境下的远程联机。内网穿透网上很多，有收费的和免费的这里就不介绍了。在本地安装成功后，创建一条隧道将我的世界服务端口映射到公网上，然后使用所生成的公网地址来远程联机就可以了。

让你免费拥有一个无限大小的存储服务器【全过程分享】 前言前段时间玩了AList，突发奇想既然RaiDrive能把Alist映射到本地，那是不是能把映射盘部署到服务器呢？这样我就有了一个无限大小的存储了，既然写了这篇文章，说明还是可以的，这篇没什么技术含量，没有什么深层次解读，只说实现过程细节步骤，查的资料也比较多，只是觉得有趣记录一下，如果涉及到谁的原创内容我没说请私聊，我给你名字和链接加上（/doge 保命）话不多说，开搂步骤大纲使用Alist管理网盘使用RaiDrive把Alist映射到本地把映射盘部署到服务器一：Alist管理网盘Alist：相当于一个聚合网盘管理器，能实现网盘的统一管理，并能达到文件在线浏览的效果AList 开源项目地址： https://github.com/alist-org/alist可以点击链接下载Alist Windows最新版：https://github.com/alist-org/alist/releases/download/v3.28.0/alist-windows-amd64.zip也可以在官网找自己喜欢的版本，但是注意新版本（V3 及更高版本）与 V2 不兼容下载完成后解压，在解压后的文件夹中打开cmd，运行 alist server 启动服务password is 后面是初始密码 start server 后面是 AList 的管理页面地址，IP(127.0.0.1/localhost) + 端口然后打开浏览器，输入链接127.0.0.1:5244 / localhost:5244输入账号（admin）密码（刚刚的初始密码）进入管理页面后可以点击右下角管理，修改初始密码，重新登录接下来添加需要管理的网盘，这里以百度网盘为例点击 管理 -> 存储 -> 添加驱动 选择百度网盘挂载路径 是根目录下的文件夹WebDav策略 推荐选择代理地址刷新令牌 为你百度网盘的 refresh_token ，官方文档里有获取方法：获取客户端ID与密匙 会自动获取，无需手动填写参数填写完毕后，直接点击添加 这就添加好了，接下来你就可以通过本机IP+端口访问网盘了如果要添加更多网盘，可以去 官方文档 查看，基本步骤都大差不差二：使用RaiDrive把alist映射到本地盘RaiDrive：将云存储和网络存储映射在本地磁盘 RaiDrive官网 ：www.raidrive.com/ （点击Download下载）下载好后，打开 install 一直无脑下一步即可（注意安装位置，不建议安装到C盘）安装好后打开，点击 添加 服务类型 ：选择 NAS —— WebDAV 虚拟驱动器 ：前面是盘符，后面是名字，可以自行选择地址 ：取消勾选，http后面填Alist的登陆地址，下面填/dav账户 ：刚刚Alist登录的账号密码随后点击 连接 ，搞定此时在我的电脑里就多一个盘，打开之后就会显示Alist中管理的网盘注意：此处的磁盘大小为 7.99EB ！！而 1EiB = 1,024 PiB = 1,048,576 TiB = 1,073,741,824 GiB 先别急着幻想，其实没有这么大，只不过是 没办法显示具体的大小，而随机生成的大小而已 /doge三：部署到服务器这里详细步骤我就不赘述了直接使用 Idea 新建一个web工程，配置Tomcat，默认 "Hello World" 出现之后（说明工程没有问题）在 Deployment 中添加映射路径选择刚刚添加的映射路径然后启动服务打开网页输入地址：localhost（ip）：8080（Tomcat端口）/bd（Application context）至此，大功告成拓展啥？你问我这有啥用？既然都部署到服务器了，那还不是被咱玩弄于股掌之间以Android为例随便写个数据库 存放个文件路径，刚刚的web项目里 随便写个接口把路径返回，Android随便写个请求 从接口拿到数据，再随便找个播放器加载（顺便提一嘴，如果不行。。。那就是你写的太随便了/doge）注意事项还记得第一步，解压 Alist 的时候，启动了 alist server 的窗口吗，这个是在线的， 不能关闭!!! 关闭会导致Alist获取不到网盘资源，那么RaiDrive自然也就映射不到本地所以如果不想麻烦每次都手动启动，可以写个脚本实现：隐藏cmd窗口开机自启动 隐藏cmd窗口 :在 Alist解压文件夹 中新建一个 txt ，填入以下内容Set ws = CreateObject("Wscript.Shell") ws.run "D:\alist\alist.exe server",vbhide MsgBox "success"ws.run 修改为你 alist.exe 的路径保存之后将后缀改为 vbs 双击打开即可，弹出 success 说明启动成功这时你会发现，并没有cmd黑窗口弹出，打开alist管理页面，一切正常，🆗可以在任务管理器的进程中找到你启动的脚本，也可以在此处结束进程开机自启动 ：将刚刚 vbs 文件，添加个快捷方式（注意！是快捷方式）打开运行( Win + R )，输入 shell:startup ，再将快捷方式拖入打开的文件夹里重启测试下，开机后等待success窗口弹出，大功告成！结语Alist加载会有延迟，对于过大的文件，有时亦会导致卡顿，对于宽带会有一定的要求，目前网上对Alist各网盘的优化和解决办法很多，大家可以根据自己的需求添加完善内网访问效果理想，若想在公网访问，无论是本地穿透，还是部署到云服务器，都需要通过服务器中转，转发之后具体能达到什么样的效果需要测试，如果有哪位大佬做出来了欢迎告知本篇没什么深层讲解，因此也没什么难度，按照步骤一步一步来即可，如果有问题随时提问，要是我哪里写错了，别客气，过来给我一巴掌，就行了（温柔点，别打肿，影响颜值就不好了/doge）。

Centos 7.9 离线安装 ORACLE 19C 本文涉及的安装包从以下百度网盘地址可获取：链接：https://pan.baidu.com/s/1XD_64B7awDjvkqcGXwWRqA提取码：42ow一、基础环境配置1、关闭系统防火墙systemctl stop firewalld systemctl disable firewalld2、关闭selinuxvim /etc/selinux/config SELINUX=disabled3、准备安装oracle安装所需依赖说明：下载安装oracle时所需依赖的rpm包，然后制作本地yum源进行安装rpm依赖包下载地址：https://pan.baidu.com/s/1zg6DUG0BTxY7H63lj78CrA 提取码：2ucbbase.zip放在/root路径下解压unzip base.zip创建离线yum源：备份原有repo文件 ，建个bakup文件夹把原路径下的文件放进去创建local.repo文件echo "[local]" > /etc/yum.repos.d/local.repo echo "name=local" >> /etc/yum.repos.d/local.repo echo "enable=1" >> /etc/yum.repos.d/local.repo echo "baseurl=file:///root/base" >> /etc/yum.repos.d/local.repo echo "gpgcheck=0" >> /etc/yum.repos.d/local.repoyum clean all4、安装oracle-database-preinstall 链接：https://pan.baidu.com/s/11TGhQ8H95umuV5PRUrfEcw 提取码：3yruyum -y localinstall oracle-database-preinstall-19c-1.0-1.el7.x86_64.rpm可能会出现如下报错：解决上面报错： 1.可以从以下网站缺少的依赖软件包：https://rpmfind.net/linux/rpm2html/search.php?query=kernel-headers&submit=Search+...&system=centos&arch=2.使用rpm命令手工安装缺少的依赖软件包rpm -i glibc-devel-2.17-317.el7.x86_64.rpm 二、安装Oracle 19c1、安装Oracle 19c 下载地址：https://www.oracle.com/database/technologies/oracle-database-software-downloads.htmlyum -y localinstall oracle-database-ee-19c-1.0-1.x86_64.rpm初始化Oracle数据库： 如需，可修改/etc/init.d/oracledb_ORCLCDB-19c，比如，CDB模式，实例ID等等。/etc/init.d/oracledb_ORCLCDB-19c configure配置环境变量： 切换用户su - oracle vim /home/oracle/.bash_profile加入以下内容：export ORACLE_HOME=/opt/oracle/product/19c/dbhome_1 export PATH=$PATH:/opt/oracle/product/19c/dbhome_1/bin export ORACLE_SID=ORCLCDB登陆oracle数据库(需要退出oracle用户重新登录):exit su - oracle sqlplus / as sysdba修改密码:alter user system identified by 123456;重新登录： ========================ORACLE19C的sqlnet.ora配置：/opt/oracle/product/19c/dbhome_1/network/adminNAMES.DIRECTORY_PATH= (TNSNAMES, ONAMES, HOSTNAME) SQLNET.AUTHENTICATION_SERVICES=(ALL) SQLNET.ALLOWED_LOGON_VERSION_SERVER=8 SQLNET.ALLOWED_LOGON_CLIENT=8 SQLNET.INBOUND_CONNECT_TIMEOUT=0 SQLNET.EXPIRE_TIME=10以上配置可解决兼容低版本客户端、客户端超时卡慢、dblink等问题，重启监听后生效lsnrctl stop lsnrctl start其中兼容低版本客户端的，在用户已经创建的情况下要再修改一次密码才能生效。表空间文件放置的文件夹需要对oracle用户授权：chown oracle:oinstall /data chown oracle:oinstall -R /data chmod 777 -R /data启动oraclelsnrctl start su - oracle sqlplus / as sysdba startup