首页 文章 精选 留言 我的

精选列表

搜索[核心工具],共10010篇文章
优秀的个人博客,低调大师

elasticsearch 核心的http api

开篇 作为一个elasticsearch的新手,我觉得elasticsearch的官网的文档归类清楚,作为新手很有必要多阅读几次,每次都能加深印象。这里作为个人理解进行一些简单的归类便于查阅。 对于新手而言,熟悉各类能够掌握简单的操作包括:索引的创建删除查询,文档的增删改查、集群信息查询等。 Indices APIs:负责索引Index的创建(create)、删除(delete)、获取(get)、索引存在(exist)等操作。 Document APIs:负责索引文档的创建(index)、删除(delete)、获取(get)等操作。 Search APIs:负责索引文档的search(查询),Document APIS根据doc_id进行查询,Search APIs]根据条件查询。 Aggregations:负责针对索引的文档各维度的聚合(Aggregation)。 cat APIs:负责查询索引相关的各类信息查询。 Cluster APIs:负责集群相关的各类信息查询。 提供一个操作的方法,就是本机搭建个es集群然后部署一个kibana,通过kibana熟悉http api就可以了,这篇文章就专注下cat和cluster相关的api。 cat APIs介绍 cat APIs cat aliases 说明: aliases 负责展示当前es集群配置别名包括filter和routing信息。 命令: GET /_cat/aliases?v GET /_cat/aliases/alias1,alias2 返回: alias index filter routing.index routing.search alias1 test1 - - - alias2 test1 * - - alias3 test1 - 1 1 alias4 test1 - 2 1,2 cat allocation 说明: allocation负责展示es的每个数据节点分配的索引分片以及使用的磁盘空间。 命令: GET /_cat/allocation?v 返回值: shards disk.indices disk.used disk.avail disk.total disk.percent host ip node 5 260b 47.3gb 43.4gb 100.7gb 46 127.0.0.1 127.0.0.1 CSUXak2 cat count 说明: count负责展示整个ES集群或者单个索引的文档数。 命令: GET /_cat/count?v GET /_cat/count/twitter?v 返回值: epoch timestamp count 1475868259 15:24:20 120 cat fielddata 说明: fielddata负责展示ES集群每个数据节点中fileddata占用的堆内存。 命令: GET /_cat/fielddata?v GET /_cat/fielddata?v&fields=body GET /_cat/fielddata/body,soul?v 返回值: id host ip node field size Nqk-6inXQq-OxUfOUI8jNQ 127.0.0.1 127.0.0.1 Nqk-6in body 544b Nqk-6inXQq-OxUfOUI8jNQ 127.0.0.1 127.0.0.1 Nqk-6in soul 480b cat health 说明: health负责展示集群的健康状态 命令: GET /_cat/health?v GET /_cat/health?v&ts=false 返回值: epoch timestamp cluster status node.total node.data shards pri relo init unassign pending_tasks max_task_wait_time active_shards_percent 1475871424 16:17:04 elasticsearch green 1 1 5 5 0 0 0 0 - 100.0% cat indices 说明: indices负责提供索引的相关信息,包括组成一个索引(index)的shard、document的数量,删除的doc数量,主存大小和所有索引的总存储大小。 命令: GET /_cat/indices/twi*?v&s=index 返回值: health status index uuid pri rep docs.count docs.deleted store.size pri.store.size yellow open twitter u8FNjxh8Rfy_awN11oDKYQ 1 1 1200 0 88.1kb 88.1kb green open twitter2 nYFWZEO7TUiOjLQXBaYJpA 5 0 0 0 260b 260b cat master 说明: master负责展示es集群的master节点信息包括节点id、节点名、ip地址等。 命令: GET /_cat/master?v 返回值: id host ip node YzWoH_2BT-6UjVGDyPdqYg 127.0.0.1 127.0.0.1 YzWoH_2 cat nodeattrs 说明: nodeattrs负责展示通用的节点信息。 命令: GET /_cat/nodeattrs?v 返回值: node host ip attr value node-0 127.0.0.1 127.0.0.1 testattr test cat nodes 说明: nodes负责展示es集群的拓扑信息。 命令: GET /_cat/nodes?v 返回值: ip heap.percent ram.percent cpu load_1m load_5m load_15m node.role master name 127.0.0.1 65 99 42 3.07 mdi * mJw06l1 cat pending tasks 说明: pending_tasks返回集群层面的未执行的任务列表包括创建索引,更新mapping,allocate分片信息。 命令: GET /_cat/pending_tasks?v 返回值: insertOrder timeInQueue priority source 1685 855ms HIGH update-mapping [foo][t] 1686 843ms HIGH update-mapping [foo][t] 1693 753ms HIGH refresh-mapping [foo][[t]] 1688 816ms HIGH update-mapping [foo][t] 1689 802ms HIGH update-mapping [foo][t] 1690 787ms HIGH update-mapping [foo][t] 1691 773ms HIGH update-mapping [foo][t] cat plugins 说明: plugins命令展示每个节点正在运行的插件信息。 命令: GET /_cat/plugins?v&s=component&h=name,component,version,description 返回值: name component version description U7321H6 analysis-icu 6.5.4 The ICU Analysis plugin integrates Lucene ICU module into elasticsearch, adding ICU relates analysis components. U7321H6 analysis-kuromoji 6.5.4 The Japanese (kuromoji) Analysis plugin integrates Lucene kuromoji analysis module into elasticsearch. U7321H6 analysis-nori 6.5.4 The Korean (nori) Analysis plugin integrates Lucene nori analysis module into elasticsearch. U7321H6 analysis-phonetic 6.5.4 The Phonetic Analysis plugin integrates phonetic token filter analysis with elasticsearch. cat recovery 说明: recovery命令展示索引分片恢复,包括正在进行的和已经完成的任务。 命令: GET _cat/recovery?v GET _cat/recovery?v&h=i,s,t,ty,st,shost,thost,f,fp,b,bp 返回值: index shard time type stage source_host source_node target_host target_node repository snapshot files files_recovered files_percent files_total bytes bytes_recovered bytes_percent bytes_total translog_ops translog_ops_recovered translog_ops_percent twitter 0 13ms store done n/a n/a 127.0.0.1 node-0 n/a n/a 0 0 100% 13 0 0 100% 9928 0 0 100.0% cat repositories 说明: repositories命令展示在集群中注册的快照仓库。 命令: GET /_cat/repositories?v GET /_cat/thread_pool/generic?v&h=id,name,active,rejected,completed 返回值: id type repo1 fs repo2 s3 cat thread pool 说明: thread_pool命令展示集群节点的线程池的统计信息,包括处于活跃、队列等待、拒绝的线程任务信息。 命令: GET /_cat/thread_pool 返回值: node-0 bulk 0 0 0 node-0 fetch_shard_started 0 0 0 node-0 fetch_shard_store 0 0 0 node-0 flush 0 0 0 node-0 force_merge 0 0 0 node-0 generic 0 0 0 node-0 get 0 0 0 node-0 index 0 0 0 node-0 listener 0 0 0 node-0 management 1 0 0 node-0 refresh 0 0 0 node-0 search 0 0 0 node-0 snapshot 0 0 0 node-0 warmer 0 0 0 cat shards 说明: shards命令展示每个节点包括哪些分片信息。 命令: GET _cat/shards GET _cat/shards/twitt* 返回值: twitter 0 p STARTED 3014 31.1mb 192.168.56.10 H5dfFeA cat segments 说明: segments命令展示索引的segments的信息。 命令: GET /_cat/segments?v 返回值: index shard prirep ip segment generation docs.count docs.deleted size size.memory committed searchable version compound test 3 p 127.0.0.1 _0 0 1 0 3kb 2042 false true 7.1.0 true test1 3 p 127.0.0.1 _0 0 1 0 3kb 2042 false true 7.1.0 true cat snapshots 说明: snapshots命令展示属于某个指定快照仓库的所有快照信息。通过/_cat/repositories查找所有的仓库列表。 命令: GET /_cat/snapshots/repo1?v&s=id 返回值: id status start_epoch start_time end_epoch end_time duration indices successful_shards failed_shards total_shards snap1 FAILED 1445616705 18:11:45 1445616978 18:16:18 4.6m 1 4 1 5 snap2 SUCCESS 1445634298 23:04:58 1445634672 23:11:12 6.2m 2 10 0 10 cat templates 说明: templates展示已存在的模板信息。 命令: GET /_cat/templates?v&s=name 返回值: name index_patterns order version template0 [te*] 0 template1 [tea*] 1 template2 [teak*] 2 7 Cluster APIs介绍 Cluster APIsCluster Health 说明: health命令展示集群的健康状态, 命令: GET _cluster/health GET /_cluster/health/test1,test2 返回值: { "cluster_name" : "testcluster", "status" : "yellow", "timed_out" : false, "number_of_nodes" : 1, "number_of_data_nodes" : 1, "active_primary_shards" : 5, "active_shards" : 5, "relocating_shards" : 0, "initializing_shards" : 0, "unassigned_shards" : 5, "delayed_unassigned_shards": 0, "number_of_pending_tasks" : 0, "number_of_in_flight_fetch": 0, "task_max_waiting_in_queue_millis": 0, "active_shards_percent_as_number": 50.0 } Cluster State 说明: state命令展示集群详细的状态信息, 其中metrics指标包括version、master_node、nodes、routing_table、metadata、blocks。 命令: GET /_cluster/state GET /_cluster/state/{metrics}/{indices} GET /_cluster/state/metadata,routing_table/foo,bar 返回值: { "cluster_name": "my-application", "compressed_size_in_bytes": 847, "version": 10, "state_uuid": "DDWoFtm7TvCydbp8SKRQdw", "master_node": "YOIwJR1bTxG0iKHur_M-Rg" } Cluster Stats 说明: Stats命令展示集群的多维度统计信息。 指标包括shard numbers, store size, memory usage、 number, roles, os, jvm versions, memory usage, cpu and installed plugins。 命令: GET /_cluster/stats?human&pretty 返回值: { "_nodes" : { "total" : 1, "successful" : 1, "failed" : 0 }, "cluster_name": "elasticsearch", "timestamp": 1459427693515, "status": "green", "indices": { "count": 1, "shards": { "total": 5, "primaries": 5, "replication": 0, "index": { "shards": { "min": 5, "max": 5, "avg": 5 }, "primaries": { "min": 5, "max": 5, "avg": 5 }, "replication": { "min": 0, "max": 0, "avg": 0 } } }, "docs": { "count": 10, "deleted": 0 }, "store": { "size": "16.2kb", "size_in_bytes": 16684 }, "fielddata": { "memory_size": "0b", "memory_size_in_bytes": 0, "evictions": 0 }, "query_cache": { "memory_size": "0b", "memory_size_in_bytes": 0, "total_count": 0, "hit_count": 0, "miss_count": 0, "cache_size": 0, "cache_count": 0, "evictions": 0 }, "completion": { "size": "0b", "size_in_bytes": 0 }, "segments": { "count": 4, "memory": "8.6kb", "memory_in_bytes": 8898, "terms_memory": "6.3kb", "terms_memory_in_bytes": 6522, "stored_fields_memory": "1.2kb", "stored_fields_memory_in_bytes": 1248, "term_vectors_memory": "0b", "term_vectors_memory_in_bytes": 0, "norms_memory": "384b", "norms_memory_in_bytes": 384, "points_memory" : "0b", "points_memory_in_bytes" : 0, "doc_values_memory": "744b", "doc_values_memory_in_bytes": 744, "index_writer_memory": "0b", "index_writer_memory_in_bytes": 0, "version_map_memory": "0b", "version_map_memory_in_bytes": 0, "fixed_bit_set": "0b", "fixed_bit_set_memory_in_bytes": 0, "max_unsafe_auto_id_timestamp" : -9223372036854775808, "file_sizes": {} } }, "nodes": { "count": { "total": 1, "data": 1, "coordinating_only": 0, "master": 1, "ingest": 1 }, "versions": [ "6.1.4" ], "os": { "available_processors": 8, "allocated_processors": 8, "names": [ { "name": "Mac OS X", "count": 1 } ], "mem" : { "total" : "16gb", "total_in_bytes" : 17179869184, "free" : "78.1mb", "free_in_bytes" : 81960960, "used" : "15.9gb", "used_in_bytes" : 17097908224, "free_percent" : 0, "used_percent" : 100 } }, "process": { "cpu": { "percent": 9 }, "open_file_descriptors": { "min": 268, "max": 268, "avg": 268 } }, "jvm": { "max_uptime": "13.7s", "max_uptime_in_millis": 13737, "versions": [ { "version": "1.8.0_74", "vm_name": "Java HotSpot(TM) 64-Bit Server VM", "vm_version": "25.74-b02", "vm_vendor": "Oracle Corporation", "count": 1 } ], "mem": { "heap_used": "57.5mb", "heap_used_in_bytes": 60312664, "heap_max": "989.8mb", "heap_max_in_bytes": 1037959168 }, "threads": 90 }, "fs": { "total": "200.6gb", "total_in_bytes": 215429193728, "free": "32.6gb", "free_in_bytes": 35064553472, "available": "32.4gb", "available_in_bytes": 34802409472 }, "plugins": [ { "name": "analysis-icu", "version": "6.1.4", "description": "The ICU Analysis plugin integrates Lucene ICU module into elasticsearch, adding ICU relates analysis components.", "classname": "org.elasticsearch.plugin.analysis.icu.AnalysisICUPlugin", "has_native_controller": false }, { "name": "ingest-geoip", "version": "6.1.4", "description": "Ingest processor that uses looksup geo data based on ip adresses using the Maxmind geo database", "classname": "org.elasticsearch.ingest.geoip.IngestGeoIpPlugin", "has_native_controller": false }, { "name": "ingest-user-agent", "version": "6.1.4", "description": "Ingest processor that extracts information from a user agent", "classname": "org.elasticsearch.ingest.useragent.IngestUserAgentPlugin", "has_native_controller": false } ], "network_types" : { "transport_types" : { "netty4" : 1 }, "http_types" : { "netty4" : 1 } } } } Pending cluster tasks 说明: pending_tasks返回集群级别的未执行的任务, 包括创建索引,、更新mapping、 分配分片等。 命令: ET /_cluster/pending_tasks 返回值: { "tasks": [ { "insert_order": 101, "priority": "URGENT", "source": "create-index [foo_9], cause [api]", "time_in_queue_millis": 86, "time_in_queue": "86ms" }, { "insert_order": 46, "priority": "HIGH", "source": "shard-started ([foo_2][1], node[tMTocMvQQgGCkj7QDHl3OA], [P], s[INITIALIZING]), reason [after recovery from shard_store]", "time_in_queue_millis": 842, "time_in_queue": "842ms" }, { "insert_order": 45, "priority": "HIGH", "source": "shard-started ([foo_2][0], node[tMTocMvQQgGCkj7QDHl3OA], [P], s[INITIALIZING]), reason [after recovery from shard_store]", "time_in_queue_millis": 858, "time_in_queue": "858ms" } ] } Cluster Reroute 说明: reroute命令允许我们人工的重新分配集群中的索引分片。 move:移动分片 cancel:取消分片分配 allocate_replica:分配副本 命令: POST /_cluster/reroute { "commands" : [ { "move" : { "index" : "test", "shard" : 0, "from_node" : "node1", "to_node" : "node2" } }, { "allocate_replica" : { "index" : "test", "shard" : 1, "node" : "node3" } } ] } Cluster Update Settings 说明: 更新集群的配置信息并返回执行更新完成的配置。 命令: PUT /_cluster/settings { "persistent" : { "indices.recovery.max_bytes_per_sec" : "50mb" } } PUT /_cluster/settings?flat_settings=true { "transient" : { "indices.recovery.max_bytes_per_sec" : "20mb" } } 返回值: { ... "persistent" : { }, "transient" : { "indices.recovery.max_bytes_per_sec" : "20mb" } } Nodes Stats 说明: _nodes/stats 展示集群中指定节点或所有节点的统计信息。 命令: GET /_nodes/stats GET /_nodes/nodeId1,nodeId2/stats 返回值: { "_nodes": { "total": 1, "successful": 1, "failed": 0 }, "cluster_name": "my-application", "nodes": { "YOIwJR1bTxG0iKHur_M-Rg": { "timestamp": 1545564028595, "name": "node-2", "transport_address": "127.0.0.1:9300", "host": "127.0.0.1", "ip": "127.0.0.1:9300", "roles": [ "master", "data", "ingest" ], "os": { "timestamp": 1545564028595, "cpu": { "percent": 12, "load_average": { "1m": 1.7255859375 } }, "mem": { "total_in_bytes": 8589934592, "free_in_bytes": 305328128, "used_in_bytes": 8284606464, "free_percent": 4, "used_percent": 96 }, "swap": { "total_in_bytes": 2147483648, "free_in_bytes": 1657012224, "used_in_bytes": 490471424 } } } } } Nodes Info 说明: _nodes展示集群中每个节点的信息。 命令: GET /_nodes GET /_nodes/nodeId1,nodeId2 指标包括settings, os, process, jvm, thread_pool, transport, http, plugins, ingest,indices 返回值: { "_nodes": { "total": 1, "successful": 1, "failed": 0 }, "cluster_name": "my-application", "nodes": { "YOIwJR1bTxG0iKHur_M-Rg": { "name": "node-2", "transport_address": "127.0.0.1:9300", "host": "127.0.0.1", "ip": "127.0.0.1", "version": "6.1.0", "build_hash": "c0c1ba0", "roles": [ "master", "data", "ingest" ], "os": { "refresh_interval_in_millis": 1000, "name": "Mac OS X", "arch": "x86_64", "version": "10.14.2", "available_processors": 4, "allocated_processors": 4 } } } } Nodes Feature Usage 说明: _nodes/usage展示集群各个节点各类服务调用次数 命令: GET _nodes/usage GET _nodes/nodeId1,nodeId2/usage 返回值: { "_nodes": { "total": 1, "successful": 1, "failed": 0 }, "cluster_name": "my_cluster", "nodes": { "pQHNt5rXTTWNvUgOrdynKg": { "timestamp": 1492553961812, "since": 1492553906606, "rest_actions": { "org.elasticsearch.rest.action.admin.cluster.RestNodesUsageAction": 1, "org.elasticsearch.rest.action.admin.indices.RestCreateIndexAction": 1, "org.elasticsearch.rest.action.document.RestGetAction": 1, "org.elasticsearch.rest.action.search.RestSearchAction": 19, "org.elasticsearch.rest.action.admin.cluster.RestNodesInfoAction": 36 } } } } Remote Cluster Info 说明: 返回集群的连接信息。 命令: GET /_remote/info Nodes hot_threads 说明: _nodes/hot_threads展示集群节点的热点线程信息。 命令: GET /_nodes/hot_threads GET /_nodes/{nodesIds}/hot_threads. 返回值: ::: {node-2}{YOIwJR1bTxG0iKHur_M-Rg}{UmVpL8zQTUa1tu6TtOhpyQ}{127.0.0.1}{127.0.0.1:9300} Hot threads at 2018-12-23T12:22:24.368Z, interval=500ms, busiestThreads=3, ignoreIdleThreads=true: Cluster Allocation Explain API 说明: 用于解释分片没有被分配的原因 命令: GET /_cluster/allocation/explain { "index": "myindex", "shard": 0, "primary": true } 返回值: { "index" : "idx", "shard" : 0, "primary" : true, "current_state" : "unassigned", "unassigned_info" : { "reason" : "INDEX_CREATED", "at" : "2017-01-04T18:08:16.600Z", "last_allocation_status" : "no" }, "can_allocate" : "no", "allocate_explanation" : "cannot allocate because allocation is not permitted to any of the nodes", "node_allocation_decisions" : [ { "node_id" : "8qt2rY-pT6KNZB3-hGfLnw", "node_name" : "node-0", "transport_address" : "127.0.0.1:9401", "node_attributes" : {}, "node_decision" : "no", "weight_ranking" : 1, "deciders" : [ { "decider" : "filter", "decision" : "NO", "explanation" : "node does not match index setting [index.routing.allocation.include] filters [_name:\"non_existent_node\"]" } ] } ] }

