{"id":10021245,"date":"2026-02-10T12:08:49","date_gmt":"2026-02-10T04:08:49","guid":{"rendered":"https:\/\/itmax.cn\/?p=10021245"},"modified":"2026-02-10T12:08:49","modified_gmt":"2026-02-10T04:08:49","slug":"flink-clickhouse-sink%ef%bc%9a%e7%94%9f%e4%ba%a7%e7%ba%a7%e9%ab%98%e5%8f%af%e7%94%a8%e5%86%99%e5%85%a5%e6%96%b9%e6%a1%88%ef%bd%9c%e5%be%97%e7%89%a9%e6%8a%80%e6%9c%af","status":"publish","type":"post","link":"https:\/\/itmax.cn\/?p=10021245","title":{"rendered":"Flink ClickHouse Sink\uff1a\u751f\u4ea7\u7ea7\u9ad8\u53ef\u7528\u5199\u5165\u65b9\u6848\uff5c\u5f97\u7269\u6280\u672f"},"content":{"rendered":"<h2>\u4e00\u3001\u80cc\u666f\u4e0e\u75db\u70b9<\/h2>\n<h3>\u4e1a\u52a1\u573a\u666f<\/h3>\n<p>\u5728\u5b9e\u65f6\u5927\u6570\u636e\u5904\u7406\u573a\u666f\u4e2d\uff0cFlink + ClickHouse \u7684\u7ec4\u5408\u88ab\u5e7f\u6cdb\u5e94\u7528\u4e8e\uff1a<\/p>\n<ul>\n<li><strong>\u65e5\u5fd7\u5904\u7406\uff1a<\/strong> \u6d77\u91cf\u5e94\u7528\u65e5\u5fd7\u5b9e\u65f6\u5199\u5165\u5206\u6790\u5e93\u3002<\/li>\n<li><strong>\u76d1\u63a7\u5206\u6790\uff1a<\/strong> \u4e1a\u52a1\u6307\u6807\u3001APM \u6570\u636e\u7684\u5b9e\u65f6\u805a\u5408\u3002<\/li>\n<\/ul>\n<p>\u8fd9\u4e9b\u573a\u666f\u7684\u5171\u540c\u7279\u70b9\uff1a<\/p>\n<ul>\n<li><strong>\u6570\u636e\u91cf\u5927<\/strong>\uff1a\u767e\u4e07\u7ea7 TPS\uff0c\u5cf0\u503c\u53ef\u8fbe\u5343\u4e07\u7ea7\u3002<\/li>\n<li><strong>\u5199\u5165\u5ef6\u8fdf\u654f\u611f\uff1a<\/strong> \u9700\u8981\u79d2\u7ea7\u53ef\u89c1\u3002<\/li>\n<li><strong>\u6570\u636e\u51c6\u786e\u6027\u8981\u6c42\u9ad8<\/strong>\uff1a\u4e0d\u5141\u8bb8\u6570\u636e\u4e22\u5931\u3002<\/li>\n<li><strong>\u591a\u8868\u5199\u5165\uff1a<\/strong> \u4e0d\u540c\u6570\u636e\u6839\u636e\u5206\u8868\u7b56\u7565\u5199\u5165\u4e0d\u540c\u7684\u8868\u3002<\/li>\n<\/ul>\n<h3>\u5f00\u6e90 Flink ClickHouse Sink \u7684\u75db\u70b9<\/h3>\n<p>Flink \u5b98\u65b9\u63d0\u4f9b\u7684 ClickHouse Sink\uff08flink-connector-jdbc\uff09\u5728\u751f\u4ea7\u73af\u5883\u4e2d\u5b58\u5728\u4ee5\u4e0b\u4e25\u91cd\u95ee\u9898\uff1a<\/p>\n<h4>\u75db\u70b9\u4e00\uff1a\u7f3a\u4e4f\u57fa\u4e8e\u6570\u636e\u91cf\u7684\u6512\u6279\u673a\u5236<\/h4>\n<p><strong>\u95ee\u9898\u8868\u73b0\uff1a<\/strong><\/p>\n<pre><code>\/\/ Flink \u5b98\u65b9 JDBC Sink \u7684\u5b9e\u73b0\npublic\u00a0class\u00a0JdbcSink&lt;T&gt;\u00a0extends\u00a0RichSinkFunction&lt;T&gt; {\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0int\u00a0batchSize; \u00a0\/\/ \u56fa\u5b9a\u6279\u6b21\u5927\u5c0f\n\u00a0 \u00a0 @Override\n\u00a0 \u00a0\u00a0public\u00a0void\u00a0invoke(T\u00a0value, Context context)\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 bufferedValues.add(value);\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(bufferedValues.size() &gt;= batchSize) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u53ea\u80fd\u57fa\u4e8e\u8bb0\u5f55\u6570\u6512\u6279\uff0c\u65e0\u6cd5\u57fa\u4e8e\u6570\u636e\u91cf\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 flush();\n\u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 }<\/code><\/pre>\n<p><strong>\u5e26\u6765\u7684\u95ee\u9898\uff1a<\/strong><\/p>\n<ol>\n<li><strong>\u5185\u5b58\u5360\u7528\u4e0d\u53ef\u63a7\uff1a<\/strong> 100 \u6761 1KB \u7684\u65e5\u5fd7\u548c 100 \u6761 10MB \u7684\u65e5\u5fd7\u5360\u7528\u5185\u5b58\u5dee\u8ddd 100 \u500d\u3002<\/li>\n<li><strong>OOM \u98ce\u9669\u9ad8\uff1a<\/strong> \u5927\u65e5\u5fd7\u8bb0\u5f55\uff08\u5982\u5806\u6808\u8f6c\u50a8\uff09\u4f1a\u8fc5\u901f\u6491\u7206\u5185\u5b58\u3002<\/li>\n<li><strong>\u5199\u5165\u6027\u80fd\u5dee\uff1a<\/strong> \u65e0\u6cd5\u6839\u636e\u8bb0\u5f55\u5927\u5c0f\u52a8\u6001\u8c03\u6574\u6279\u6b21\uff0c\u5bfc\u81f4\u5c0f\u8bb0\u5f55\u6279\u6b21\u8fc7\u5927\u6d6a\u8d39\u7f51\u7edc\u5f00\u9500\u3002<\/li>\n<\/ol>\n<h4>\u75db\u70b9\u4e8c\uff1a\u65e0\u6cd5\u652f\u6301\u52a8\u6001\u8868\u7ed3\u6784<\/h4>\n<p><strong>\u95ee\u9898\u8868\u73b0\uff1a<\/strong><\/p>\n<pre><code>\/\/ Flink \u5b98\u65b9 Sink \u53ea\u80fd\u5199\u5165\u56fa\u5b9a\u8868\npublic\u00a0class\u00a0JdbcSink\u00a0{\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0String\u00a0sql; \u00a0\/\/ \u56fa\u5b9a\u7684 INSERT SQL\n\u00a0 \u00a0\u00a0public\u00a0JdbcSink(String\u00a0jdbcUrl,\u00a0String\u00a0sql, ...) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0this.sql\u00a0= sql; \u00a0\/\/ \u786c\u7f16\u7801\u7684\u8868\u7ed3\u6784\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u5e26\u6765\u7684\u95ee\u9898\uff1a<\/strong><\/p>\n<ol>\n<li><strong>\u591a\u5e94\u7528\u65e0\u6cd5\u9694\u79bb\uff1a<\/strong> \u6240\u6709\u5e94\u7528\u7684\u6570\u636e\u5199\u5165\u540c\u4e00\u5f20\u8868\uff0c\u901a\u8fc7\u7279\u5b9a\u5206\u8868\u7b56\u7565\u533a\u5206\u3002<\/li>\n<li><strong>\u6269\u5c55\u6027\u5dee\uff1a<\/strong> \u65b0\u589e\u5e94\u7528\u9700\u8981\u624b\u52a8\u5efa\u8868\uff0c\u65e0\u6cd5\u52a8\u6001\u8def\u7531\u3002<\/li>\n<li><strong>\u6027\u80fd\u74f6\u9888\uff1a<\/strong> \u5355\u8868\u6570\u636e\u91cf\u8fc7\u5927\uff08\u767e\u4ebf\u7ea7\uff09\uff0c\u67e5\u8be2\u548c\u5199\u5165\u6027\u80fd\u6025\u5267\u4e0b\u964d\u3002<\/li>\n<\/ol>\n<h4>\u75db\u70b9\u4e09\uff1a\u5206\u5e03\u5f0f\u8868\u5199\u5165\u6027\u80fd\u95ee\u9898<\/h4>\n<p><strong>\u95ee\u9898\u8868\u73b0\uff1a<\/strong><\/p>\n<pre><code>\/\/ \u5927\u591a\u6570\u751f\u4ea7\u5b9e\u73b0\u76f4\u63a5\u5199\u5165\u5206\u5e03\u5f0f\u8868\nINSERT INTO distributed_table_all\u00a0VALUES\u00a0(...)<\/code><\/pre>\n<p><strong>ClickHouse \u5206\u5e03\u5f0f\u8868\u7684\u5de5\u4f5c\u539f\u7406\uff1a<\/strong><\/p>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603249\" alt=\"\" title=\"\"><\/p>\n<p><strong>\u5e26\u6765\u7684\u95ee\u9898\uff1a<\/strong><\/p>\n<ol>\n<li><strong>\u7f51\u7edc\u5f00\u9500\u5927\uff1a<\/strong> \u6570\u636e\u9700\u8981\u7ecf\u8fc7\u5206\u5e03\u5f0f\u8868\u5c42\u8f6c\u53d1\uff0c\u5ef6\u8fdf\u589e\u52a0\u3002<\/li>\n<li><strong>\u5199\u5165\u6027\u80fd\u5dee\uff1a<\/strong> \u5206\u5e03\u5f0f\u8868\u589e\u52a0\u4e86\u8def\u7531\u548c\u8f6c\u53d1\u903b\u8f91\uff0c\u541e\u5410\u91cf\u964d\u4f4e\u3002<\/li>\n<li><strong>\u70ed\u70b9\u95ee\u9898\uff1a<\/strong> \u6240\u6709\u6570\u636e\u5148\u5230\u5206\u5e03\u5f0f\u8868\u8282\u70b9\uff0c\u518d\u8f6c\u53d1\uff0c\u9020\u6210\u5355\u70b9\u74f6\u9888\u3002<\/li>\n<\/ol>\n<h3>\u751f\u4ea7\u7ea7\u65b9\u6848\u7684\u6838\u5fc3\u6539\u8fdb<\/h3>\n<p>\u9488\u5bf9\u4ee5\u4e0a\u75db\u70b9\uff0c\u672c\u65b9\u6848\u63d0\u4f9b\u4e86\u4ee5\u4e0b\u6838\u5fc3\u6539\u8fdb\uff1a<\/p>\n<h4>\u6539\u8fdb\u4e00\uff1a\u57fa\u4e8e\u6570\u636e\u91cf\u7684\u6512\u6279\u673a\u5236<\/h4>\n<pre><code>public\u00a0class\u00a0ClickHouseSinkCounter\u00a0{\n\u00a0 \u00a0\u00a0private\u00a0Long\u00a0metaSize; \u00a0\/\/ \u7d2f\u8ba1\u6570\u636e\u91cf\uff08\u5b57\u8282\uff09\n\u00a0 \u00a0\u00a0public\u00a0void\u00a0add(LogModel\u00a0value) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0this.values.add(value);\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0this.metaSize\u00a0+= value.getMetaSize(); \u00a0\/\/ \u7d2f\u52a0\u6570\u636e\u91cf\n\u00a0 \u00a0 }\n}\n\/\/ \u89e6\u53d1\u6761\u4ef6\nprivate\u00a0boolean\u00a0flushCondition(String\u00a0application) {\n\u00a0 \u00a0\u00a0return\u00a0checkMetaSize(application) \u00a0\/\/ metaSize &gt;= 10000 \u5b57\u8282\n\u00a0 \u00a0 \u00a0 \u00a0 ||\u00a0checkTime(application); \u00a0 \u00a0\u00a0\/\/ \u6216\u8d85\u65f6 30 \u79d2\n}<\/code><\/pre>\n<p><strong>\u4f18\u52bf\uff1a<\/strong><\/p>\n<ul>\n<li><strong>\u5185\u5b58\u53ef\u63a7\uff1a<\/strong> \u6839\u636e\u6570\u636e\u91cf\u800c\u975e\u8bb0\u5f55\u6570\u6512\u6279\u3002<\/li>\n<li><strong>\u7cbe\u786e\u63a7\u5236\uff1a<\/strong> 1KB \u7684\u8bb0\u5f55\u6512 10000 \u6761 = 10MB\uff0c1MB \u7684\u8bb0\u5f55\u6512 10 \u6761 = 10MB\u3002<\/li>\n<li><strong>\u907f\u514dOOM\uff1a<\/strong> \u5927\u65e5\u5fd7\u8bb0\u5f55\u4e0d\u4f1a\u6491\u7206\u5185\u5b58\u3002<\/li>\n<\/ul>\n<h4>\u6539\u8fdb\u4e8c\uff1a\u52a8\u6001\u8868\u7ed3\u6784\u4e0e\u5206\u7247\u7b56\u7565<\/h4>\n<pre><code>public\u00a0abstract\u00a0class\u00a0ClickHouseShardStrategy&lt;T&gt; {\n\u00a0 \u00a0\u00a0public\u00a0abstract\u00a0String\u00a0getTableName(T data);\n}\n\/\/\u65e5\u5fd7\u4fa7\u5b9e\u73b0\u4e3a\u5e94\u7528\u7ea7\u5206\u8868\npublic\u00a0class\u00a0LogClickHouseShardStrategy\u00a0extends\u00a0ClickHouseShardStrategy&lt;String&gt; {\n\u00a0 \u00a0\u00a0@Override\n\u00a0 \u00a0\u00a0public\u00a0String\u00a0getTableName(String\u00a0application) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u52a8\u6001\u8def\u7531\uff1aorder-service \u2192 tb_logs_order_service\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0String.format(&quot;tb_logs_%s&quot;, application);\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u4f18\u52bf\uff1a<\/strong><\/p>\n<ul>\n<li><strong>\u5e94\u7528\u9694\u79bb\uff1a<\/strong> \u65e5\u5fd7\u4fa7\u5185\u7f6e\u5e94\u7528\u7ea7\u5206\u8868\uff0c\u6bcf\u4e2a\u5e94\u7528\u72ec\u7acb\u5206\u8868\u3002<\/li>\n<li><strong>\u52a8\u6001\u8def\u7531\uff1a<\/strong> \u6839\u636e application \u81ea\u52a8\u8def\u7531\u5230\u76ee\u6807\u8868\u3002<\/li>\n<li><strong>\u6269\u5c55\u6027\u5f3a\uff1a<\/strong> \u65b0\u589e\u5e94\u7528\u65e0\u9700\u624b\u52a8\u5efa\u8868\uff08\u914d\u5408 ClickHouse \u81ea\u52a8\u5efa\u8868\uff09\u3002<\/li>\n<\/ul>\n<h4>\u6539\u8fdb\u4e09\uff1a\u672c\u5730\u8868\u5199\u5165 + \u52a8\u6001\u8282\u70b9\u53d1\u73b0<\/h4>\n<pre><code>public\u00a0class\u00a0ClickHouseLocalWriter\u00a0extends\u00a0ClickHouseWriter\u00a0{\n\u00a0 \u00a0\u00a0\/\/ \u76f4\u63a5\u5199\u672c\u5730\u8868\uff0c\u907f\u514d\u5206\u5e03\u5f0f\u8868\u8f6c\u53d1\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0ConcurrentMap&lt;String,\u00a0HikariDataSource&gt; dataSourceMap;\n\u00a0 \u00a0\u00a0@Override\n\u00a0 \u00a0\u00a0public\u00a0HikariDataSource\u00a0getNextDataSource(Set&lt;String&gt; exceptionHosts) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ 1. \u52a8\u6001\u83b7\u53d6\u96c6\u7fa4\u8282\u70b9\u5217\u8868\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0List&lt;String&gt; healthyHosts =\u00a0getHealthyHosts(exceptionHosts);\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ 2. \u968f\u673a\u9009\u62e9\u5065\u5eb7\u8282\u70b9\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0dataSourceMap.get(healthyHosts.get(random.nextInt(size)));\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u4f18\u52bf\uff1a<\/strong><\/p>\n<ul>\n<li><strong>\u6027\u80fd\u63d0\u5347\uff1a<\/strong> \u76f4\u63a5\u5199\u672c\u5730\u8868\uff0c\u907f\u514d\u7f51\u7edc\u8f6c\u53d1\u3002<\/li>\n<li><strong>\u9ad8\u53ef\u7528\uff1a<\/strong> \u52a8\u6001\u8282\u70b9\u53d1\u73b0 + \u6545\u969c\u8282\u70b9\u5254\u9664\u3002<\/li>\n<li><strong>\u8d1f\u8f7d\u5747\u8861\uff1a<\/strong> \u968f\u673a\u9009\u62e9 + Shuffle \u521d\u59cb\u5316\u3002<\/li>\n<\/ul>\n<h3>\u6280\u672f\u65b9\u6848\u6982\u89c8<\/h3>\n<p>\u57fa\u4e8e\u4ee5\u4e0a\u6539\u8fdb\uff0c\u672c\u65b9\u6848\u63d0\u4f9b\u4e86\u4ee5\u4e0b\u6838\u5fc3\u80fd\u529b\uff1a<\/p>\n<ol>\n<li><strong>\u672c\u5730\u8868\/\u5206\u5e03\u5f0f\u8868\u5199\u5165\uff1a<\/strong> \u6027\u80fd\u4f18\u5316\u4e0e\u9ad8\u53ef\u7528\u5e73\u8861\u3002<\/li>\n<li><strong>\u5206\u7247\u7b56\u7565\uff1a<\/strong> \u6309\u5e94\u7528\u7ef4\u5ea6\u8def\u7531\u4e0e\u9694\u79bb\u3002<\/li>\n<li><strong>\u6512\u6279\u4e0e\u5185\u5b58\u63a7\u5236\uff1a<\/strong> \u53cc\u89e6\u53d1\u673a\u5236\uff08\u6570\u636e\u91cf + \u8d85\u65f6\uff09\u3002<\/li>\n<li><strong>\u6d41\u91cf\u63a7\u5236\u4e0e\u9650\u6d41\uff1a<\/strong> \u6709\u754c\u961f\u5217 + \u8fde\u63a5\u6c60\u3002<\/li>\n<li><strong>\u5065\u58ee\u7684\u91cd\u8bd5\u673a\u5236\uff1a<\/strong> \u9012\u5f52\u91cd\u8bd5 + \u6545\u969c\u8282\u70b9\u5254\u9664\u3002<\/li>\n<li><strong>Checkpoint \u8bed\u4e49\u4fdd\u8bc1\uff1a<\/strong> At-Least-Once \u6570\u636e\u4e00\u81f4\u6027\u3002<\/li>\n<\/ol>\n<h2>\u4e8c\u3001\u6838\u5fc3\u67b6\u6784\u8bbe\u8ba1<\/h2>\n<h3>\u67b6\u6784\u56fe<\/h3>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603250\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<h3>\u6838\u5fc3\u7ec4\u4ef6<\/h3>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603251\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<h3>\u6838\u5fc3\u6d41\u7a0b<\/h3>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603252\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<h2>\u4e09\u3001\u672c\u5730\u8868 vs \u5206\u5e03\u5f0f\u8868\u5199\u5165<\/h2>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603253\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<h3>ClickHouse \u8868\u7ed3\u6784\u8bf4\u660e<\/h3>\n<p>ClickHouse <strong>\u63a8\u8350\u76f4\u63a5\u5199\u672c\u5730\u8868<\/strong>\uff0c\u539f\u56e0\uff1a<\/p>\n<ol>\n<li><strong>\u5199\u5165\u6027\u80fd\uff1a<\/strong> \u907f\u514d\u5206\u5e03\u5f0f\u8868\u7684\u7f51\u7edc\u5206\u53d1\u3002<\/li>\n<li><strong>\u6570\u636e\u4e00\u81f4\u6027\uff1a<\/strong> \u76f4\u63a5\u5199\u5165\u76ee\u6807\u8282\u70b9\uff0c\u51cf\u5c11\u4e2d\u95f4\u73af\u8282\u6545\u969c\u70b9\uff0c\u6bd4\u5206\u5e03\u5f0f\u8868\u5199\u5165\u66f4\u5b89\u5168\uff0c\u5229\u4e8e\u5de5\u7a0b\u5316\u3002<\/li>\n<li><strong>\u8d1f\u8f7d\u5747\u8861\uff1a<\/strong> \u5ba2\u6237\u7aef\u8def\u7531\u5b9e\u73b0\u8d1f\u8f7d\u5206\u6563\u3002<\/li>\n<\/ol>\n<pre><code>-- \u672c\u5730\u8868\uff08\u5b9e\u9645\u5b58\u50a8\u6570\u636e\uff09\nCREATE TABLE\u00a0tb_logs_local\u00a0ON\u00a0CLUSTER\u00a0'default'\u00a0(\n\u00a0 \u00a0 application String,\n\u00a0 \u00a0 environment String,\n\u00a0 \u00a0 message String,\n\u00a0 \u00a0 log_time DateTime\n) ENGINE\u00a0=\u00a0MergeTree()\nPARTITION\u00a0BY\u00a0toYYYYMM(log_time)\nORDER\u00a0BY\u00a0(application, log_time);\n-- \u5206\u5e03\u5f0f\u8868\uff08\u903b\u8f91\u89c6\u56fe\uff0c\u4e0d\u5b58\u50a8\u6570\u636e\uff09\nCREATE TABLE\u00a0tb_logs_all\u00a0ON\u00a0CLUSTER\u00a0'default'\u00a0AS\u00a0tb_logs_local\nENGINE\u00a0=\u00a0Distributed('default', dw_log, tb_logs_local, cityHash64(application));<\/code><\/pre>\n<h3>HikariCP \u8fde\u63a5\u6c60\u914d\u7f6e<\/h3>\n<pre><code>\/\/ HikariCP \u8fde\u63a5\u6c60\u914d\u7f6e\npublic\u00a0class\u00a0ClickHouseDataSourceUtils\u00a0{\n\u00a0 \u00a0\u00a0private\u00a0static\u00a0HikariConfig\u00a0getHikariConfig(DataSourceImpl dataSource)\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 HikariConfig config =\u00a0new\u00a0HikariConfig();\n\u00a0 \u00a0 \u00a0 \u00a0 config.setConnectionTimeout(30000L); \u00a0 \u00a0\/\/ \u8fde\u63a5\u8d85\u65f6 30s\n\u00a0 \u00a0 \u00a0 \u00a0 config.setMaximumPoolSize(20); \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\/\/ \u6700\u5927\u8fde\u63a5\u6570 20\n\u00a0 \u00a0 \u00a0 \u00a0 config.setMinimumIdle(2); \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u6700\u5c0f\u7a7a\u95f2 2\n\u00a0 \u00a0 \u00a0 \u00a0 config.setDataSource(dataSource);\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0config;\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0private\u00a0static\u00a0Properties\u00a0getClickHouseProperties(ClickHouseSinkCommonParams\u00a0params)\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 Properties props =\u00a0new\u00a0Properties();\n\u00a0 \u00a0 \u00a0 \u00a0 props.setProperty(&quot;user&quot;,\u00a0params.getUser());\n\u00a0 \u00a0 \u00a0 \u00a0 props.setProperty(&quot;password&quot;,\u00a0params.getPassword());\n\u00a0 \u00a0 \u00a0 \u00a0 props.setProperty(&quot;database&quot;,\u00a0params.getDatabase());\n\u00a0 \u00a0 \u00a0 \u00a0 props.setProperty(&quot;socket_timeout&quot;,\u00a0&quot;180000&quot;); \u00a0 \u00a0 \u00a0\/\/ Socket \u8d85\u65f6 3 \u5206\u949f\n\u00a0 \u00a0 \u00a0 \u00a0 props.setProperty(&quot;socket_keepalive&quot;,\u00a0&quot;true&quot;); \u00a0 \u00a0 \u00a0\/\/ \u4fdd\u6301\u8fde\u63a5\n\u00a0 \u00a0 \u00a0 \u00a0 props.setProperty(&quot;http_connection_provider&quot;,\u00a0&quot;APACHE_HTTP_CLIENT&quot;);\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0props;\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u914d\u7f6e\u8bf4\u660e\uff1a<\/strong><\/p>\n<ul>\n<li>maxPoolSize=20\uff1a\u6bcf\u4e2a ClickHouse \u8282\u70b9\u6700\u591a 20 \u4e2a\u8fde\u63a5\u3002<\/li>\n<li>minIdle=2\uff1a\u4fdd\u6301 2 \u4e2a\u7a7a\u95f2\u8fde\u63a5\uff0c\u907f\u514d\u9891\u7e41\u521b\u5efa\u3002<\/li>\n<li>socket_timeout=180s\uff1aSocket \u8d85\u65f6 3 \u5206\u949f\uff0c\u9632\u6b62\u957f\u65f6\u95f4\u67e5\u8be2\u963b\u585e\u3002<\/li>\n<\/ul>\n<h3>ClickHouseLocalWriter\uff1a\u52a8\u6001\u8282\u70b9\u53d1\u73b0<\/h3>\n<pre><code>public\u00a0class\u00a0ClickHouseLocalWriter\u00a0extends\u00a0ClickHouseWriter\u00a0{\n\u00a0 \u00a0\u00a0\/\/ \u672c\u5730\u8282\u70b9\u7f13\u5b58\uff0c\u6309 IP \u7ef4\u62a4\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0ConcurrentMap&lt;String,\u00a0HikariDataSource&gt; dataSourceMap;\n\u00a0 \u00a0\u00a0\/\/ \u52a8\u6001\u83b7\u53d6\u96c6\u7fa4\u672c\u5730\u8868\u8282\u70b9\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0ClusterIpsUtils\u00a0clusterIpsUtils;\n\u00a0 \u00a0\u00a0\/\/ IP \u53d8\u66f4\u6807\u5fd7\uff08CAS \u9501\uff0c\u907f\u514d\u5e76\u53d1\u66f4\u65b0\uff09\n\u00a0 \u00a0\u00a0private\u00a0static\u00a0final\u00a0AtomicBoolean\u00a0IP_CHANGING\u00a0=\u00a0new\u00a0AtomicBoolean(false);\n\u00a0 \u00a0\u00a0@Override\n\u00a0 \u00a0\u00a0public\u00a0HikariDataSource\u00a0getNextDataSource(Set&lt;String&gt; exceptionHosts) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ 1\ufe0f\u20e3 \u68c0\u6d4b\u96c6\u7fa4\u8282\u70b9\u53d8\u5316\uff08\u901a\u8fc7 CAS \u907f\u514d\u5e76\u53d1\u66f4\u65b0\uff09\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(clusterIpsChanged() &amp;&amp;\u00a0IP_CHANGING.compareAndSet(false,\u00a0true)) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0try\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0ipChanged();\u00a0\/\/ \u52a8\u6001\u66f4\u65b0 dataSourceMap\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\u00a0finally\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0IP_CHANGING.set(false);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ 2\ufe0f\u20e3 \u83b7\u53d6\u5f02\u5e38\u8282\u70b9\u5217\u8868\uff08\u4ece Redis + APM \u5b9e\u65f6\u67e5\u8be2\uff09\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0Set&lt;String&gt; exceptIps = clusterIpsUtils.getExceptIps();\n\u00a0 \u00a0 \u00a0 \u00a0 exceptIps.addAll(exceptionHosts);\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ 3\ufe0f\u20e3 \u8fc7\u6ee4\u5065\u5eb7\u8282\u70b9\uff0c\u968f\u673a\u9009\u62e9\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0List&lt;String&gt; healthyHosts = dataSourceMap.keySet().stream()\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 .filter(host -&gt; !exceptIps.contains(host))\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 .collect(Collectors.toList());\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(CollectionUtils.isEmpty(healthyHosts)) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0throw\u00a0new\u00a0RuntimeException(&quot;Can't get datasource from local cache&quot;);\n\u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0dataSourceMap.get(healthyHosts.get(random.nextInt(healthyHosts.size())));\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0private\u00a0void\u00a0ipChanged() {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0List&lt;String&gt; clusterIps = clusterIpsUtils.getClusterIps();\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u65b0\u589e\u8282\u70b9\uff1a\u81ea\u52a8\u521b\u5efa\u8fde\u63a5\u6c60\n\u00a0 \u00a0 \u00a0 \u00a0 clusterIps.forEach(ip -&gt;\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 dataSourceMap.computeIfAbsent(ip, v -&gt;\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0createHikariDataSource(ip, port)\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 )\n\u00a0 \u00a0 \u00a0 \u00a0 );\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u79fb\u9664\u4e0b\u7ebf\u8282\u70b9\uff1a\u5173\u95ed\u8fde\u63a5\u6c60\n\u00a0 \u00a0 \u00a0 \u00a0 dataSourceMap.forEach((ip, ds) -&gt; {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(!clusterIps.contains(ip)) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 dataSourceMap.remove(ip);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 ds.close();\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 });\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u6838\u5fc3\u903b\u8f91\uff1a<\/strong><\/p>\n<ol>\n<li><strong>\u52a8\u6001\u8282\u70b9\u53d1\u73b0\uff1a<\/strong> \u4ece system.clusters \u67e5\u8be2\u6240\u6709\u8282\u70b9\u3002<\/li>\n<li><strong>\u81ea\u52a8\u6269\u7f29\u5bb9\uff1a<\/strong> \u8282\u70b9\u4e0a\u7ebf\u81ea\u52a8\u52a0\u5165\uff0c\u4e0b\u7ebf\u81ea\u52a8\u5254\u9664\u3002<\/li>\n<li><strong>\u6545\u969c\u8282\u70b9\u5254\u9664\uff1a<\/strong> \u901a\u8fc7 APM \u76d1\u63a7\uff0c\u81ea\u52a8\u5254\u9664\u5f02\u5e38\u8282\u70b9\u3002<\/li>\n<li><strong>\u8d1f\u8f7d\u5747\u8861\uff1a<\/strong> \u968f\u673a\u9009\u62e9\u5065\u5eb7\u8282\u70b9\uff0c\u907f\u514d\u70ed\u70b9\u3002<\/li>\n<\/ol>\n<h3>\u96c6\u7fa4\u8282\u70b9\u52a8\u6001\u53d1\u73b0\uff08ClusterIpsUtils\uff09<\/h3>\n<pre><code>public\u00a0class\u00a0ClusterIpsUtils\u00a0{\n\u00a0 \u00a0\u00a0\/\/ \u4ece system.clusters \u67e5\u8be2\u6240\u6709\u8282\u70b9\n\u00a0 \u00a0\u00a0private\u00a0static\u00a0final\u00a0String\u00a0QUERY_CLUSTER_IPS\u00a0=\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0&quot;select host_address from system.clusters where cluster = 'default'&quot;;\n\u00a0 \u00a0\u00a0\/\/ LoadingCache\uff1a\u5b9a\u65f6\u5237\u65b0\u8282\u70b9\u5217\u8868\uff081 \u5c0f\u65f6\uff09\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0LoadingCache&lt;String,\u00a0List&lt;String&gt;&gt; clusterIpsCache =\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0CacheBuilder.newBuilder()\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 .expireAfterAccess(10,\u00a0TimeUnit.HOURS)\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 .refreshAfterWrite(1,\u00a0TimeUnit.HOURS)\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 .build(CacheLoader.asyncReloading(new\u00a0CacheLoader&lt;&gt;() {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0@Override\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0public\u00a0List&lt;String&gt;\u00a0load(String\u00a0dbName) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0queryClusterIps(); \u00a0\/\/ \u5b9a\u65f6\u5237\u65b0\u8282\u70b9\u5217\u8868\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }));\n\u00a0 \u00a0\u00a0\/\/ \u5f02\u5e38\u8282\u70b9\u7f13\u5b58\uff081 \u5206\u949f\u5237\u65b0\uff09\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0LoadingCache&lt;String,\u00a0FlinkExceptIpModel&gt; exceptIpsCache =\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0CacheBuilder.newBuilder()\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 .refreshAfterWrite(1,\u00a0TimeUnit.MINUTES)\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 .build(CacheLoader.asyncReloading(new\u00a0CacheLoader&lt;&gt;() {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0@Override\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0public\u00a0FlinkExceptIpModel\u00a0load(String\u00a0dbName) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0queryExceptIp(); \u00a0\/\/ \u4ece Redis + APM \u67e5\u8be2\u5f02\u5e38\u8282\u70b9\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }));\n}<\/code><\/pre>\n<p><strong>\u5f02\u5e38\u8282\u70b9\u76d1\u63a7\u7b56\u7565\uff1a<\/strong><\/p>\n<ul>\n<li><strong>\u78c1\u76d8\u4f7f\u7528\u7387 &gt;= 90%\uff1a<\/strong> \u4ece APM \u67e5\u8be2 Prometheus \u6307\u6807\uff0c\u81ea\u52a8\u52a0\u5165\u9ed1\u540d\u5355\u3002<\/li>\n<li><strong>HTTP \u8fde\u63a5\u6570 &gt;= 50\uff1a<\/strong> \u8fde\u63a5\u6570\u8fc7\u591a\u8bf4\u660e\u8282\u70b9\u538b\u529b\u5927\uff0c\u81ea\u52a8\u52a0\u5165\u9ed1\u540d\u5355\u3002<\/li>\n<li><strong>\u4eba\u5de5\u914d\u7f6e\uff1a<\/strong> \u901a\u8fc7 Redis \u914d\u7f6e\u624b\u52a8\u5254\u9664\u8282\u70b9<\/li>\n<\/ul>\n<p><strong>\u6570\u636e\u6765\u6e90\uff1a<\/strong><\/p>\n<ol>\n<li><strong>ClickHouse system.clusters \u8868\uff1a<\/strong> \u83b7\u53d6\u6240\u6709\u96c6\u7fa4\u8282\u70b9\u3002<\/li>\n<li><strong>APM Prometheus \u63a5\u53e3\uff1a<\/strong> \u76d1\u63a7\u8282\u70b9\u5065\u5eb7\u72b6\u6001\u3002<\/li>\n<li><strong>Redis \u7f13\u5b58\uff1a<\/strong> \u4eba\u5de5\u914d\u7f6e\u7684\u5f02\u5e38\u8282\u70b9\u3002<\/li>\n<\/ol>\n<h3>\u8d1f\u8f7d\u5747\u8861\u4f18\u5316<\/h3>\n<pre><code>public\u00a0class\u00a0ClickHouseWriter\u00a0{\n\u00a0 \u00a0\u00a0public\u00a0&lt;T&gt; ClickHouseWriter(...) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ Shuffle\uff1a\u968f\u673a\u6253\u4e71\u6570\u636e\u6e90\u987a\u5e8f\n\u00a0 \u00a0 \u00a0 \u00a0 Collections.shuffle(clickHouseDataSources);\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0this.clickHouseDataSources = clickHouseDataSources;\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0public\u00a0HikariDataSource getNextDataSource(Set&lt;String&gt; exceptionHosts) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u8f6e\u8be2 + \u968f\u673a\u9009\u62e9\uff08\u5df2 shuffle\uff0c\u907f\u514d\u70ed\u70b9\uff09\n\u00a0 \u00a0 \u00a0 \u00a0 int current =\u00a0this.currentRandom.getAndIncrement();\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(current &gt;= clickHouseDataSources.size()) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0this.currentRandom.set(0);\n\u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0clickHouseDataSources.get(currentRandom.get());\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u4f18\u52bf\uff1a<\/strong><\/p>\n<ul>\n<li>\u521d\u59cb\u5316\u65f6 shuffle\uff0c\u907f\u514d\u6240\u6709 writer \u540c\u65f6\u4ece\u7b2c\u4e00\u4e2a\u8282\u70b9\u5f00\u59cb\u3002<\/li>\n<li>\u8f6e\u8be2 + \u968f\u673a\u9009\u62e9\uff0c\u8d1f\u8f7d\u5206\u6563\u66f4\u5747\u5300\u3002<\/li>\n<li>\u6545\u969c\u8282\u70b9\u81ea\u52a8\u5254\u9664\u3002<\/li>\n<\/ul>\n<h2>\u56db\u3001\u652f\u6301\u5206\u8868\u7b56\u7565<\/h2>\n<h3>\u5206\u7247\u7b56\u7565\u62bd\u8c61<\/h3>\n<pre><code>public\u00a0abstract\u00a0class\u00a0ClickHouseShardStrategy&lt;T&gt; {\n\u00a0 \u00a0\u00a0private\u00a0String tableName; \u00a0 \u00a0 \u00a0\/\/ \u8868\u540d\u6a21\u677f\uff0c\u5982 &quot;tb_log_%s&quot;\n\u00a0 \u00a0\u00a0private\u00a0Integer tableCount; \u00a0 \u00a0\/\/ \u5206\u8868\u6570\u91cf\n\u00a0 \u00a0\u00a0\/\/ \u6839\u636e\u6570\u636e\u51b3\u5b9a\u76ee\u6807\u8868\u540d\n\u00a0 \u00a0\u00a0public\u00a0abstract\u00a0String\u00a0getTableName(T data);\n}<\/code><\/pre>\n<h3>\u65e5\u5fd7\u5206\u7247\u5b9e\u73b0<\/h3>\n<pre><code>public\u00a0class\u00a0LogClickHouseShardStrategy\u00a0extends\u00a0ClickHouseShardStrategy&lt;String&gt; {\n\u00a0 \u00a0\u00a0@Override\n\u00a0 \u00a0\u00a0public\u00a0String\u00a0getTableName(String\u00a0application) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u8868\u540d\u683c\u5f0f\uff1atb_log_{application}\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u4f8b\u5982\uff1aapplication = &quot;order-service&quot; -&gt; table = &quot;tb_log_order_service&quot;\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0String.format(\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0this.getTableName(),\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 application.replace(&quot;-&quot;,\u00a0&quot;_&quot;).toLowerCase()\n\u00a0 \u00a0 \u00a0 \u00a0 );\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<h3>\u6309\u8868\uff08\u5e94\u7528\uff09\u7ef4\u5ea6\u7684\u7f13\u51b2\u533a<\/h3>\n<p>\u65e5\u5fd7\u4fa7\u7ef4\u5ea6\u964d\u7ea7\u4e3a\u5e94\u7528\u540d\u79f0\u7ef4\u5ea6\u7f13\u51b2\u533a\uff0c\u5b9e\u5219\u56e0\u4e3a\u6309\u7167\u5e94\u7528\u5206\u8868\uff0c<\/p>\n<p>\u4e1a\u52a1\u65b9\u53ef\u4f7f\u7528\u81ea\u8eab\u5206\u8868\u7b56\u7565\u643a\u5e26\u8868\u540d\u5143\u6570\u636e\uff0c\u8fdb\u884c\u8868\u7ef4\u5ea6\u7f13\u51b2\u3002<\/p>\n<pre><code>public\u00a0class\u00a0ClickHouseShardSinkBuffer\u00a0{\n\u00a0 \u00a0\u00a0\/\/ \u6309 application \u5206\u7ec4\u7684\u7f13\u51b2\u533a\uff08ConcurrentHashMap \u4fdd\u8bc1\u5e76\u53d1\u5b89\u5168\uff09\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0ConcurrentHashMap&lt;String,\u00a0ClickHouseSinkCounter&gt; localValues;\n\u00a0 \u00a0\u00a0public\u00a0void\u00a0put(LogModel\u00a0value) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0String\u00a0application = value.getApplication();\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ 1\ufe0f\u20e3 \u68c0\u67e5\u662f\u5426\u9700\u8981 flush\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(flushCondition(application)) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0addToQueue(application);\u00a0\/\/ \u89e6\u53d1\u5199\u5165\n\u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ 2\ufe0f\u20e3 \u6dfb\u52a0\u5230\u7f13\u51b2\u533a\uff08\u7ebf\u7a0b\u5b89\u5168\u7684 compute \u64cd\u4f5c\uff09\n\u00a0 \u00a0 \u00a0 \u00a0 localValues.compute(application, (k, v) -&gt; {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(v ==\u00a0null) v =\u00a0new\u00a0ClickHouseSinkCounter();\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 v.add(value);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0v;\n\u00a0 \u00a0 \u00a0 \u00a0 });\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0private\u00a0void\u00a0addToQueue(String\u00a0application) {\n\u00a0 \u00a0 \u00a0 \u00a0 localValues.computeIfPresent(application, (k, v) -&gt; {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u6df1\u62f7\u8d1d\u5e76\u6e05\u7a7a\uff08\u907f\u514d\u5e76\u53d1\u4fee\u6539\u5f02\u5e38\uff09\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0List&lt;LogModel&gt; deepCopy = v.copyValuesAndClear();\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u6784\u9020\u8bf7\u6c42 Blank\uff1aapplication + targetTable + values\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0String\u00a0targetTable = shardStrategy.getTableName(application);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0ClickHouseRequestBlank\u00a0blank =\u00a0new\u00a0ClickHouseRequestBlank(deepCopy, application, targetTable);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u653e\u5165\u961f\u5217\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 writer.put(blank);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0v;\n\u00a0 \u00a0 \u00a0 \u00a0 });\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u6838\u5fc3\u8bbe\u8ba1\uff1a<\/strong><\/p>\n<ul>\n<li><strong>\u5e94\u7528\u9694\u79bb\uff1a<\/strong> \u6bcf\u4e2a\u8868\uff08\u5e94\u7528\uff09\u72ec\u7acb\u7684 buffer\uff0c\u4e92\u4e0d\u5f71\u54cd\u3002<\/li>\n<li><strong>\u7ebf\u7a0b\u5b89\u5168\uff1a<\/strong> \u4f7f\u7528 ConcurrentHashMap.compute()\u4fdd\u8bc1\u5e76\u53d1\u5b89\u5168\u3002<\/li>\n<li><strong>\u6df1\u62f7\u8d1d\uff1a<\/strong> List.copyOf() \u521b\u5efa\u4e0d\u53ef\u53d8\u526f\u672c\uff0c\u907f\u514d\u5e76\u53d1\u4fee\u6539\u3002<\/li>\n<li><strong>\u6279\u91cf\u6e05\u7a7a\uff1a<\/strong> \u4e00\u6b21\u6027\u53d6\u51fa\u6240\u6709\u6570\u636e\uff0c\u6e05\u7a7a\u8ba1\u6570\u5668\u3002<\/li>\n<\/ul>\n<h2>\u4e94\u3001\u6512\u6279\u4e0e\u5185\u5b58\u63a7\u5236<\/h2>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603254\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<h3>\u53cc\u89e6\u53d1\u673a\u5236<\/h3>\n<pre><code>public\u00a0class\u00a0ClickHouseShardSinkBuffer\u00a0{\n\u00a0 \u00a0\u00a0private\u00a0final int maxFlushBufferSize; \u00a0\/\/ \u6700\u5927\u6279\u6b21\u5927\u5c0f\uff08\u5982 10000\uff09\n\u00a0 \u00a0\u00a0private\u00a0final long timeoutMillis; \u00a0 \u00a0 \u00a0\/\/ \u8d85\u65f6\u65f6\u95f4\uff08\u5982 30s\uff09\n\u00a0 \u00a0\u00a0\/\/ \u89e6\u53d1\u6761\u4ef6\u68c0\u67e5\uff08\u6ee1\u8db3\u4efb\u4e00\u5373\u89e6\u53d1\uff09\n\u00a0 \u00a0\u00a0private\u00a0boolean\u00a0flushCondition(String\u00a0application) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0localValues.get(application) !=\u00a0null\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 &amp;&amp; (checkMetaSize(application) ||\u00a0checkTime(application));\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0\/\/ \u6761\u4ef61\uff1a\u8fbe\u5230\u6279\u6b21\u5927\u5c0f\n\u00a0 \u00a0\u00a0private\u00a0boolean\u00a0checkMetaSize(String\u00a0application) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0localValues.get(application).getMetaSize() &gt;= maxFlushBufferSize;\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0\/\/ \u6761\u4ef62\uff1a\u8d85\u65f6\n\u00a0 \u00a0\u00a0private\u00a0boolean\u00a0checkTime(String\u00a0application) {\n\u00a0 \u00a0 \u00a0 \u00a0 long current =\u00a0System.currentTimeMillis();\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0current - localValues.get(application).getInsertTime() &gt; timeoutMillis;\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<h3>\u6279\u6b21\u5927\u5c0f\u8ba1\u7b97<\/h3>\n<pre><code>public\u00a0class\u00a0ClickHouseSinkCounter\u00a0{\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0List&lt;LogModel&gt; values;\n\u00a0 \u00a0\u00a0private\u00a0Long\u00a0metaSize;\u00a0\/\/ \u7d2f\u8ba1\u7684 metaSize\uff08\u5b57\u8282\uff09\n\u00a0 \u00a0\u00a0public\u00a0void add(LogModel value) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0this.values.add(value);\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0this.metaSize += value.getMetaSize();\u00a0\/\/ \u7d2f\u52a0 metaSize\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0public\u00a0List&lt;LogModel&gt; copyValuesAndClear() {\n\u00a0 \u00a0 \u00a0 \u00a0 List&lt;LogModel&gt; logModels = List.copyOf(this.values);\u00a0\/\/ \u6df1\u62f7\u8d1d\uff08\u4e0d\u53ef\u53d8\uff09\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0this.values.clear();\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0this.metaSize =\u00a00L;\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0this.insertTime = System.currentTimeMillis();\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0logModels;\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u5173\u952e\u70b9\uff1a<\/strong><\/p>\n<ul>\n<li>\u4f7f\u7528 metaSize\uff08\u5b57\u8282\u6570\uff09\u800c\u975e\u8bb0\u5f55\u6570\u63a7\u5236\u6279\u6b21\uff0c\u5185\u5b58\u63a7\u5236\u66f4\u7cbe\u786e\u3002<\/li>\n<li>List.copyOf() \u521b\u5efa\u4e0d\u53ef\u53d8\u526f\u672c\uff0c\u907f\u514d\u5e76\u53d1\u4fee\u6539\u3002<\/li>\n<li>\u6e05\u7a7a\u540e\u91cd\u7f6e insertTime\uff0c\u4fdd\u8bc1\u8d85\u65f6\u89e6\u53d1\u51c6\u786e\u6027\u3002<\/li>\n<\/ul>\n<h3>\u5e26\u968f\u673a\u6296\u52a8\u7684\u8d85\u65f6<\/h3>\n<pre><code>private\u00a0final\u00a0long\u00a0timeoutMillis;\npublic\u00a0ClickHouseShardSinkBuffer(...,\u00a0int\u00a0timeoutSec, ...)\u00a0{\n\u00a0 \u00a0\u00a0\/\/ \u57fa\u7840\u8d85\u65f6 + 10% \u968f\u673a\u6296\u52a8\uff08\u907f\u514d\u60ca\u7fa4\u6548\u5e94\uff09\n\u00a0 \u00a0\u00a0this.timeoutMillis = TimeUnit.SECONDS.toMillis(timeoutSec)\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 +\u00a0new\u00a0SecureRandom().nextInt((int) (timeoutSec *\u00a00.1\u00a0*\u00a01000));\n}<\/code><\/pre>\n<p><strong>\u76ee\u7684\uff1a<\/strong> \u907f\u514d\u591a\u4e2aTM \u540c\u65f6\u89e6\u53d1 flush\uff0c\u9020\u6210\u5199\u5165\u6d41\u91cf\u5cf0\u503c\u3002<\/p>\n<h3>\u914d\u7f6e\u793a\u4f8b<\/h3>\n<pre><code>ClickHouseShardSinkBuffer.Builder\n\u00a0 \u00a0 .aClickHouseSinkBuffer()\n\u00a0 \u00a0 .withTargetTable(&quot;single_table&quot;) \u00a0\/\/\u5355\u8868\u65f6\uff0c\u53ef\u76f4\u63a5\u4f7f\u7528\u6307\u5b9a\u8868\u540d\n\u00a0 \u00a0 .withMaxFlushBufferSize(10000) \u00a0\/\/ \u5bf9\u5e94\u5b57\u8282\u6570\n\u00a0 \u00a0 .withTimeoutSec(30) \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\/\/ 30 \u79d2\u8d85\u65f6\n\u00a0 \u00a0 .withClickHouseShardStrategy(new\u00a0LogClickHouseShardStrategy(&quot;table_prefix_%s&quot;,\u00a08)) \u00a0\/\/\u5206\u8868\u7b56\u7565\u65f6\uff0c\u4f7f\u7528\n\u00a0 \u00a0\u00a0\/\/ \u5206\u8868\u7b56\u7565\u53ef\u6839\u636e\u4e1a\u52a1\u5b9e\u9645\u60c5\u51b5\u8fdb\u884c\u6269\u5c55\n\u00a0 \u00a0 .build(clickHouseWriter);<\/code><\/pre>\n<h2>\u516d\u3001\u5199\u5165\u9650\u6d41\u4e0e\u6d41\u91cf\u63a7\u5236<\/h2>\n<h3>\u6709\u754c\u961f\u5217\u8bbe\u8ba1<\/h3>\n<pre><code>public\u00a0class\u00a0ClickHouseWriter\u00a0{\n\u00a0 \u00a0\u00a0\/\/ \u6709\u754c\u963b\u585e\u961f\u5217\n\u00a0 \u00a0\u00a0private\u00a0final BlockingQueue&lt;ClickHouseRequestBlank&gt; commonQueue;\n\u00a0 \u00a0\u00a0public\u00a0ClickHouseWriter(ClickHouseSinkCommonParams sinkParams, ...)\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u961f\u5217\u6700\u5927\u5bb9\u91cf\u914d\u7f6e\uff08\u9ed8\u8ba4 10\uff09\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0this.commonQueue =\u00a0new\u00a0LinkedBlockingQueue&lt;&gt;(sinkParams.getQueueMaxCapacity());\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0public\u00a0void\u00a0put(ClickHouseRequestBlank\u00a0params)\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 unProcessedCounter.incrementAndGet();\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ put() \u65b9\u6cd5\u5728\u961f\u5217\u6ee1\u65f6\u4f1a\u963b\u585e\uff0c\u5b9e\u73b0\u80cc\u538b\n\u00a0 \u00a0 \u00a0 \u00a0 commonQueue.put(params);\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p>\u80cc\u538b\u4f20\u5bfc\uff1a<\/p>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603255\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<h3>\u7ebf\u7a0b\u6c60\u5e76\u53d1\u63a7\u5236<\/h3>\n<pre><code>public\u00a0class\u00a0ClickHouseWriter\u00a0{\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0int\u00a0numWriters;\u00a0\/\/ \u5199\u5165\u7ebf\u7a0b\u6570\n\u00a0 \u00a0\u00a0private\u00a0ExecutorService service;\n\u00a0 \u00a0\u00a0private\u00a0void\u00a0buildComponents()\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0ThreadFactory\u00a0threadFactory\u00a0=\u00a0ThreadUtil.threadFactory(&quot;clickhouse-writer&quot;);\n\u00a0 \u00a0 \u00a0 \u00a0 service = Executors.newFixedThreadPool(numWriters, threadFactory);\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u521b\u5efa\u591a\u4e2a WriterTask \u5e76\u63d0\u4ea4\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0for\u00a0(int\u00a0i\u00a0=\u00a00; i &lt; numWriters; i++) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0WriterTask\u00a0task\u00a0=\u00a0new\u00a0WriterTask(i, commonQueue, sinkParams, futures, unProcessedCounter);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 service.submit(task);\n\u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<h3>WriterTask \u6d88\u8d39\u903b\u8f91<\/h3>\n<pre><code>class\u00a0WriterTask\u00a0implements Runnable {\n\u00a0 \u00a0 @Override\n\u00a0 \u00a0\u00a0public\u00a0void\u00a0run()\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 isWorking =\u00a0true;\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0while\u00a0(isWorking || !queue.isEmpty()) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ poll() \u8d85\u65f6\u8fd4\u56de\uff08100ms\uff09\uff0c\u907f\u514d\u65e0\u9650\u7b49\u5f85\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 ClickHouseRequestBlank blank = queue.poll(100, TimeUnit.MILLISECONDS);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(blank != null) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u521b\u5efa Future \u5e76\u8bbe\u7f6e\u8d85\u65f6\uff083 \u5206\u949f\uff09\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 CompletableFuture&lt;Boolean&gt; future =\u00a0new\u00a0CompletableFuture&lt;&gt;();\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 future.orTimeout(3, TimeUnit.MINUTES);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 futures.add(future);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0try\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0send(blank, future,\u00a0new\u00a0HashSet&lt;&gt;());\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 } finally {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ final \u8fdb\u884c\u672a\u77e5\u5f02\u5e38\u515c\u5e95\uff0c\u9632\u6b62\u4e3a\u6355\u83b7\u5f02\u5e38\u9020\u6210future\u72b6\u6001\u4e0d\u5b8c\u6210\uff0c\u6c38\u4e45\u963b\u585e\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(!future.isDone()) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 future.completeExceptionally(new\u00a0RuntimeException(&quot;Unknown exception&quot;));\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 queueCounter.decrementAndGet();\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<h3>\u914d\u7f6e\u53c2\u6570<\/h3>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603256\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<h2>\u4e03\u3001\u91cd\u8bd5\u673a\u5236\u4e0e\u8d85\u65f6\u63a7\u5236<\/h2>\n<h3>Future \u8d85\u65f6\u63a7\u5236<\/h3>\n<pre><code>public\u00a0class\u00a0ClickHouseWriter\u00a0{\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0int\u00a0numWriters;\u00a0\/\/ \u5199\u5165\u7ebf\u7a0b\u6570\n\u00a0 \u00a0\u00a0private\u00a0ExecutorService service;\n\u00a0 \u00a0\u00a0private\u00a0void\u00a0buildComponents()\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0ThreadFactory\u00a0threadFactory\u00a0=\u00a0ThreadUtil.threadFactory(&quot;clickhouse-writer&quot;);\n\u00a0 \u00a0 \u00a0 \u00a0 service = Executors.newFixedThreadPool(numWriters, threadFactory);\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u521b\u5efa\u591a\u4e2a WriterTask \u5e76\u63d0\u4ea4\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0for\u00a0(int\u00a0i\u00a0=\u00a00; i &lt; numWriters; i++) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0WriterTask\u00a0task\u00a0=\u00a0new\u00a0WriterTask(i, commonQueue, sinkParams, futures, unProcessedCounter);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 service.submit(task);\n\u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u8d85\u65f6\u7b56\u7565\uff1a<\/strong><\/p>\n<ul>\n<li><strong>Future \u8d85\u65f6\uff1a<\/strong> 3 \u5206\u949f\uff08orTimeout\uff09\u3002<\/li>\n<li><strong>Socket \u8d85\u65f6\uff1a<\/strong> 3 \u5206\u949f\uff08socket_timeout=180000\uff09\u3002<\/li>\n<li><strong>\u8fde\u63a5\u8d85\u65f6\uff1a<\/strong> 30 \u79d2\uff08connectionTimeout=30000\uff09\u3002<\/li>\n<\/ul>\n<h3>\u91cd\u8bd5\u903b\u8f91<\/h3>\n<pre><code>class\u00a0WriterTask\u00a0implements Runnable {\n\u00a0 \u00a0 @Override\n\u00a0 \u00a0\u00a0public\u00a0void\u00a0run()\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 isWorking =\u00a0true;\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0while\u00a0(isWorking || !queue.isEmpty()) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ poll() \u8d85\u65f6\u8fd4\u56de\uff08100ms\uff09\uff0c\u907f\u514d\u65e0\u9650\u7b49\u5f85\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 ClickHouseRequestBlank blank = queue.poll(100, TimeUnit.MILLISECONDS);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(blank != null) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u521b\u5efa Future \u5e76\u8bbe\u7f6e\u8d85\u65f6\uff083 \u5206\u949f\uff09\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 CompletableFuture&lt;Boolean&gt; future =\u00a0new\u00a0CompletableFuture&lt;&gt;();\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 future.orTimeout(3, TimeUnit.MINUTES);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 futures.add(future);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0try\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0send(blank, future,\u00a0new\u00a0HashSet&lt;&gt;());\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 } finally {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ final \u8fdb\u884c\u672a\u77e5\u5f02\u5e38\u515c\u5e95\uff0c\u9632\u6b62\u4e3a\u6355\u83b7\u5f02\u5e38\u9020\u6210future\u72b6\u6001\u4e0d\u5b8c\u6210\uff0c\u6c38\u4e45\u963b\u585e\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(!future.isDone()) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 future.completeExceptionally(new\u00a0RuntimeException(&quot;Unknown exception&quot;));\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 queueCounter.decrementAndGet();\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<h3>\u91cd\u8bd5\u63a7\u5236\u903b\u8f91<\/h3>\n<pre><code>private\u00a0void\u00a0handleUnsuccessfulResponse(..., Set&lt;String&gt; exceptHosts)\u00a0{\n\u00a0 \u00a0\u00a0\/\/ \u68c0\u67e5 Future \u662f\u5426\u5df2\u5b8c\u6210\uff08\u907f\u514d\u91cd\u590d\u5b8c\u6210\uff09\n\u00a0 \u00a0\u00a0if\u00a0(future.isDone()) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return;\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0if\u00a0(attemptCounter &gt;= maxRetries) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u8fbe\u5230\u6700\u5927\u91cd\u8bd5\u6b21\u6570\uff0c\u6807\u8bb0\u5931\u8d25\n\u00a0 \u00a0 \u00a0 \u00a0 future.completeExceptionally(new\u00a0RuntimeException(&quot;Max retries exceeded&quot;));\n\u00a0 \u00a0 }\u00a0else\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u9012\u5f52\u91cd\u8bd5\n\u00a0 \u00a0 \u00a0 \u00a0 requestBlank.incrementCounter();\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0send(requestBlank, future, exceptHosts);\u00a0\/\/ \u9012\u5f52\u8c03\u7528\uff0c\u6392\u9664\u5931\u8d25\u8282\u70b9\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u91cd\u8bd5\u7b56\u7565\uff1a<\/strong><\/p>\n<ul>\n<li><strong>\u9012\u5f52\u91cd\u8bd5\uff1a<\/strong> \u5931\u8d25\u540e\u9012\u5f52\u8c03\u7528\uff0c\u76f4\u5230\u6210\u529f\u6216\u8fbe\u5230\u6700\u5927\u6b21\u6570\u3002<\/li>\n<li><strong>\u5f02\u5e38\u8282\u70b9\u9694\u79bb\uff1a<\/strong> \u6bcf\u6b21\u91cd\u8bd5\u65f6\u6392\u9664\u5931\u8d25\u7684\u8282\u70b9\uff08exceptHosts\uff09\u3002<\/li>\n<li><strong>\u8d85\u65f6\u63a7\u5236\uff1a<\/strong> Future \u8d85\u65f6\uff083 \u5206\u949f\uff09\u9632\u6b62\u6c38\u4e45\u963b\u585e\u3002<\/li>\n<\/ul>\n<p><strong>\u4e3a\u4ec0\u4e48\u9012\u5f52\u91cd\u8bd5\u662f\u66f4\u597d\u7684\u9009\u62e9<\/strong><\/p>\n<p><strong>\u9012\u5f52\u91cd\u8bd5\uff08\u5f53\u524d\u5b9e\u73b0\uff09<\/strong><\/p>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603257\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<p><strong>\u961f\u5217\u91cd\u8bd5\uff08\u5047\u8bbe\u65b9\u6848\uff09<\/strong><\/p>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603258\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<h4>\u4fdd\u8bc1\u4e00\u81f4\u6027<\/h4>\n<pre><code>\u00a0\u00a0\/\/ ClickHouseWriter.java:139-158\n\u00a0\u00a0while\u00a0(!futures.isEmpty() || unProcessedCounter.get() &gt;\u00a00) {\n\u00a0 \u00a0 \u00a0 CompletableFuture&lt;Void&gt; future = FutureUtil.allOf(futures);\n\u00a0 \u00a0 \u00a0 future.get(3, TimeUnit.MINUTES); \u00a0\/\/ \u963b\u585e\u76f4\u5230\u5168\u90e8\u5b8c\u6210\n\u00a0 }<\/code><\/pre>\n<ul>\n<li>Checkpoint \u65f6\u6240\u6709\u6570\u636e\u8981\u4e48\u5168\u90e8\u6210\u529f\uff0c\u8981\u4e48\u5168\u90e8\u5931\u8d25\u3002<\/li>\n<li>\u91cd\u542f\u540e\u4e0d\u4f1a\u6709\u90e8\u5206\u6570\u636e\u91cd\u590d\u7684\u95ee\u9898\u3002<\/li>\n<\/ul>\n<h4>\u7b80\u5355\u53ef\u9760<\/h4>\n<ul>\n<li>\u4ee3\u7801\u903b\u8f91\u6e05\u6670\u3002<\/li>\n<li>\u5bf9\u4e8e\u961f\u5217\u91cd\u8bd5\u4e14\u4e0d\u91cd\u590d\uff0c\u9700\u8981\u590d\u6742\u7684\u4e8c\u9636\u6bb5\u63d0\u4ea4\uff08\u8fd9\u91cc\u6682\u4e0d\u5c55\u5f00\uff09\uff0c\u5927\u5e45\u589e\u52a0\u4ee3\u7801\u590d\u6742\u5ea6\u3002<\/li>\n<\/ul>\n<h4>\u6027\u80fd\u53ef\u63a5\u53d7<\/h4>\n<pre><code>class\u00a0WriterTask\u00a0implements Runnable {\n\u00a0 \u00a0 @Override\n\u00a0 \u00a0\u00a0public\u00a0void\u00a0run()\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0while\u00a0(isWorking || !queue.isEmpty()) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 ClickHouseRequestBlank blank = queue.poll(100, TimeUnit.MILLISECONDS);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(blank != null) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u521b\u5efa Future \u5e76\u8bbe\u7f6e 3 \u5206\u949f\u8d85\u65f6\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 CompletableFuture&lt;Boolean&gt; future =\u00a0new\u00a0CompletableFuture&lt;&gt;();\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 future.orTimeout(3, TimeUnit.MINUTES);\u00a0\/\/ \u9632\u6b62\u6c38\u4e45\u963b\u585e\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 futures.add(future);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0try\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0send(blank, future,\u00a0new\u00a0HashSet&lt;&gt;());\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 } finally {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(!future.isDone()) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 future.completeExceptionally(new\u00a0RuntimeException(&quot;Timeout&quot;));\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 queueCounter.decrementAndGet();\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<ul>\n<li>\u867d\u7136\u963b\u585e\uff0c\u4f46\u6709\u8d85\u65f6\u4fdd\u62a4\u3002<\/li>\n<li>ClickHouse \u5199\u5165\u901a\u5e38\u5f88\u5feb\uff08\u79d2\u7ea7\uff09\u3002<\/li>\n<li>\u7f51\u7edc\u6545\u969c\u65f6\u91cd\u8bd5\u4e5f\u5408\u7406\u3002<\/li>\n<\/ul>\n<h4>\u907f\u5f00\u6545\u969c\u8282\u70b9<\/h4>\n<pre><code>\u00a0 \/\/ ClickHouseWriter.java:259-260\n\u00a0 HikariDataSource dataSource =\u00a0getNextDataSource(exceptHosts);<\/code><\/pre>\n<ul>\n<li>\u9012\u5f52\u65f6\u53ef\u4ee5\u4f20\u9012 exceptHosts\u3002<\/li>\n<li>\u81ea\u52a8\u907f\u5f00\u5931\u8d25\u7684\u8282\u70b9\u3002<\/li>\n<li>\u63d0\u9ad8\u6210\u529f\u7387\u3002<\/li>\n<\/ul>\n<h3>\u5f02\u5e38\u8282\u70b9\u5254\u9664<\/h3>\n<pre><code>\/\/ \u7279\u6b8a\u9519\u8bef\u7801\u5217\u8868\uff08\u81ea\u52a8\u52a0\u5165\u9ed1\u540d\u5355\uff09\nprivate\u00a0final\u00a0List&lt;Integer&gt; ignoreHostCodes = Arrays.asList(210,\u00a01002);\npublic\u00a0HikariDataSource getNextDataSource(Set&lt;String&gt; exceptionHosts) {\n\u00a0 \u00a0\u00a0if\u00a0(CollectionUtils.isNotEmpty(exceptionHosts)) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u8fc7\u6ee4\u5f02\u5e38\u8282\u70b9\n\u00a0 \u00a0 \u00a0 \u00a0 List&lt;HikariDataSource&gt; healthyHosts = clickHouseDataSources.stream()\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 .filter(ds -&gt; !exceptionHosts.contains(getHostFromUrl(ds)))\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 .collect(Collectors.toList());\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(CollectionUtils.isEmpty(healthyHosts)) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0null;\u00a0\/\/ \u6240\u6709\u8282\u70b9\u90fd\u5f02\u5e38\n\u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0healthyHosts.get(random.nextInt(healthyHosts.size()));\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0\/\/ \u6b63\u5e38\u8f6e\u8be2\uff08\u5df2 shuffle\uff0c\u907f\u514d\u70ed\u70b9\uff09\n\u00a0 \u00a0\u00a0return\u00a0clickHouseDataSources.get(currentRandom.getAndIncrement() % size);\n}<\/code><\/pre>\n<p><strong>\u6545\u969c\u8282\u70b9\u5254\u9664\u7b56\u7565\uff1a<\/strong><\/p>\n<ol>\n<li><strong>\u9519\u8bef\u7801 210\uff08\u7f51\u7edc\u5f02\u5e38\uff09\uff1a<\/strong> \u81ea\u52a8\u52a0\u5165\u9ed1\u540d\u5355\u3002<\/li>\n<li><strong>\u9519\u8bef\u7801 1002\uff08\u8fde\u63a5\u6c60\u5f02\u5e38\uff09\uff1a<\/strong> \u81ea\u52a8\u52a0\u5165\u9ed1\u540d\u5355\u3002<\/li>\n<li><strong>APM \u76d1\u63a7\uff1a<\/strong> \u78c1\u76d8 &gt;= 90%\u3001HTTP \u8fde\u63a5 &gt;= 50 \u7684\u8282\u70b9\u3002<\/li>\n<li><strong>\u624b\u52a8\u914d\u7f6e\uff1a<\/strong> \u901a\u8fc7 Redis \u914d\u7f6e\u5254\u9664\u3002<\/li>\n<\/ol>\n<p><strong>\u6062\u590d\u673a\u5236\uff1a<\/strong><\/p>\n<ul>\n<li>LoadingCache \u5b9a\u65f6\u5237\u65b0\uff081 \u5206\u949f\uff09\u3002<\/li>\n<li>\u8282\u70b9\u6062\u590d\u5065\u5eb7\u540e\u81ea\u52a8\u4ece\u9ed1\u540d\u5355\u79fb\u9664\u3002<\/li>\n<\/ul>\n<h3>\u91cd\u8bd5\u6d41\u7a0b\u56fe<\/h3>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603259\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<h2>\u516b\u3001\u5f02\u5e38\u5904\u7406\u6a21\u5f0f<\/h2>\n<h3>\u4e24\u79cd Sink \u6a21\u5f0f<\/h3>\n<pre><code>public\u00a0Sink\u00a0buildSink(String\u00a0targetTable,\u00a0String\u00a0targetCount, int maxBufferSize) {\n\u00a0 \u00a0\u00a0IClickHouseSinkBuffer\u00a0buffer =\u00a0ClickHouseShardSinkBuffer.Builder\n\u00a0 \u00a0 \u00a0 \u00a0 .aClickHouseSinkBuffer()\n\u00a0 \u00a0 \u00a0 \u00a0 .withTargetTable(targetTable)\n\u00a0 \u00a0 \u00a0 \u00a0 .withMaxFlushBufferSize(maxBufferSize)\n\u00a0 \u00a0 \u00a0 \u00a0 .withClickHouseShardStrategy(new\u00a0LogClickHouseShardStrategy(targetTable, count))\n\u00a0 \u00a0 \u00a0 \u00a0 .build(clickHouseWriter);\n\u00a0 \u00a0\u00a0\/\/ \u6839\u636e\u914d\u7f6e\u9009\u62e9\u6a21\u5f0f\n\u00a0 \u00a0\u00a0if\u00a0(ignoringClickHouseSendingExceptionEnabled) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0new\u00a0UnexceptionableSink(buffer); \u00a0\/\/ \u5ffd\u7565\u5f02\u5e38\n\u00a0 \u00a0 }\u00a0else\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0new\u00a0ExceptionsThrowableSink(buffer);\u00a0\/\/ \u629b\u51fa\u5f02\u5e38\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<h3>UnexceptionableSink\uff08\u5ffd\u7565\u5f02\u5e38 &#8211; At-Most-Once\uff09<\/h3>\n<pre><code>public\u00a0class\u00a0UnexceptionableSink\u00a0implements\u00a0Sink&lt;LogModel&gt; {\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0IClickHouseSinkBuffer&lt;LogModel&gt; buffer;\n\u00a0 \u00a0\u00a0@Override\n\u00a0 \u00a0\u00a0public\u00a0void\u00a0put(LogModel\u00a0message) {\n\u00a0 \u00a0 \u00a0 \u00a0 buffer.put(message); \u00a0\/\/ \u4e0d\u68c0\u67e5 Future \u72b6\u6001\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0@Override\n\u00a0 \u00a0\u00a0public\u00a0void\u00a0flush() {\n\u00a0 \u00a0 \u00a0 \u00a0 buffer.flush();\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u9002\u7528\u573a\u666f\uff1a<\/strong><\/p>\n<ul>\n<li>\u5141\u8bb8\u90e8\u5206\u6570\u636e\u4e22\u5931\u3002<\/li>\n<li>\u4e0d\u5e0c\u671b\u56e0\u5199\u5165\u5f02\u5e38\u5bfc\u81f4\u4efb\u52a1\u5931\u8d25\u3002<\/li>\n<li>\u5bf9\u6570\u636e\u51c6\u786e\u6027\u8981\u6c42\u4e0d\u9ad8\uff08\u5982\u65e5\u5fd7\u7edf\u8ba1\uff09\u3002<\/li>\n<\/ul>\n<p><strong>\u8bed\u4e49\u4fdd\u8bc1\uff1aAt-Most-Once\uff08\u6700\u591a\u4e00\u6b21\uff09<\/strong><\/p>\n<h3>ExceptionsThrowableSink\uff08\u629b\u51fa\u5f02\u5e38 &#8211; At-Least-Once\uff09<\/h3>\n<pre><code>public\u00a0class\u00a0ExceptionsThrowableSink\u00a0implements\u00a0Sink&lt;LogModel&gt; {\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0IClickHouseSinkBuffer&lt;LogModel&gt; buffer;\n\u00a0 \u00a0\u00a0@Override\n\u00a0 \u00a0\u00a0public\u00a0void\u00a0put(LogModel message)\u00a0throws\u00a0ExecutionException, InterruptedException {\n\u00a0 \u00a0 \u00a0 \u00a0 buffer.put(message);\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u6bcf\u6b21\u5199\u5165\u90fd\u68c0\u67e5 Future \u72b6\u6001\n\u00a0 \u00a0 \u00a0 \u00a0 buffer.assertFuturesNotFailedYet();\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0@Override\n\u00a0 \u00a0\u00a0public\u00a0void\u00a0flush()\u00a0throws\u00a0ExecutionException, InterruptedException {\n\u00a0 \u00a0 \u00a0 \u00a0 buffer.flush();\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>Future \u72b6\u6001\u68c0\u67e5\uff1a<\/strong><\/p>\n<pre><code>public\u00a0void\u00a0assertFuturesNotFailedYet()\u00a0throws ExecutionException, InterruptedException\u00a0{\n\u00a0 \u00a0 CompletableFuture&lt;Void&gt; future = FutureUtil.allOf(futures);\n\u00a0 \u00a0\u00a0\/\/ \u975e\u963b\u585e\u68c0\u67e5\n\u00a0 \u00a0\u00a0if\u00a0(future.isCompletedExceptionally()) {\n\u00a0 \u00a0 \u00a0 \u00a0 logger.error(&quot;There is something wrong with the future. exist sink now&quot;);\n\u00a0 \u00a0 \u00a0 \u00a0 future.get();\u00a0\/\/ \u629b\u51fa\u5f02\u5e38\uff0c\u5bfc\u81f4 Flink \u4efb\u52a1\u5931\u8d25\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u9002\u7528\u573a\u666f\uff1a<\/strong><\/p>\n<ul>\n<li>\u6570\u636e\u51c6\u786e\u6027\u8981\u6c42\u9ad8\u3002<\/li>\n<li>\u9700\u8981\u4fdd\u8bc1\u6240\u6709\u6570\u636e\u5199\u5165\u6210\u529f\u3002<\/li>\n<li>\u5f02\u5e38\u65f6\u5e0c\u671b Flink \u4efb\u52a1\u5931\u8d25\u5e76\u91cd\u542f\u3002<\/li>\n<\/ul>\n<p><strong>\u8bed\u4e49\u4fdd\u8bc1\uff1aAt-Least-Once\uff08\u81f3\u5c11\u4e00\u6b21\uff09<\/strong><\/p>\n<h3>Future \u6e05\u7406\u7b56\u7565\u4e0e\u5e76\u53d1\u63a7\u5236<\/h3>\n<h4>\u5b9a\u65f6\u68c0\u67e5\u5668<\/h4>\n<pre><code>public\u00a0class\u00a0ClickHouseSinkScheduledCheckerAndCleaner\u00a0{\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0ScheduledExecutorService scheduledExecutorService;\n\u00a0 \u00a0\u00a0private\u00a0final\u00a0List&lt;CompletableFuture&lt;Boolean&gt;&gt; futures;\n\u00a0 \u00a0\u00a0\/\/ \u26a0\ufe0f volatile \u4fdd\u8bc1\u591a\u7ebf\u7a0b\u53ef\u89c1\u6027\uff08\u5173\u952e\u5e76\u53d1\u63a7\u5236\u70b9\uff09\n\u00a0 \u00a0\u00a0private\u00a0volatile\u00a0boolean\u00a0isFlushing\u00a0=\u00a0false;\n\u00a0 \u00a0\u00a0public\u00a0ClickHouseSinkScheduledCheckerAndCleaner(...)\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u5355\u7ebf\u7a0b\u5b9a\u65f6\u6267\u884c\u5668\n\u00a0 \u00a0 \u00a0 \u00a0 scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(factory);\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u5b9a\u65f6\u6267\u884c\u6e05\u7406\u4efb\u52a1\uff08\u6bcf\u9694 checkTimeout \u79d2\uff0c\u9ed8\u8ba4 30 \u79d2\uff09\n\u00a0 \u00a0 \u00a0 \u00a0 scheduledExecutorService.scheduleWithFixedDelay(getTask(), ...);\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0private\u00a0Runnable\u00a0getTask()\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return\u00a0() -&gt; {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0synchronized\u00a0(this) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/  \u5173\u952e\uff1a\u68c0\u67e5\u662f\u5426\u6b63\u5728 flush\uff0c\u907f\u514d\u5e76\u53d1\u51b2\u7a81\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(isFlushing) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0return;\u00a0\/\/ Checkpoint \u671f\u95f4\u6682\u505c\u6e05\u7406\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ 1\ufe0f\u20e3 \u6e05\u7406\u5df2\u5b8c\u6210\u7684 Future\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 futures.removeIf(filter);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ 2\ufe0f\u20e3 \u89e6\u53d1\u6240\u6709 Buffer \u7684 flush\uff08\u68c0\u67e5\u662f\u5426\u9700\u8981\u5199\u5165\uff09\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 clickHouseSinkBuffers.forEach(IClickHouseSinkBuffer::tryAddToQueue);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 };\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0\/\/ Checkpoint flush \u524d\u8c03\u7528\uff08\u6682\u505c cleaner\uff09\n\u00a0 \u00a0\u00a0public\u00a0synchronized\u00a0void\u00a0beforeFlush()\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 isFlushing =\u00a0true;\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0\/\/ Checkpoint flush \u540e\u8c03\u7528\uff08\u6062\u590d cleaner\uff09\n\u00a0 \u00a0\u00a0public\u00a0synchronized\u00a0void\u00a0afterFlush()\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 isFlushing =\u00a0false;\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u6838\u5fc3\u8bbe\u8ba1\uff1a<\/strong><\/p>\n<ul>\n<li><strong>volatile boolean isFlushing\uff1a<\/strong> \u6807\u5fd7\u4f4d\uff0c\u534f\u8c03 cleaner \u4e0e checkpoint \u7ebf\u7a0b\u3002<\/li>\n<li><strong>synchronized (this)\uff1a<\/strong> \u4fdd\u8bc1\u539f\u5b50\u6027\uff0c\u907f\u514d\u5e76\u53d1\u51b2\u7a81\u3002<\/li>\n<li><strong>\u5355\u7ebf\u7a0b\u6267\u884c\u5668\uff1a<\/strong> \u907f\u514d cleaner \u5185\u90e8\u5e76\u53d1\u95ee\u9898\u3002<\/li>\n<\/ul>\n<h4>\u5e76\u53d1\u63a7\u5236\u673a\u5236<\/h4>\n<p><strong>\u95ee\u9898\u573a\u666f\uff1a<\/strong><\/p>\n<pre><code>\u65f6\u95f4\u8f74\u51b2\u7a81\uff1a\nT1: Cleaner \u7ebf\u7a0b\u6b63\u5728\u6267\u884c\u00a0tryAddToQueue()\nT2: Checkpoint \u89e6\u53d1\uff0c\u8c03\u7528 sink.flush()\nT3: Cleaner \u540c\u65f6\u4e5f\u5728\u6267\u884c\u00a0tryAddToQueue()\n\u00a0 \u00a0 \u251c\u2500 \u53ef\u80fd\u5bfc\u81f4\uff1a\u6570\u636e\u91cd\u590d\u5199\u5165\n\u00a0 \u00a0 \u251c\u2500 \u53ef\u80fd\u5bfc\u81f4\uff1aBuffer \u6e05\u7a7a\u987a\u5e8f\u6df7\u4e71\n\u00a0 \u00a0 \u2514\u2500 \u53ef\u80fd\u5bfc\u81f4\uff1aFuture \u72b6\u6001\u4e0d\u4e00\u81f4<\/code><\/pre>\n<p><strong>\u89e3\u51b3\u65b9\u6848\uff1a<\/strong><\/p>\n<pre><code>\/\/ ClickHouseSinkManager.flush()\npublic\u00a0void\u00a0flush()\u00a0{\n\u00a0 \u00a0\u00a0\/\/ 1\ufe0f\u20e3 \u6682\u505c\u5b9a\u65f6\u6e05\u7406\u4efb\u52a1\uff08\u8bbe\u7f6e\u6807\u5fd7\uff09\n\u00a0 \u00a0 clickHouseSinkScheduledCheckerAndCleaner.beforeFlush();\u00a0\/\/ isFlushing = true\n\u00a0 \u00a0\u00a0try\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ 2\ufe0f\u20e3 \u6267\u884c flush\uff08\u6b64\u65f6 cleaner \u7ebf\u7a0b\u4f1a\u8df3\u8fc7\u6267\u884c\uff09\n\u00a0 \u00a0 \u00a0 \u00a0 clickHouseWriter.waitUntilAllFuturesDone(false,\u00a0false);\n\u00a0 \u00a0 }\u00a0finally\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ 3\ufe0f\u20e3 \u6062\u590d\u5b9a\u65f6\u6e05\u7406\u4efb\u52a1\n\u00a0 \u00a0 \u00a0 \u00a0 clickHouseSinkScheduledCheckerAndCleaner.afterFlush();\u00a0\/\/ isFlushing = false\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p>\u5e76\u53d1\u63a7\u5236\u6d41\u7a0b\uff1a<\/p>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603260\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<p><strong>\u5173\u952e\u8bbe\u8ba1\u70b9\uff1a<\/strong><\/p>\n<ol>\n<li><strong>volatile \u4fdd\u8bc1\u53ef\u89c1\u6027\uff1a<\/strong> isFlushing \u4f7f\u7528 volatile\uff0c\u786e\u4fdd\u591a\u7ebf\u7a0b\u95f4\u7684\u53ef\u89c1\u6027\u3002<\/li>\n<li><strong>synchronized \u4fdd\u8bc1\u539f\u5b50\u6027\uff1a<\/strong> getTask() \u6574\u4e2a\u65b9\u6cd5\u4f53\u4f7f\u7528 synchronized (this)\u3002<\/li>\n<li><strong>\u6807\u5fd7\u4f4d\u534f\u8c03\uff1a<\/strong> \u901a\u8fc7 isFlushing \u6807\u5fd7\u5b9e\u73b0\u4e24\u4e2a\u7ebf\u7a0b\u95f4\u7684\u534f\u8c03\u3002<\/li>\n<li><strong>finally \u786e\u4fdd\u6062\u590d\uff1a<\/strong> \u5373\u4f7f waitUntilAllFuturesDone() \u5f02\u5e38\uff0c\u4e5f\u4f1a\u5728 finally \u4e2d\u6062\u590d cleaner\u3002<\/li>\n<\/ol>\n<p><strong>\u907f\u514d\u7684\u5e76\u53d1\u95ee\u9898\uff1a<\/strong><\/p>\n<ul>\n<li><strong>\u6570\u636e\u91cd\u590d\u5199\u5165\uff1a<\/strong> Cleaner \u548c Checkpoint \u540c\u65f6 flush\u3002<\/li>\n<li><strong>Buffer \u72b6\u6001\u4e0d\u4e00\u81f4\uff1a<\/strong> \u4e00\u8fb9\u6e05\u7a7a\u4e00\u8fb9\u5199\u5165\u3002<\/li>\n<li><strong>Future \u6e05\u7406\u51b2\u7a81\uff1a<\/strong> \u6b63\u5728\u4f7f\u7528\u7684 Future \u88ab\u6e05\u7406\u3002<\/li>\n<\/ul>\n<p><strong>\u6027\u80fd\u5f71\u54cd\uff1a<\/strong><\/p>\n<ul>\n<li>Checkpoint flush \u671f\u95f4\uff0ccleaner \u6682\u505c\u6267\u884c\uff08\u901a\u5e38 1-3 \u79d2\uff09\u3002<\/li>\n<li>Cleaner \u8df3\u8fc7\u7684\u5468\u671f\u4f1a\u5728\u4e0b\u6b21\u6b63\u5e38\u6267\u884c\u65f6\u8865\u507f\u3002<\/li>\n<li>\u5bf9\u6574\u4f53\u541e\u5410\u5f71\u54cd\u6781\u5c0f\uff08cleaner \u95f4\u9694\u901a\u5e38 30 \u79d2\uff09\u3002<\/li>\n<\/ul>\n<h2>\u4e5d\u3001Checkpoint \u8bed\u4e49\u4fdd\u8bc1<\/h2>\n<h3>\u4e3a\u4ec0\u4e48 Checkpoint \u65f6\u5fc5\u987b Flush\uff1f<\/h3>\n<h4>\u4e0d Flush \u7684\u540e\u679c<\/h4>\n<p><strong>\u4e0dFlush\u5bfc\u81f4\u6570\u636e\u6c38\u4e45\u4e22\u5931<\/strong><\/p>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603261\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<h4>\u6b63\u786e\u505a\u6cd5<\/h4>\n<pre><code>@Override\npublic\u00a0void\u00a0snapshotState(FunctionSnapshotContext context)\u00a0throws\u00a0Exception {\n\u00a0 \u00a0 logger.info(&quot;start doing snapshot. flush sink to ck&quot;);\n\u00a0 \u00a0\u00a0\/\/ 1. \u5148 flush buffer\uff08\u5c06\u5185\u5b58\u6570\u636e\u5199\u5165 ClickHouse\uff09\n\u00a0 \u00a0\u00a0if\u00a0(sink !=\u00a0null) {\n\u00a0 \u00a0 \u00a0 \u00a0 sink.flush();\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0\/\/ 2. \u7b49\u5f85\u6240\u6709\u5199\u5165\u5b8c\u6210\n\u00a0 \u00a0\u00a0if\u00a0(sinkManager !=\u00a0null\u00a0&amp;&amp; !sinkManager.isClosed()) {\n\u00a0 \u00a0 \u00a0 \u00a0 sinkManager.flush();\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0\/\/ \u6b64\u65f6 Checkpoint \u624d\u80fd\u6807\u8bb0\u4e3a\u6210\u529f\n\u00a0 \u00a0 logger.info(&quot;doing snapshot. flush sink to ck&quot;);\n}<\/code><\/pre>\n<h3>Flush \u5b9e\u73b0\u4e0e\u5e76\u53d1\u534f\u8c03<\/h3>\n<pre><code>public\u00a0class\u00a0ClickHouseSinkManager\u00a0{\n\u00a0 \u00a0\u00a0public\u00a0void\u00a0flush()\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/  \u6b65\u9aa41\uff1a\u6682\u505c\u5b9a\u65f6\u6e05\u7406\u4efb\u52a1\n\u00a0 \u00a0 \u00a0 \u00a0 clickHouseSinkScheduledCheckerAndCleaner.beforeFlush();\u00a0\/\/ isFlushing = true\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0try\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/  \u6b65\u9aa42\uff1a\u6267\u884c buffer flush + \u7b49\u5f85\u6240\u6709\u5199\u5165\u5b8c\u6210\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 clickHouseWriter.waitUntilAllFuturesDone(false,\u00a0false);\n\u00a0 \u00a0 \u00a0 \u00a0 }\u00a0finally\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/  \u6b65\u9aa43\uff1a\u6062\u590d\u5b9a\u65f6\u6e05\u7406\u4efb\u52a1\uff08finally \u786e\u4fdd\u6267\u884c\uff09\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 clickHouseSinkScheduledCheckerAndCleaner.afterFlush();\u00a0\/\/ isFlushing = false\n\u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u5e76\u53d1\u534f\u8c03\u8be6\u89e3\uff1a<\/strong><\/p>\n<pre><code>\/\/ cleaner \u7ebf\u7a0b\u6267\u884c\u6d41\u7a0b\nsynchronized (this) {\n\u00a0 \u00a0\u00a0if\u00a0(isFlushing) {\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0return;\u00a0\/\/ Checkpoint \u671f\u95f4\u8df3\u8fc7\u672c\u6b21\u6267\u884c\n\u00a0 \u00a0 }\n\u00a0 \u00a0\u00a0\/\/ \u6b63\u5e38\u6267\u884c\uff1a\u6e05\u7406\u5df2\u5b8c\u6210\u7684 Future + \u89e6\u53d1 Buffer flush\n\u00a0 \u00a0 futures.removeIf(filter);\n\u00a0 \u00a0 buffers.forEach(Buffer::tryAddToQueue);\n}<\/code><\/pre>\n<p><strong>\u5173\u952e\u70b9\uff1a<\/strong><\/p>\n<ul>\n<li><strong>volatile \u53ef\u89c1\u6027\uff1a<\/strong> isFlushing \u4f7f\u7528 volatile \u786e\u4fdd cleaner \u7ebf\u7a0b\u7acb\u5373\u770b\u5230\u72b6\u6001\u53d8\u5316\u3002<\/li>\n<li><strong>synchronized\u4e92\u65a5\uff1a<\/strong> getTask()\u65b9\u6cd5\u4f53\u4f7f\u7528 synchronized (this) \u786e\u4fdd\u539f\u5b50\u6027\u3002<\/li>\n<li><strong>\u6807\u5fd7\u4f4d\u534f\u8c03\uff1a<\/strong> \u901a\u8fc7 beforeFlush() \/ afterFlush() \u7ba1\u7406\u6807\u5fd7\u4f4d\u3002<\/li>\n<li><strong>finally \u4fdd\u8bc1\u6062\u590d\uff1a<\/strong> \u5373\u4f7f flush \u5f02\u5e38\uff0c\u4e5f\u4f1a\u5728 finally \u4e2d\u6062\u590d cleaner\u3002<\/li>\n<\/ul>\n<h3>\u7b49\u5f85\u6240\u6709 Future \u5b8c\u6210<\/h3>\n<pre><code>public\u00a0synchronized\u00a0void\u00a0waitUntilAllFuturesDone(boolean\u00a0stopWriters,\u00a0boolean\u00a0clearFutures) {\n\u00a0 \u00a0\u00a0try\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u5faa\u73af\u7b49\u5f85\uff1a\u76f4\u5230\u6240\u6709 Future \u5b8c\u6210 + \u961f\u5217\u6e05\u7a7a\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0while\u00a0(!futures.isEmpty() || unProcessedCounter.get() &gt;\u00a00) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0CompletableFuture&lt;Void&gt; all =\u00a0FutureUtil.allOf(futures);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u6700\u591a\u7b49\u5f85 3 \u5206\u949f\uff08\u4e0e Future \u8d85\u65f6\u4e00\u81f4\uff09\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 all.get(3,\u00a0TimeUnit.MINUTES);\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u79fb\u9664\u5df2\u5b8c\u6210\u7684 Future\uff08\u975e\u5f02\u5e38\uff09\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 futures.removeIf(f -&gt; f.isDone() &amp;&amp; !f.isCompletedExceptionally());\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u68c0\u67e5\u662f\u5426\u6709\u5f02\u5e38 Future\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(anyFutureFailed()) {\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0break;\u00a0\/\/ \u6709\u5f02\u5e38\u5219\u9000\u51fa\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 \u00a0 \u00a0 }\n\u00a0 \u00a0 }\u00a0finally\u00a0{\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(stopWriters)\u00a0stopWriters();\n\u00a0 \u00a0 \u00a0 \u00a0\u00a0if\u00a0(clearFutures) futures.clear();\n\u00a0 \u00a0 }\n}<\/code><\/pre>\n<p><strong>\u5173\u952e\u903b\u8f91\uff1a<\/strong><\/p>\n<ul>\n<li>\u5faa\u73af\u7b49\u5f85\u76f4\u5230\u6240\u6709 Future \u5b8c\u6210 + \u961f\u5217\u6e05\u7a7a\u3002<\/li>\n<li>\u8d85\u65f6 3 \u5206\u949f\uff08\u4e0e Future \u8d85\u65f6\u4e00\u81f4\uff09\u3002<\/li>\n<li>\u79fb\u9664\u5df2\u5b8c\u6210\u7684\u975e\u5f02\u5e38 Future\u3002<\/li>\n<li>\u6709\u5f02\u5e38\u65f6\u9000\u51fa\u5faa\u73af\u3002<\/li>\n<\/ul>\n<h3>\u4e09\u79cd Flush \u89e6\u53d1\u65b9\u5f0f\u5bf9\u6bd4<\/h3>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603262\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<h3>Checkpoint \u53c2\u6570\u914d\u7f6e<\/h3>\n<pre><code>\/\/ Checkpoint \u914d\u7f6e\u5efa\u8bae\nStreamExecutionEnvironment\u00a0env\u00a0=\u00a0StreamExecutionEnvironment.getExecutionEnvironment();\n\/\/ \u542f\u7528 Checkpoint\uff08\u95f4\u9694 1 \u5206\u949f\uff09\nenv.enableCheckpointing(60000);\n\/\/ Checkpoint \u8d85\u65f6\uff08\u5fc5\u987b\u5927\u4e8e Future \u8d85\u65f6 + \u91cd\u8bd5\u65f6\u95f4\uff09\n\/\/ \u5efa\u8bae\uff1aCheckpointTimeout &gt; FutureTimeout * MaxRetries\nenv.getCheckpointConfig().setCheckpointTimeout(600000);\u00a0\/\/ 10 \u5206\u949f\n\/\/ \u4e00\u81f4\u6027\u6a21\u5f0f\nenv.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);\n\/\/ \u6700\u5c0f\u95f4\u9694\uff08\u907f\u514d\u8fc7\u4e8e\u9891\u7e41\uff09\nenv.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);\u00a0\/\/ 30 \u79d2\n\/\/ \u6700\u5927\u5e76\u53d1 Checkpoint \u6570\nenv.getCheckpointConfig().setMaxConcurrentCheckpoints(1);<\/code><\/pre>\n<h3>\u8bed\u4e49\u4fdd\u8bc1<\/h3>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603263\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<p><strong>\u63a8\u8350\u914d\u7f6e\uff1a<\/strong><\/p>\n<p><strong>\u751f\u4ea7\u73af\u5883\uff1a<\/strong> \u4f7f\u7528 ExceptionsThrowableSink + Checkpoint\u3002<\/p>\n<p><strong>\u5141\u8bb8\u90e8\u5206\u4e22\u5931\uff1a<\/strong> \u4f7f\u7528 UnexceptionableSink\u3002<\/p>\n<h2>\u5341\u3001\u6700\u4f73\u5b9e\u8df5\u4e0e\u8c03\u4f18<\/h2>\n<h3>\u751f\u4ea7\u914d\u7f6e<\/h3>\n<pre><code>\/\/ ========== ClickHouse \u8fde\u63a5\u53c2\u6570 ==========\nclickhouse.sink.target-table = tb_logs_local\nclickhouse.sink.max-buffer-size =\u00a0104857600\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u6279\u6b21\u5927\u5c0f\nclickhouse.sink.table-count =\u00a00\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ 0 \u8868\u793a\u4e0d\u5206\u8868\n\/\/ ========== \u5199\u5165\u6027\u80fd\u53c2\u6570 ==========\nclickhouse.sink.num-writers =\u00a010\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\/\/ \u5199\u5165\u7ebf\u7a0b\u6570\nclickhouse.sink.queue-max-capacity =\u00a010\u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u961f\u5217\u5bb9\u91cf\nclickhouse.sink.timeout-sec =\u00a030\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\/\/ flush \u8d85\u65f6\nclickhouse.sink.retries =\u00a010\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\/\/ \u6700\u5927\u91cd\u8bd5\u6b21\u6570\nclickhouse.sink.check.timeout-sec =\u00a030\u00a0 \u00a0 \u00a0 \u00a0 \u00a0\/\/ \u5b9a\u65f6\u68c0\u67e5\u95f4\u9694\n\/\/ ========== \u5f02\u5e38\u5904\u7406\u53c2\u6570 ==========\nclickhouse.sink.ignoring-clickhouse-sending-exception-enabled =\u00a0false\nclickhouse.sink.local-address-enabled =\u00a0true\u00a0 \u00a0\/\/ \u542f\u7528\u672c\u5730\u8868\u5199\u5165\n\/\/ ========== ClickHouse \u96c6\u7fa4\u914d\u7f6e ==========\nclickhouse.access.hosts =\u00a0192.168.1.1:8123,192.168.1.2:8123,192.168.1.3:8123\nclickhouse.access.user =\u00a0default\nclickhouse.access.password = ***\nclickhouse.access.database = dw_xx_xx\nclickhouse.access.cluster =\u00a0default\n\/\/ ========== HikariCP \u8fde\u63a5\u6c60\u914d\u7f6e ==========\nconnectionTimeout =\u00a030000\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u8fde\u63a5\u8d85\u65f6 30s\nmaximumPoolSize =\u00a020\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\/\/ \u6700\u5927\u8fde\u63a5\u6570 20\nminimumIdle =\u00a02\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ \u6700\u5c0f\u7a7a\u95f2 2\nsocket_timeout =\u00a0180000\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0\u00a0\/\/ Socket \u8d85\u65f6 3mi<\/code><\/pre>\n<h3>\u6027\u80fd\u8c03\u4f18<\/h3>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603264\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<h3>\u6545\u969c\u6392\u67e5<\/h3>\n<p><img decoding=\"async\"   referrerpolicy=\"no-referrer\" src=\"https:\/\/segmentfault.com\/img\/remote\/1460000047603265\" alt=\"\" title=\"\" loading=\"lazy\"><\/p>\n<h2>\u5341\u4e00\u3001\u603b\u7ed3<\/h2>\n<p>\u672c\u6587\u6df1\u5165\u5206\u6790\u4e86 Flink ClickHouse Sink \u7684\u5b9e\u73b0\u65b9\u6848\uff0c\u6838\u5fc3\u4eae\u70b9\u5305\u62ec\uff1a<\/p>\n<h3>\u6280\u672f\u4eae\u70b9<\/h3>\n<ul>\n<li><strong>\u8fde\u63a5\u6c60\u9009\u578b\uff1a<\/strong> \u4f7f\u7528 HikariCP\uff0c\u6027\u80fd\u4f18\u5f02\uff0c\u8fde\u63a5\u7ba1\u7406\u53ef\u9760\u3002<\/li>\n<li><strong>Future \u8d85\u65f6\u63a7\u5236\uff1a<\/strong> orTimeout(3min) \u9632\u6b62\u6c38\u4e45\u963b\u585e\u3002<\/li>\n<li><strong>\u663e\u5f0f\u8d44\u6e90\u7ba1\u7406\uff1a<\/strong> Connection \u548c PreparedStatement \u663e\u5f0f\u5173\u95ed\uff0c\u9632\u6b62\u8fde\u63a5\u6cc4\u6f0f\u3002<\/li>\n<li><strong>\u8d1f\u8f7d\u5747\u8861\u4f18\u5316\uff1a<\/strong> Shuffle \u521d\u59cb\u5316 + \u8f6e\u8be2\u9009\u62e9\uff0c\u907f\u514d\u70ed\u70b9\u3002<\/li>\n<li><strong>\u5f02\u5e38\u5904\u7406\u589e\u5f3a\uff1a<\/strong> future.isDone() \u68c0\u67e5\uff0c\u907f\u514d\u91cd\u590d\u5b8c\u6210\u3002<\/li>\n<li><strong>\u672c\u5730\u8868\u5199\u5165\uff1a<\/strong> \u52a8\u6001\u8282\u70b9\u53d1\u73b0 + \u6545\u969c\u5254\u9664\uff0c\u5199\u5165\u6027\u80fd\u63d0\u5347\u3002<\/li>\n<li><strong>\u5206\u7247\u7b56\u7565\uff1a<\/strong> \u6309\u8868\uff08\u5e94\u7528\uff09\u7ef4\u5ea6\u8def\u7531\uff0c\u72ec\u7acb\u7f13\u51b2\u548c\u9694\u79bb\u3002<\/li>\n<li><strong>\u6512\u6279\u4f18\u5316\uff1a<\/strong> \u53cc\u89e6\u53d1\u673a\u5236\uff08\u5927\u5c0f + \u8d85\u65f6\uff09+ \u968f\u673a\u6296\u52a8\u3002<\/li>\n<li><strong>\u6d41\u91cf\u63a7\u5236\uff1a<\/strong> \u6709\u754c\u961f\u5217 + \u7ebf\u7a0b\u6c60\uff0c\u5b9e\u73b0\u80cc\u538b\u3002<\/li>\n<li><strong>\u5065\u58ee\u91cd\u8bd5\uff1a<\/strong> \u9012\u5f52\u91cd\u8bd5 + \u5f02\u5e38\u8282\u70b9\u5254\u9664 + \u6700\u5927\u91cd\u8bd5\u9650\u5236\u3002<\/li>\n<\/ul>\n<h3>Checkpoint \u8bed\u4e49<\/h3>\n<ul>\n<li><strong>At-Least-Once\uff1a<\/strong> ExceptionsThrowableSink + Checkpoint\u3002<\/li>\n<li><strong>At-Most-Once\uff1a<\/strong> UnexceptionableSink\u3002<\/li>\n<li><strong>Exactly-Once\uff1a<\/strong> \u9700\u8981\u914d\u5408 ClickHouse \u4e8b\u52a1\uff08\u672a\u5b9e\u73b0\uff09\u3002<\/li>\n<\/ul>\n<h3>\u751f\u4ea7\u5efa\u8bae<\/h3>\n<ol>\n<li><strong>\u5fc5\u987b\uff1a<\/strong> Checkpoint \u65f6 flush\uff0c\u5426\u5219\u4f1a\u4e22\u6570\u636e\u3002<\/li>\n<li><strong>\u63a8\u8350\uff1a<\/strong> \u4f7f\u7528 HikariCP + \u672c\u5730\u8868\u5199\u5165\u3002<\/li>\n<li><strong>\u63a8\u8350\uff1a<\/strong> \u914d\u7f6e\u5408\u7406\u7684\u8d85\u65f6\uff08Future &lt; Socket &lt; Checkpoint\uff09\u3002<\/li>\n<li><strong>\u63a8\u8350\uff1a<\/strong> \u76d1\u63a7\u961f\u5217\u5927\u5c0f\u3001Future \u5931\u8d25\u7387\u3001\u91cd\u8bd5\u6b21\u6570\u3002<\/li>\n<\/ol>\n<p>\u8be5\u65b9\u6848\u5df2\u5728\u751f\u4ea7\u73af\u5883\u5927\u89c4\u6a21\u9a8c\u8bc1\uff0c\u80fd\u591f\u7a33\u5b9a\u652f\u6491<strong>\u767e\u4e07\u7ea7 TPS<\/strong> \u7684\u65e5\u5fd7\u5199\u5165\u573a\u666f\u3002<\/p>\n<h2>\u5f80\u671f\u56de\u987e<\/h2>\n<p>1.\u670d\u52a1\u62c6\u5206\u4e4b\u65c5\uff1a\u6d4b\u8bd5\u8fc7\u7a0b\u5168\u63ed\u79d8\uff5c\u5f97\u7269\u6280\u672f<\/p>\n<p>2.\u5927\u6a21\u578b\u7f51\u5173\uff1a\u5927\u6a21\u578b\u65f6\u4ee3\u7684\u667a\u80fd\u4ea4\u901a\u67a2\u7ebd\uff5c\u5f97\u7269\u6280\u672f<\/p>\n<p>3.\u4ece\u201c\u4eba\u6cbb\u201d\u5230\u201c\u673a\u6cbb\u201d\uff1a\u5f97\u7269\u79bb\u7ebf\u6570\u4ed3\u53d1\u5e03\u6d41\u6c34\u7ebf\u8d28\u91cf\u95e8\u7981\u5b9e\u8df5<\/p>\n<p>4.AI\u7f16\u7a0b\u5b9e\u8df5\uff1a\u4eceClaude Code\u5b9e\u8df5\u5230\u56e2\u961f\u534f\u4f5c\u7684\u4f18\u5316\u601d\u8003\uff5c\u5f97\u7269\u6280\u672f<\/p>\n<p>5.\u5165\u9009AAAI-PerFM\uff5c\u5f97\u7269\u793e\u533a\u63a8\u8350\u4e4b\u57fa\u4e8e\u5927\u8bed\u8a00\u6a21\u578b\u7684\u65b0\u9896\u6027\u63a8\u8350\u7b97\u6cd5<\/p>\n<h2>\u6587 \/\u865a\u767d<\/h2>\n<p>\u5173\u6ce8\u5f97\u7269\u6280\u672f\uff0c\u6bcf\u5468\u66f4\u65b0\u6280\u672f\u5e72\u8d27<\/p>\n<p>\u8981\u662f\u89c9\u5f97\u6587\u7ae0\u5bf9\u4f60\u6709\u5e2e\u52a9\u7684\u8bdd\uff0c\u6b22\u8fce\u8bc4\u8bba\u8f6c\u53d1\u70b9\u8d5e\uff5e<\/p>\n<p>\u672a\u7ecf\u5f97\u7269\u6280\u672f\u8bb8\u53ef\u4e25\u7981\u8f6c\u8f7d\uff0c\u5426\u5219\u4f9d\u6cd5\u8ffd\u7a76\u6cd5\u5f8b\u8d23\u4efb\u3002<\/p>\n<p><a href=\"https:\/\/segmentfault.com\/a\/1190000047603247\">https:\/\/segmentfault.com\/a\/1190000047603247<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"<p>\u4e00\u3001\u80cc\u666f\u4e0e\u75db\u70b9 \u4e1a\u52a1\u573a\u666f \u5728\u5b9e\u65f6\u5927\u6570\u636e\u5904\u7406\u573a\u666f\u4e2d\uff0cFlink + ClickHouse \u7684\u7ec4\u5408\u88ab\u5e7f\u6cdb\u5e94\u7528\u4e8e\uff1a  [&hellip;]<\/p>\n","protected":false},"author":4,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"_jetpack_memberships_contains_paid_content":false,"footnotes":""},"categories":[4],"tags":[],"class_list":["post-10021245","post","type-post","status-publish","format-standard","hentry","category-share"],"jetpack_sharing_enabled":true,"jetpack_featured_media_url":"","_links":{"self":[{"href":"https:\/\/itmax.cn\/index.php?rest_route=\/wp\/v2\/posts\/10021245","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/itmax.cn\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/itmax.cn\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/itmax.cn\/index.php?rest_route=\/wp\/v2\/users\/4"}],"replies":[{"embeddable":true,"href":"https:\/\/itmax.cn\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=10021245"}],"version-history":[{"count":0,"href":"https:\/\/itmax.cn\/index.php?rest_route=\/wp\/v2\/posts\/10021245\/revisions"}],"wp:attachment":[{"href":"https:\/\/itmax.cn\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=10021245"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/itmax.cn\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=10021245"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/itmax.cn\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=10021245"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}