主頁 > 前端設計 > es客戶端執行緒池配置

es客戶端執行緒池配置

2020-11-03 02:16:32 前端設計

業務背景

原大資料團隊不再維護DB資料同步至es的服務，由我們業務團隊自己維護，實作方案：使用canal監聽DB binlog，將資料寫入es

問題描述

為啥要看es執行緒池呢？因為線上突然瘋狂報錯es執行緒池被打滿，但竟然看不懂該如何修改es執行緒池配置，線上例外堆疊如下：

EsRejectedExecutionException[rejected execution of org.elasticsearch.transport.TransportService$7@4d334adf on EsThreadPoolExecutor[bulk, queue capacity = 2000, org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@3c0a3704[Running, pool size = 16, active threads = 16, queued tasks = 2009, completed tasks = 284167676]]]
	at org.elasticsearch.common.util.concurrent.EsAbortPolicy.rejectedExecution(EsAbortPolicy.java:50)
	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
	at org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor.doExecute(EsThreadPoolExecutor.java:94)
	at org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor.execute(EsThreadPoolExecutor.java:89)
	at org.elasticsearch.transport.TransportService.sendLocalRequest(TransportService.java:614)
	at org.elasticsearch.transport.TransportService.access$000(TransportService.java:73)
	at org.elasticsearch.transport.TransportService$3.sendRequest(TransportService.java:133)
	at org.elasticsearch.transport.TransportService.sendRequestInternal(TransportService.java:562)
	at org.elasticsearch.transport.TransportService.sendRequest(TransportService.java:495)
	at org.elasticsearch.transport.TransportService.sendRequest(TransportService.java:483)
	at org.elasticsearch.action.support.replication.TransportReplicationAction$ReroutePhase.performAction(TransportReplicationAction.java:751)
	at org.elasticsearch.action.support.replication.TransportReplicationAction$ReroutePhase.performLocalAction(TransportReplicationAction.java:670)
	at org.elasticsearch.action.support.replication.TransportReplicationAction$ReroutePhase.doRun(TransportReplicationAction.java:658)
	at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:37)
	at org.elasticsearch.action.support.replication.TransportReplicationAction.doExecute(TransportReplicationAction.java:147)
	at org.elasticsearch.action.support.replication.TransportReplicationAction.doExecute(TransportReplicationAction.java:93)
	at org.elasticsearch.action.support.TransportAction$RequestFilterChain.proceed(TransportAction.java:170)
	at org.elasticsearch.action.support.TransportAction.execute(TransportAction.java:142)
	at org.elasticsearch.action.support.replication.TransportReplicationAction$OperationTransportHandler.messageReceived(TransportReplicationAction.java:222)
	at org.elasticsearch.action.support.replication.TransportReplicationAction$OperationTransportHandler.messageReceived(TransportReplicationAction.java:219)
	at org.elasticsearch.transport.RequestHandlerRegistry.processMessageReceived(RequestHandlerRegistry.java:69)
	at org.elasticsearch.transport.TcpTransport$RequestHandler.doRun(TcpTransport.java:1488)
	at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:37)
	at org.elasticsearch.common.util.concurrent.EsExecutors$1.execute(EsExecutors.java:109)
	at org.elasticsearch.transport.TcpTransport.handleRequest(TcpTransport.java:1445)
	at org.elasticsearch.transport.TcpTransport.messageReceived(TcpTransport.java:1329)
	at org.elasticsearch.transport.netty4.Netty4MessageChannelHandler.channelRead(Netty4MessageChannelHandler.java:74)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:341)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:293)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:280)
	at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:396)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:248)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:341)
	at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:341)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1334)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:926)
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:129)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:642)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysPlain(NioEventLoop.java:527)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:481)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:441)
	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
	at java.lang.Thread.run(Thread.java:745)

問題分析

分析個錘子哦，很明顯es執行緒池被打滿了，且執行緒池配置相當過分，只有16的size，es的執行緒池默認值也是很過分啦
服務中es客戶端初始化代碼

InetSocketTransportAddress[] addresses = new InetSocketTransportAddress[hostNames.length];
for (int i = 0; i < hostNames.length; i++) {
    addresses[i] = new InetSocketTransportAddress(
            new InetSocketAddress(hostNames[i], esOutputConfig.getPort()));
}
Settings.Builder builder = Settings.builder()
        .put("cluster.name", esOutputConfig.getClusterName());