优秀的个人博客,低调大师

Spring Cloud的核心特性

SOA和微服务的区别 其实服务化架构已经可以解决大部分企业的需求了,那么我们为什么要研究微服务呢?先说说它们的区别; 微服务架构强调业务系统需要彻底的组件化和服务化,一个组件就是一个产品,可以独立对外提供服务微服务不再强调传统SOA架构里面比较重的ESB企业服务总线微服务强调每个微服务都有自己独立的运行空间,包括数据库资源。微服务架构本身来源于互联网的思路,因此组件对外发布的服务强调了采用HTTP Rest API的方式来进行微服务的切分粒度会更小总结:微服务架构是 SOA 架构思想的一种扩展,更加强调服务个体的独立性、拆分粒度更小。 为什么考虑Spring Cloud Spring Cloud来源于Spring,质量、稳定性、持续性都可以得到保证Spirng Cloud天然支持Spring Boot,更加便于业务落地。Spring Cloud发展非常的快,从16年开始接触的时候相关组件版本为1.x,到现在将要发布2.x系列Spring Cloud是Java领域最适合做微服务的框架。相比于其它框架,Spring Cloud对微服务周边环境的支持力度最大。对于中小企业来讲,使用门槛较低。Spring Cloud 是微服务架构的最佳落地方案 分布式/版本化配置服务注册和发现路由服务和服务之间的调用负载均衡断路器分布式消息传递这些特性都是由不同的组件来完成的,在架构的演进过程中扮演着重要的角色。