Settings settings = builder.build();
transportClient = new PreBuiltTransportClient(settings);
transportClient.addTransportAddresses(addresses);
transportClients.add(transportClient);

問題解決

調大執行緒池配置，
起初由于不知道怎樣配置es執行緒池數量，且查資料也沒查到，臨時方案：增加es客戶端數量（變相調增執行緒池大小），

查看es代碼，增加配置如下

Settings.Builder builder = Settings.builder()
        .put("cluster.name", esOutputConfig.getClusterName());
if (taskConfig.getName().contains("order_detail_shard")
        || taskConfig.getName().contains("order_shard")) {
    builder.put("thread_pool.bulk.size", 100);

看到V1版本就應該知道結論了吧，沒錯，沒生效，線上依然瘋狂報錯執行緒被打滿，size依然是16

查看es客戶端執行緒池原始碼

執行緒池代碼，N多個執行緒池，重點在于要搞懂es配置Settings的使用方法

public ThreadPool(final Settings settings, final ExecutorBuilder<?>... customBuilders) {
    super(settings);
    assert Node.NODE_NAME_SETTING.exists(settings);
    final Map<String, ExecutorBuilder> builders = new HashMap<>();
    final int availableProcessors = EsExecutors.boundedNumberOfProcessors(settings);
    final int halfProcMaxAt5 = halfNumberOfProcessorsMaxFive(availableProcessors);
    final int halfProcMaxAt10 = halfNumberOfProcessorsMaxTen(availableProcessors);
    final int genericThreadPoolMax = boundedBy(4 * availableProcessors, 128, 512);
    builders.put(Names.GENERIC, new ScalingExecutorBuilder(Names.GENERIC, 4, genericThreadPoolMax, TimeValue.timeValueSeconds(30)));
    builders.put(Names.INDEX, new FixedExecutorBuilder(settings, Names.INDEX, availableProcessors, 200));
    builders.put(Names.BULK, new FixedExecutorBuilder(settings, Names.BULK, availableProcessors, 200)); // now that we reuse 
	...
}

我們業務場景主要使用的寫入執行緒池，BULK執行緒池：FixedExecutorBuilder

FixedExecutorBuilder(final Settings settings, final String name, final int size, final int queueSize) {
    this(settings, name, size, queueSize, "thread_pool." + name);
}

size為availableProcessors引數

availableProcessors引數取值邏輯

public static final Setting<Integer> PROCESSORS_SETTING =
Setting.intSetting("processors", Math.min(32, Runtime.getRuntime().availableProcessors()), 1, Property.NodeScope);
public static int boundedNumberOfProcessors(Settings settings) {
    /* This relates to issues where machines with large number of cores
     * ie. >= 48 create too many threads and run into OOM see #3478
     * We just use an 32 core upper-bound here to not stress the system
     * too much with too many created threads */
    return PROCESSORS_SETTING.get(settings);
}

初始化Settings，該Settings構造器引數含義：

  1. 配置key：processors
  2. 默認值提供者函式：與入參無關，直接回傳默認值
  3. 最小值提供者函式：將入參轉為Integer，并且入參在最小值-最大值之間，否則拋出引數例外
  4. 屬性
public static Setting<Integer> intSetting(String key, int defaultValue, int minValue, Property... properties) {
    return new Setting<>(key, (s) -> Integer.toString(defaultValue), (s) -> parseInt(s, minValue, key), properties);
}

即availableProcessors的取值邏輯為：min(32與cpu核心數)取最小值作為默認值，最小值為硬編碼1

FixedExecutorBuilder構造器

  1. settings：構造客戶端時由開發者自定義的配置資訊
  2. name：配置名稱，例如：Names.BULK
  3. size：大小
  4. prefix：配置前綴
public FixedExecutorBuilder(final Settings settings, final String name, final int size, final int queueSize, final String prefix) {
    super(name);
    final String sizeKey = settingsKey(prefix, "size");
    this.sizeSetting =
        new Setting<>(
            sizeKey,
            s -> Integer.toString(size),
            s -> Setting.parseInt(s, 1, applyHardSizeLimit(settings, name), sizeKey),
            Setting.Property.NodeScope);
    final String queueSizeKey = settingsKey(prefix, "queue_size");
    this.queueSizeSetting =
        Setting.intSetting(queueSizeKey, queueSize, Setting.Property.NodeScope);
}

暫不關心佇列配置，sizeSetting配置構造器引數含義：

  1. sizeKey：配置key
  2. 默認值提供者函式：與入參無關，直接回傳默認值，即：availableProcessors
  3. parser：配置決議函式，根據配置的字串決議出結果型別資料，
  4. 屬性

配置決議函式parser

  1. 最小值：硬編碼為1
  2. 最大值：applyHardSizeLimit，如果配置名稱為：BULK、INDEX，則使用availableProcessors+1（即最大值實際為availableProcessors），否則為Integer最大值
  3. sizeKey：前綴+size拼接，實際規則：“thread_pool.”+name+“size”，例如：“thread_pool.bulk.size”

構建執行緒池：FixedExecutorBuilder
根據開發者提供的settings配置構建執行緒池

final ExecutorBuilder.ExecutorSettings executorSettings = entry.getValue().getSettings(settings);
final ExecutorHolder executorHolder = entry.getValue().build(executorSettings, threadContext);

獲取配置：org.elasticsearch.threadpool.ExecutorBuilder#getSettings，執行緒池構建者實作類：org.elasticsearch.threadpool.FixedExecutorBuilder#getSettings

@Override
FixedExecutorSettings getSettings(Settings settings) {
    final String nodeName = Node.NODE_NAME_SETTING.get(settings);
    final int size = sizeSetting.get(settings);
    final int queueSize = queueSizeSetting.get(settings);
    return new FixedExecutorSettings(nodeName, size, queueSize);
}

Setting獲取配置邏輯

public T get(Settings settings) {
    String value = getRaw(settings);
    try {
        return parser.apply(value);
    } ...
}

獲取配置值

  1. 配置key為：sizeSetting.getKey()，即：thread_pool.bulk.size
  2. 默認值提供者函式：與入參無關，直接回傳默認值，即：availableProcessors，有效cpu核心數
// 1. settings：開發者自定義引數配置
// 2. getKey()：當前setting實體的key，例如：sizeSetting
// 3. defaultValue：當前setting實體的默認值提供者函式，例如：sizeSetting
public String getRaw(Settings settings) {
    checkDeprecation(settings);
    return settings.get(getKey(), defaultValue.apply(settings));
}

此時不會回傳默認值，因為我們配置了thread_pool.bulk.size=100

配置決議函式決議配置值，配置最小值為1，最大值為有效cpu核心數量，即：availableProcessors+1，線上機器為16核，最大值為：16

s -> Setting.parseInt(s, 1, applyHardSizeLimit(settings, name), sizeKey);
// applyHardSizeLimit
private int applyHardSizeLimit(final Settings settings, final String name) {
    if (name.equals(ThreadPool.Names.BULK) || name.equals(ThreadPool.Names.INDEX)) {
        return 1 + EsExecutors.boundedNumberOfProcessors(settings);
    } else {
        return Integer.MAX_VALUE;
    }
}

因此修改的配置沒有生效，因為100（thread_pool.bulk.size）>16（有效cpu核心數）直接拋出引數例外

總結

es客戶端配置，對于bulk、index兩個執行緒池的大小如果想要修改不能單單修改執行緒池配置，還需要一并修改jvm所在服務器的cpu有效核心數配置，修改后的代碼如下，問題解決

Settings.Builder builder = Settings.builder()
        .put("cluster.name", esOutputConfig.getClusterName());
if (taskConfig.getName().contains("order_detail_shard")
        || taskConfig.getName().contains("order_shard")) {
    builder.put("thread_pool.bulk.size", 100)
            .put("thread_pool.bulk.queue_size", 100)
            .put("processors", 200);

轉載請註明出處，本文鏈接：https://www.uj5u.com/qianduan/200368.html

標籤：其他

上一篇：哈西路由跳轉實作

下一篇：Kudu

標籤雲
其他(157675) Python(38076) JavaScript(25376) Java(17977) C(15215) 區塊鏈(8255) C＃(7972) AI(7469) 爪哇(7425) MySQL(7132) html(6777) 基礎類(6313) sql(6102) 熊猫(6058) PHP(5869) 数组(5741) R(5409) Linux(5327) 反应(5209) 腳本語言(PerlPython)(5129) 非技術區(4971) Android(4554) 数据框(4311) css(4259) 节点.js(4032) C語言(3288) json(3245) 列表(3129) 扑(3119) C++語言(3117) 安卓(2998) 打字稿(2995) VBA(2789) Java相關(2746) 疑難問題(2699) 细绳(2522) 單片機工控(2479) iOS(2429) ASP.NET(2402) MongoDB(2323) 麻木的(2285) 正则表达式(2254) 字典(2211) 循环(2198) 迅速(2185) 擅长(2169) 镖(2155) 功能(1967) .NET技术(1958) Web開發(1951) python-3.x(1918) HtmlCss(1915) 弹簧靴(1913) C++(1909) xml(1889) PostgreSQL(1872) .NETCore(1853) 谷歌表格(1846) Unity3D(1843) for循环(1842)

熱門瀏覽
  • vue移動端上拉加載

    可能做得過于簡單或者比較low，請各位大佬留情，一起探討技術 ......

    uj5u.com 2020-09-10 04:38:07 more
  • 優美網站首頁，頂部多層導航

    一個個人用的瀏覽器首頁，可以把一下常用的網站放在這里，平常打開會比較方便。 第一步，HTML代碼 <script src=https://www.cnblogs.com/szharf/p/"js/jquery-3.4.1.min.js"></script> <div id="navigate"> <ul> <li class="labels labels_1"> ......

    uj5u.com 2020-09-10 04:38:47 more
  • 頁面為要加<!DOCTYPE html>

    最近因為寫一個js函式，需要用到$(window).height(); 由于手寫demo的時候，過于自信，其實對前端方面的認識也不夠體系，用文本檔案直接敲出來的html代碼，第一行沒有加上<!DOCTYPE html> 導致了$(window).height();的結果直接是整個document的高 ......

    uj5u.com 2020-09-10 04:38:52 more
  • WordPress網站程式手動升級要做好資料備份

    WordPress博客網站程式在進行升級前，必須要做好網站資料的備份，這個問題良家佐言是遇見過的；在剛開始接觸WordPress博客程式的時候，因為升級問題和博客網站的修改的一些嘗試，良家佐言是吃盡了苦頭。因為購買的是西部數碼的空間和域名，每當佐言把自己的WordPress博客網站搞到一塌糊涂的時候 ......

    uj5u.com 2020-09-10 04:39:30 more
  • WordPress程式不能升級為5.4.2版本的原因

    WordPress是一款個人博客系統，受到英文博客愛好者和中文博客愛好者的追捧，并逐步演化成一款內容管理系統軟體；它是使用PHP語言和MySQL資料庫開發的,用戶可以在支持PHP和MySQL資料庫的服務器上使用自己的博客。每一次WordPress程式的更新，就會牽動無數WordPress愛好者的心， ......

    uj5u.com 2020-09-10 04:39:49 more
  • 使用CSS3的偽元素進行首字母下沉和首行改變樣式

    網頁中常見的一種效果，首字改變樣式或者首行改變樣式，效果如下圖。 代碼： <!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, ......

    uj5u.com 2020-09-10 04:40:09 more
  • 關于a標簽的講解

    什么是a標簽？ <a> 標簽定義超鏈接，用于從一個頁面鏈接到另一個頁面。 <a> 元素最重要的屬性是 href 屬性，它指定鏈接的目標。 a標簽的語法格式：<a href=https://www.cnblogs.com/summerxbc/p/"指定要跳轉的目標界面的鏈接">需要展示給用戶看見的內容</a> a標簽 在所有瀏覽器中，鏈接的默認外觀如下： 未被訪問的鏈接帶 ......

    uj5u.com 2020-09-10 04:40:11 more
  • 前端輪播圖

    在需要輪播的頁面是引入swiper.min.js和swiper.min.css swiper.min.js地址： 鏈接：https://pan.baidu.com/s/15Uh516YHa4CV3X-RyjEIWw 提取碼：4aks swiper.min.css地址 鏈接：https://pan.b ......

    uj5u.com 2020-09-10 04:40:13 more
  • 如何設定html中的背景圖片（全屏顯示，且不拉伸）

    1 <style>2 body{background-image:url(https://uploadbeta.com/api/pictures/random/?key=BingEverydayWallpaperPicture); 3 background-size:cover;background ......

    uj5u.com 2020-09-10 04:40:16 more
  • Java學習——HTML詳解（上）

    HTML詳解 初識HTML Hyper Text Markup Language（超文本標記語言） 1 <!--DOCTYPE:告訴瀏覽器我們要使用什么規范--> 2 <!DOCTYPE html> 3 <html lang="en"> 4 <head> 5 <!--meta 描述性的標簽，描述一些 ......

    uj5u.com 2020-09-10 04:40:33 more
最新发布
  • 我的第一個NPM包:panghu-planebattle-esm(胖虎飛機大戰)使用說明

    好家伙,我的包終于開發完啦 歡迎使用胖虎的飛機大戰包!! 為你的主頁添加色彩 這是一個有趣的網頁小游戲包,使用canvas和js開發 使用ES6模塊化開發 效果圖如下: (覺得圖片太sb的可以自己改) 代碼已開源!! Git: https://gitee.com/tang-and-han-dynas ......

    uj5u.com 2023-04-20 07:59:23 more
  • 生產事故-走近科學之消失的JWT

    入職多年，面對生產環境，盡管都是小心翼翼，慎之又慎，還是難免捅出簍子。輕則滿頭大汗，面紅耳赤。重則系統停擺，損失資金。每一個生產事故的背后，都是寶貴的經驗和教訓，都是專案成員的血淚史。為了更好地防范和遏制今后的各類事故，特開此專題，長期更新和記錄大大小小的各類事故。有些是親身經歷，有些是經人耳傳口授 ......

    uj5u.com 2023-04-18 07:55:04 more
  • 記錄--Canvas實作打飛字游戲

    這里給大家分享我在網上總結出來的一些知識，希望對大家有所幫助 打開游戲界面，看到一個畫面簡潔、卻又富有挑戰性的游戲。螢屏上，有一個白色的矩形框，里面不斷下落著各種單詞，而我需要迅速地輸入這些單詞。如果我輸入的單詞與螢屏上的單詞匹配，那么我就可以獲得得分；如果我輸入的單詞錯誤或者時間過長，那么我就會輸 ......

    uj5u.com 2023-04-04 08:35:30 more
  • 了解 HTTP 看這一篇就夠

    在學習網路之前，了解它的歷史能夠幫助我們明白為何它會發展為如今這個樣子，引發探究網路的興趣。下面的這張圖片就展示了“互聯網”誕生至今的發展歷程。 ......

    uj5u.com 2023-03-16 11:00:15 more
  • 藍牙-低功耗中心設備

    //11.開啟藍牙配接器 openBluetoothAdapter //21.開始搜索藍牙設備 startBluetoothDevicesDiscovery //31.開啟監聽搜索藍牙設備 onBluetoothDeviceFound //30.停止監聽搜索藍牙設備 offBluetoothDevi ......

    uj5u.com 2023-03-15 09:06:45 more
  • canvas畫板（滑鼠和觸摸）

    <!DOCTYPE html> <html> <head> <meta charset="utf-8"> <title>canves</title> <style> #canvas { cursor:url(../images/pen.png),crosshair; } #canvasdiv{ bo ......

    uj5u.com 2023-02-15 08:56:31 more
  • 手機端H5 實作自定義拍照界面

    手機端 H5 實作自定義拍照界面也可以使用 MediaDevices API 和 <video> 標簽來實作，和在桌面端做法基本一致。 首先，使用 MediaDevices.getUserMedia() 方法獲取攝像頭媒體流，并將其傳遞給 <video> 標簽進行渲染。 接著，使用 HTML 的 < ......

    uj5u.com 2023-01-12 07:58:22 more
  • 記錄--短視頻滑動播放在 H5 下的實作

    這里給大家分享我在網上總結出來的一些知識，希望對大家有所幫助 短視頻已經無數不在了，但是主體還是使用 app 來承載的。本文講述 H5 如何實作 app 的視頻滑動體驗。 無聲勝有聲，一圖頂百辯，且看下圖： 網址鏈接（需在微信或者手Q中瀏覽） 從上圖可以看到，我們主要實作的功能也是本文要講解的有： ......

    uj5u.com 2023-01-04 07:29:05 more
  • 一文讀懂 HTTP/1 HTTP/2 HTTP/3

    從 1989 年萬維網（www）誕生，HTTP（HyperText Transfer Protocol）經歷了眾多版本迭代，WebSocket 也在期間萌芽。1991 年 HTTP0.9 被發明。1996 年出現了 HTTP1.0。2015 年 HTTP2 正式發布。2020 年 HTTP3 或能正... ......

    uj5u.com 2022-12-24 06:56:02 more
  • 【HTML基礎篇002】HTML之form表單超詳解

    ??一、form表單是什么

    ??二、form表單的屬性

    ??三、input中的各種Type屬性值

    ??四、標簽 ......

    uj5u.com 2022-12-18 07:17:06 more

Copyright © 2010-2020 有解無憂