优秀的个人博客,低调大师

Spring Reactor 项目核心库

Reactor Core Non-Blocking Reactive Streams Foundation for the JVM both implementing a Reactive Extensions inspired API and efficient event streaming support. Getting it Reactor 3 requires Java 8 or + to run. With Gradle from repo.spring.io or Maven Central repositories (stable releases only): repositories { // maven { url 'http://repo.spring.io/snapshot' } maven { url 'http://repo.spring.io/milestone' } mavenCentral() } dependencies { //compile "io.projectreactor:reactor-core:3.1.4.RELEASE" //testCompile("io.projectreactor:reactor-test:3.1.4.RELEASE") compile "io.projectreactor:reactor-core:3.2.0.M1" testCompile("io.projectreactor:reactor-test:3.2.0.M1") } See the reference documentation for more information on getting it (eg. using Maven, or on how to get milestones and snapshots). Note about Android support: Reactor 3 doesn't officially support nor target Android. However it should work fine with Android SDK 26 (Android O) and above. See thecomplete note in the reference guide. Getting Started New to Reactive Programming or bored of reading already ? Try the Introduction to Reactor Core hands-on ! If you are familiar with RxJava or if you want to check more detailled introduction, be sure to checkhttps://www.infoq.com/articles/reactor-by-example ! Flux A Reactive Streams Publisher with basic flow operators. Static factories on Flux allow for source generation from arbitrary callbacks types. Instance methods allows operational building, materialized on each Flux#subscribe(), Flux#subscribe() or multicasting operations such as Flux#publish and Flux#publishNext. <img src="https://raw.githubusercontent.com/reactor/reactor-core/v3.1.3.RELEASE/src/docs/marble/flux.png" width="500"> Flux in action : Flux.fromIterable(getSomeLongList()) .mergeWith(Flux.interval(100)) .doOnNext(serviceA::someObserver) .map(d -> d * 2) .take(3) .onErrorResumeWith(errorHandler::fallback) .doAfterTerminate(serviceM::incrementTerminate) .subscribe(System.out::println); Mono A Reactive Streams Publisher constrained to ZERO or ONE element with appropriate operators. Static factories on Mono allow for deterministic zero or one sequence generation from arbitrary callbacks types. Instance methods allows operational building, materialized on each Mono#subscribe() or Mono#get() eventually called. <img src="https://raw.githubusercontent.com/reactor/reactor-core/v3.1.3.RELEASE/src/docs/marble/mono.png" width="500"> Mono in action : Mono.fromCallable(System::currentTimeMillis) .flatMap(time -> Mono.first(serviceA.findRecent(time), serviceB.findRecent(time))) .timeout(Duration.ofSeconds(3), errorHandler::fallback) .doOnSuccess(r -> serviceM.incrementSuccess()) .subscribe(System.out::println); Blocking Mono result : Tuple2<Long, Long> nowAndLater = Mono.zip( Mono.just(System.currentTimeMillis()), Flux.just(1).delay(1).map(i -> System.currentTimeMillis())) .block(); Schedulers Reactor uses a Scheduler as a contract for arbitrary task execution. It provides some guarantees required by Reactive Streams flows like FIFO execution. You can use or create efficient schedulers to jump thread on the producing flows (subscribeOn) or receiving flows (publishOn): Mono.fromCallable( () -> System.currentTimeMillis() ) .repeat() .publishOn(Schedulers.single()) .log("foo.bar") .flatMap(time -> Mono.fromCallable(() -> { Thread.sleep(1000); return time; }) .subscribeOn(Schedulers.parallel()) , 8) //maxConcurrency 8 .subscribe(); ParallelFlux ParallelFlux can starve your CPU's from any sequence whose work can be subdivided in concurrent tasks. Turn back into a Flux with ParallelFlux#sequential(), an unordered join or use abitrary merge strategies via 'groups()'. Mono.fromCallable( () -> System.currentTimeMillis() ) .repeat() .parallel(8) //parallelism .runOn(Schedulers.parallel()) .doOnNext( d -> System.out.println("I'm on thread "+Thread.currentThread()) ) .subscribe() Custom sources : Flux.create and FluxSink, Mono.create and MonoSink To bridge a Subscriber or Processor into an outside context that is taking care of producing non concurrently, use Flux#create, Mono#create. Flux.create(sink -> { ActionListener al = e -> { sink.next(textField.getText()); }; // without cancellation support: button.addActionListener(al); // with cancellation support: sink.onCancel(() -> { button.removeListener(al); }); }, // Overflow (backpressure) handling, default is BUFFER FluxSink.OverflowStrategy.LATEST) .timeout(3) .doOnComplete(() -> System.out.println("completed!")) .subscribe(System.out::println) The Backpressure Thing Most of this cool stuff uses bounded ring buffer implementation under the hood to mitigate signal processing difference between producers and consumers. Now, the operators and processors or any standard reactive stream component working on the sequence will be instructed to flow in when these buffers have free room AND only then. This means that we make sure we both have a deterministic capacity model (bounded buffer) and we never block (request more data on write capacity). Yup, it's not rocket science after all, the boring part is already being worked by us in collaboration with Reactive Streams Commons on going research effort. What's more in it ? "Operator Fusion" (flow optimizers), health state observers, helpers to build custom reactive components, bounded queue generator, hash-wheel timer, converters from/to Java 9 Flow, Publisher and Java 8 CompletableFuture. The repository contains a reactor-test project with test features like the StepVerifier. Reference Guide http://projectreactor.io/docs/core/release/reference/docs/index.html Javadoc https://projectreactor.io/docs/core/release/api/ Getting started with Flux and Mono https://github.com/reactor/lite-rx-api-hands-on Reactor By Example https://www.infoq.com/articles/reactor-by-example Head-First Spring & Reactor https://github.com/reactor/head-first-reactive-with-spring-and-reactor/ Beyond Reactor Core Everything to jump outside the JVM with the non-blocking drivers from Reactor Netty. Reactor Addons provide for adapters and extra operators for Reactor 3. Powered by Reactive Streams Commons Licensed under Apache Software License 2.0 Sponsored by Pivotal

优秀的个人博客,低调大师

Hadoop核心-HDFS读写流程

HDFS写数据步骤: 1. client发起文件上传请求,通过RPC与NameNode建立连接,NameNode检查目标文件是否已经存在,父目录是否存在,并检查用户是否有相应的权限,若检查通过, 会为该文件创建一个新的记录,否则的话文件创建失败,客户端得到异常信息, 2. client通过请求NameNode,第一个block应该传输到哪些DataNode服务器上; 3. NameNode根据配置文件中指定的备份(replica)数量及机架感知原理进行文件分配,返回可用的DataNode的地址 以三台DataNode为例:A B C 注: Hadoop在设计时考虑到数据的安全与高效,数据文件默认在HDFS上存放三份,存储策略为:第一个备份放在客户端相同的datanode上(若客户端在集群外运行, 就随机选取一个datanode来存放第一个replica),第二个replica放在与第一个replica不同机架的一个随机datanode上,第三个replica放在与第二个repli ca相同机架的随机datanode上,如果replica数大于三,则随后的replica在集群中随机存放,Hadoop会尽量避免过多的replica存放在同一个机架上.选取repl ica存放在同一个机架上.(Hadoop 1.x以后允许replica是可插拔的,意思是说可以定制自己需要的replica分配策略) 4. client请求3台的DataNode的一台A上传数据,(本质是一个RPC调用,建立pipeline),A收到请求会继续调用B,然后B调用C,将整个pipeline建立完成后,逐 级返回client; 5. client开始往A上传第一个block(先从磁盘读取数据放到一个本地内存缓存),以packet为单位(默认 64K),A收到一个packet就会传给B,B传递给C;A每传一 个packet会放入一个应答队列等待应答. 注: 如果某个datanode在写数据的时候宕掉了下面这些对用户透明的步骤会被执行: 1) 管道线关闭,所有确认队列上的数据会被挪到数据队列的首部重新发送,这样也就确保管道线中宕掉的datanode下流的datanode不会因为宕掉的datanode而丢 失数据包 2) 在还在正常运行datanode上的当前block上做一个标志,这样当宕掉的datanode重新启动以后namenode就会知道该datanode上哪个block是刚才宕机残留下 的局部损坏block,从而把他删除掉 3) 已经宕掉的datanode从管道线中被移除,未写完的block的其他数据继续呗写入到其他两个还在正常运行的datanode中,namenode知道这个block还处在und er-replicated状态(即备份数不足的状态)下,然后它会安排一个新的replica从而达到要求的备份数,后续的block写入方法同前面正常时候一样 4) 有可能管道线中的多个datanode宕掉(一般这种情况很少),但只要dfs.relication.min(默认值为1)个replica被创建,我么就认为该创建成功了,剩余的re lica会在以后异步创建以达到指定的replica数. 6. 数据被分割成一个个packet数据包在pipeline上一次传输,在pipeline反方向上,逐个发送ack(命令正确应答),最终由pipeline中第一个DataNode节点A 将pipeline ack发送给client 7. 当一个block传输完成后,client再次发送请求NameNode上传第二个block到服务器 HDFS读数据步骤: 1. Client向NameNode发起RPC请求,来确定请求文件block所在的位置; 2. NameNode会视情况返回文件的部分或者全部block列表,对于每个block,NameNode都会返回含有该block副本的DataNode地址 3. 这些返回的DN地址,会按照集群拓扑结构得出DataNode与客户端的距离,然后进行排序,排序两个规则:网络拓扑结构中距离Client的排在前;心跳机制中超时汇 报的DN状态为STALE,这样的排在后 4. Clietn选取排序靠前的DataNode来读取block,如果客户端本身就是DataNode,那么将从本地直接获取数据 5. 底层本质是建立Socket Stream(FSDataInputStream) ,重复调用父类DataInputStream的read方法,知道这个块上的数据读取完毕; 6. 当读完列表的block后,若文件读取还没有结束,客户端会继续想NameNode获取下一批的block列表 7. 读取完一个Block都会进行checksum验证,如果读取DataNode时出现错误,客户端会通知NameNode,然后再从下一个拥有该block副本的DataNode继续读取 注: 如果在读取过程中DFSInputStream检测到block错误,DFSInputStream也会检查从datanode读取来的数据的校验和,如果发现有数据损坏,它会把坏掉的bl ock报告给namenode同时重新读取其他datanode上的其他block备份 8. read方法是并行的读取block信息,不是一块一块的读取,NameNode只是返回Client请求包含块的DataNode地址,并不是返回请求块的数据; 9. 最终读取哎所有的block会合并成一个完整的最终文件

优秀的个人博客,低调大师

Spark 核心之 Driver 原理剖析

摘要:如果把 Spark 应用比作一个人,Driver 就是它的大脑。从 spark-submit 敲下回车的那一刻起,SparkContext 初始化、DAGScheduler 切分 Stage、TaskScheduler 分发 Task、SchedulerBackend 与集群通信、SparkEnv 管理运行时环境——所有这些都在 Driver 内部精密协作。本文从 Driver 内部架构全景、SparkContext 初始化链路、三大调度器协作模型、SparkEnv 七大组件、Driver 生命周期六个维度,配合 1 张原创深色架构图和完整源码级分析,带你彻底看清 Driver 的内部世界。

优秀的个人博客,低调大师

Spark 核心之 ClusterManager 原理剖析

摘要:你有没有好奇过——为什么 --master spark:// 和 --master yarn 切换如此丝滑?这背后是 Spark 设计中最精妙的抽象层之一——ClusterManager。它通过 ExternalClusterManager 接口 + ServiceLoader 机制,将 Standalone、YARN、Kubernetes 三种资源管理器统一为可插拔的适配器。本文从接口设计、Master URL 路由、三种实现源码对比、自定义扩展四个维度,配合 1 张原创深色架构图 + 完整源码分析,带你彻底理解 Spark 资源调度层的设计哲学。

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长,特征是大鼻子、头戴帽子、身穿背带裤,还留着胡子。与他的双胞胎兄弟路易基一起,长年担任任天堂的招牌角色。

腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题,腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构,目前腾讯云软件源站支持公网访问和内网访问。

Spring

Spring

Spring框架(Spring Framework)是由Rod Johnson于2002年提出的开源Java企业级应用框架,旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念,提供核心容器、应用上下文、数据访问集成等模块,支持整合Hibernate、Struts等第三方框架,其适用范围不仅限于服务器端开发,绝大多数Java应用均可从中受益。

WebStorm

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源,继承了IntelliJ IDEA强大的JS部分的功能。

用户登录
用户注册