分庫分表下極緻的優化

題外話

這邊說一句題外話,就是

ShardingCore

目前已經正式加入 NCC 開源組織了,也是希望架構和社群能發展的越來越好,希望為更多.netter提供解決方案和開源元件

介紹

依照慣例首先介紹本期主角:

ShardingCore

一款ef-core下高性能、輕量級針對分表分庫讀寫分離的解決方案，具有零依賴、零學習成本、零業務代碼入侵

dotnet下唯一一款全自動分表,多字段分表架構,擁有高性能,零依賴、零學習成本、零業務代碼入侵,并且支援讀寫分離動态分表分庫,同一種路由可以完全自定義的新星元件架構

你的star和點贊是我堅持下去的最大動力,一起為.net生态提供更好的解決方案

項目位址

github位址 https://github.com/dotnetcore/sharding-core

本次優化點

直奔主題來講下本次的極緻優化具體是優化了什麼,簡單說就是

CircuitBreaker

和

FastFail

斷路器CircuitBreaker

我們假設這麼一個場景,現在我們有一個訂單order表,訂單會按照月份進行分片,那麼訂單表會有如下幾個

order_202201

、

order_202202

、

order_202203

、

order_202204

、

order_202205

,假設我們有5張表。

首先我們來看一條普通的語句

select * from order where id='xxx' limit 1

這是一條普通的不能在普通的sql了,查詢第一條id是xxx的訂單，

那麼他在分表下面會如何運作

//開啟5個線程并發查詢
select * from order_202201 where id='xxx' limit 1
select * from order_202202 where id='xxx' limit 1
select * from order_202203 where id='xxx' limit 1
select * from order_202204 where id='xxx' limit 1
select * from order_202205 where id='xxx' limit 1
//查詢出來的結果在記憶體中進行聚合成一個list集合
//然後在對這個list集合進行第一條的擷取
list.Where(o=>o is not null).FirstOrDefault()

這個操作我相信很多同學都是可以了解的,稍微熟悉點分表分庫的同學應該都知道這是基本操作了,但是這個操作看似高效(時間上)但是在連接配接數上而言并不是那麼的高效,因為同一時間需要開打的連接配接數将由5個

那麼在這個背景下

ShardingCore

參考ShardingSphere 提供了更加友好的連接配接控制和記憶體聚合模式

ConnectionMode

這個張圖上我們可以清晰的看到不同的資料庫直接才用了一個并發限制,比如設定的是2,那麼在相同庫裡面的查詢将是每2個一組,進行查詢,這樣可以控制在同一個資料庫下的連接配接數,進而解決了用戶端連接配接模式下的連接配接數消耗猛烈的一個弊端。

//開啟5個線程并發查詢
{
  //并行
  select * from order_202201 where id='xxx' limit 1
  select * from order_202202 where id='xxx' limit 1
}
  //串行
{
  //并行
  select * from order_202203 where id='xxx' limit 1
  select * from order_202204 where id='xxx' limit 1
}
  //串行
{
  select * from order_202205 where id='xxx' limit 1
}
//查詢出來的結果在記憶體中進行聚合成一個list集合
//然後在對這個list集合進行第一條的擷取
list.Where(o=>o is not null).FirstOrDefault()

到目前為止這邊已經對分片的查詢優化到了一個新的高度。但是雖然我們優化了連接配接數的處理,但是就查詢速度而言基本上是沒有之前的那麼快,可以說和你分組的組數成線性增加時間的消耗。

是以到此為止

ShardingCore

又再一次

進化

出了全新的翅膀

CircuitBreaker

斷路器,我們繼續往下看

我們現在的sql是

select * from order where id='xxx' limit 1

那麼如果我們針對這個sql進行優化呢,譬如

select * from order where id='xxx' order by create_time desc limit 1

同樣是查詢第一條,添加了一個order排序那麼情況就會大大的不一樣,首先我們來觀察我們的分片查詢

//開啟5個線程并發查詢
--  select * from order_202201 where id='xxx' order by create_time desc  limit 1
--  select * from order_202202 where id='xxx' order by create_time desc  limit 1
--  select * from order_202203 where id='xxx' order by create_time desc  limit 1
--  select * from order_202204 where id='xxx' order by create_time desc  limit 1
--  select * from order_202205 where id='xxx' order by create_time desc  limit 1
-- 抛棄上述寫法

  select * from order_202205 where id='xxx' order by create_time desc  limit 1
  select * from order_202204 where id='xxx' order by create_time desc  limit 1
  select * from order_202203 where id='xxx' order by create_time desc  limit 1
  select * from order_202202 where id='xxx' order by create_time desc  limit 1
  select * from order_202201 where id='xxx' order by create_time desc  limit 1

如果在連接配接模式下那麼他們将會是2個一組,那麼我們在查詢第一組的結果後是否就可以直接抛棄掉下面的所有查詢,也就是我們隻需要查詢

select * from order_202205 where id='xxx' order by create_time desc  limit 1
  select * from order_202204 where id='xxx' order by create_time desc  limit 1

隻要他們是有傳回一個以上的資料那麼本次分片查詢将會被終止,

ShardingCore

目前的大殺器,本來年前已經開發完成了,奈何太懶隻是釋出了版本并沒有相關的說明和使用方法

CircuitBreaker

斷路器,它具有類似拉閘中斷操作的功能,這邊簡單說下linq操作下的部分方法的斷路器點在哪裡

方法名	是否支援中斷操作	中斷條件
First	支援	按順序查詢到第一個時就可以放棄其餘查詢
FirstOrDefault	支援	按順序查詢到第一個時就可以放棄其餘查詢
Last	支援	按順序倒叙查詢到第一個時就可以放棄其餘查詢
LastOrDefault	支援	按順序倒叙查詢到第一個時就可以放棄其餘查詢
Single	支援	查詢到兩個時就可以放棄,因為元素個數大于1個了需要抛錯
SingleOrDefault	支援	查詢到兩個時就可以放棄,因為元素個數大于1個了需要抛錯
Any	支援	查詢一個結果true就可以放棄其餘查詢
All	支援	查詢到一個結果fasle就可以放棄其餘查詢
Contains	支援	查詢一個結果true就可以放棄其餘查詢
Count	不支援	--
LongCount	不支援	--
Max	支援	按順序最後一條并且查詢最大字段是分片順序同字段是,max的屬性隻需要查詢一條記錄
Min	支援	按順序第一條并且查詢最小字段是分片順序同字段,min的屬性隻需要查詢一條記錄
Average	不支援	--
Sum	不支援	--

這邊其實隻有三個操作是任何狀态下都可以支援中斷,其餘操作需要在額外條件順序查詢的情況下才可以,并且我們本次查詢分片涉及到過多的字尾表那麼性能和資源的利用将會大大提升

查詢配置

廢話不多說我們開始以mysql作為本次案例(不要問我為什麼不用SqlServer,因為寫文章的時候我是mac電腦),這邊我們建立一個項目建立一個訂單按月分表

建立項目

安裝依賴

添加訂單表和訂單表映射

public class Order
    {
        public string Id  { get; set; }
        public string Name  { get; set; }
        public DateTime Createtime  { get; set; }
    }

    public class OrderMap : IEntityTypeConfiguration<Order>
    {
        public void Configure(EntityTypeBuilder<Order> builder)
        {
            builder.HasKey(o => o.Id);
            builder.Property(o => o.Id).HasMaxLength(32).IsUnicode(false);
            builder.Property(o => o.Name).HasMaxLength(255);
            builder.ToTable(nameof(Order));
        }
    }

添加DbContext

public class ShardingDbContext:AbstractShardingDbContext,IShardingTableDbContext
    {
        public ShardingDbContext(DbContextOptions<ShardingDbContext> options) : base(options)
        {
        }

        protected override void OnModelCreating(ModelBuilder modelBuilder)
        {
            base.OnModelCreating(modelBuilder);
            modelBuilder.ApplyConfiguration(new OrderMap());
        }

        public IRouteTail RouteTail { get; set; }
    }

添加訂單分片路由

從5月份開始按建立時間建表

public class OrderRoute:AbstractSimpleShardingMonthKeyDateTimeVirtualTableRoute<Order>
    {
        public override void Configure(EntityMetadataTableBuilder<Order> builder)
        {
            builder.ShardingProperty(o => o.Createtime);
        }

        public override bool AutoCreateTableByTime()
        {
            return true;
        }

        public override DateTime GetBeginTime()
        {
            return new DateTime(2021, 5, 1);
        }
    }

啟動配置

簡單的配置啟動建立表和庫,并且添加種子資料

ILoggerFactory efLogger = LoggerFactory.Create(builder =>
{
    builder.AddFilter((category, level) => category == DbLoggerCategory.Database.Command.Name && level == LogLevel.Information).AddConsole();
});
builder.Services.AddControllers();
builder.Services.AddShardingDbContext<ShardingDbContext>()
    .AddEntityConfig(op =>
    {
        op.CreateShardingTableOnStart = true;
        op.EnsureCreatedWithOutShardingTable = true;
        op.AddShardingTableRoute<OrderRoute>();
        op.UseShardingQuery((conStr, b) =>
        {
            b.UseMySql(conStr, new MySqlServerVersion(new Version())).UseLoggerFactory(efLogger);
        });
        op.UseShardingTransaction((conn, b) =>
        {
            b.UseMySql(conn, new MySqlServerVersion(new Version())).UseLoggerFactory(efLogger);
        });
    }).AddConfig(op =>
    {
        op.ConfigId = "c1";
        op.AddDefaultDataSource("ds0", "server=127.0.0.1;port=3306;database=db2;userid=root;password=root;");
        op.ReplaceTableEnsureManager(sp=>new MySqlTableEnsureManager<ShardingDbContext>());
    }).EnsureConfig();
var app = builder.Build();

app.Services.GetRequiredService<IShardingBootstrapper>().Start();
using (var scope=app.Services.CreateScope())
{
    var shardingDbContext = scope.ServiceProvider.GetRequiredService<ShardingDbContext>();
    if (!shardingDbContext.Set<Order>().Any())
    {
        var begin = new DateTime(2021, 5, 2);
        List<Order> orders = new List<Order>(8);
        for (int i = 0; i < 8; i++)
        {
            orders.Add(new Order()
            {
                Id = i.ToString(),
                Name = $"{begin:yyyy-MM-dd HH:mm:ss}",
                Createtime = begin
            });
            begin = begin.AddMonths(1);
        }
        shardingDbContext.AddRange(orders);
        shardingDbContext.SaveChanges();
    }
}
app.UseAuthorization();
app.MapControllers();
app.Run();

這邊預設連接配接模式的分組是Environment.ProcessorCount

編寫查詢

沒有配置的情況下那麼這個查詢将是十分糟糕

接下來我們将配置Order的查詢

public class OrderQueryConfiguration:IEntityQueryConfiguration<Order>
    {
        public void Configure(EntityQueryBuilder<Order> builder)
        {
            //202105，202106...是預設的順序,false表示使用反向排序,就是如果存在分片那麼分片的tail将進行反向排序202202,202201,202112,202111....
            builder.ShardingTailComparer(Comparer<string>.Default, false);
            //order by createTime asc的順序和分片ShardingTailComparer一樣那麼就用true
            //但是目前ShardingTailComparer是倒序是以order by createTime asc需要和他一樣必須要是倒序,倒序就是false
            builder.AddOrder(o => o.CreateTime,false);
            //配置當不存在Order的時候如果我是FirstOrDefault那麼将采用和ShardingTailComparer相反的排序執行因為是false
            //預設從最早的表開始查詢
            builder.AddDefaultSequenceQueryTrip(false, CircuitBreakerMethodNameEnum.FirstOrDefault);
            ////預設從最近表開始查詢
            //builder.AddDefaultSequenceQueryTrip(true, CircuitBreakerMethodNameEnum.FirstOrDefault);
            //内部配置單表查詢的FirstOrDefault connections limit限制為1
            builder.AddConnectionsLimit(1, LimitMethodNameEnum.FirstOrDefault);
        }
    }

    public class OrderRoute:AbstractSimpleShardingMonthKeyDateTimeVirtualTableRoute<Order>
    {
        //......
        //配置路由才用這個對象查詢
        public override IEntityQueryConfiguration<Order> CreateEntityQueryConfiguration()
        {
            return new OrderQueryConfiguration();
        }
    }

帶配置的Order

現在我們将預設的配置修改回正确

//不合适因為一般而言我們肯定是查詢最新的是以應該和ShardingComparer一樣都是倒序查詢
//builder.AddDefaultSequenceQueryTrip(false, CircuitBreakerMethodNameEnum.FirstOrDefault);
builder.AddDefaultSequenceQueryTrip(true, CircuitBreakerMethodNameEnum.FirstOrDefault);

當然如果你希望本次查詢不使用配置的連接配接數限制可以進行如下操作

_shardingDbContext.Set<Order>().UseConnectionMode(2).Where(o=>o.Id=="7").FirstOrDefaultAsync();

結論:當我們配置了預設分片表應該以何種順序進行分片聚合時,如果相應的查詢方法也進行了配置那麼将這種查詢視為順序查詢,

所有的順序查詢都符合上述表格模式,遇到對應的将直接進行熔斷,不在進行後續的處理直接傳回,保證高性能和防止無意義的查詢。

快速失敗FastFail

顧名思義就是快速失敗,但是很多小夥伴可能不清楚這個快速失敗的意思,失敗就是失敗了為什麼有快速失敗一說,因為ShardingCore内部的本質是将一個sql語句進行才分N條然後并行執行

-- 普通sql

select * from order where id='1' or id='2'

-- 分片sql
select * from order_1 where id='1' or id='2'
select * from order_2 where id='1' or id='2'
-- 分别對這兩個sql進行并行執行

在正常情況下程式是沒有什麼問題的,但是由于程式是并行查詢後疊代聚合是以會帶來一個問題,就是假設執行order_1的線程挂掉了,那麼Task.WhenAll會一緻等待所有線程完成,然後抛出響應的錯誤,

那麼這在很多情況下等于其餘線程都在多無意義的操作,各自管各自。

static async Task Main(string[] args)
        {
            try
            {
                await Task.WhenAll(DoSomething1(), DoSomething2());
                Console.WriteLine("execute success");
            }
            catch 
            {
                Console.WriteLine("error");
            }

            Console.ReadLine();
        }

        static async Task<int> DoSomething1()
        {
            for (int i = 0; i < 10; i++)
            {
                if (i == 2)
                    throw new Exception("111");
                await Task.Delay(1000);
                Console.WriteLine("DoSomething1"+i);
            }

            return 1;
        }
        static async Task<int> DoSomething2()
        {
            for (int i = 0; i < 10; i++)
            {
                await Task.Delay(1000);
                Console.WriteLine("DoSomething2"+i);
            }
            return 1;
        }

代碼很簡單就是

Task.WhenAll

的時候執行兩個委托方法,然後讓其中一個快速抛異常的情況下看看是否馬上傳回

結果是

TaskWhenAll

哪怕出現異常也需要等待所有的線程完成任務,這會在某些情況下浪費不必要的性能,是以這邊

ShardingCore

參考資料采用了

FastFail

版本的

public static Task WhenAllFailFast(params Task[] tasks)
        {
            if (tasks is null || tasks.Length == 0) return Task.CompletedTask;

            // defensive copy.
            var defensive = tasks.Clone() as Task[];

            var tcs = new TaskCompletionSource();
            var remaining = defensive.Length;

            Action<Task> check = t =>
            {
                switch (t.Status)
                {
                    case TaskStatus.Faulted:
                        // we 'try' as some other task may beat us to the punch.
                        tcs.TrySetException(t.Exception.InnerException);
                        break;
                    case TaskStatus.Canceled:
                        // we 'try' as some other task may beat us to the punch.
                        tcs.TrySetCanceled();
                        break;
                    default:

                        // we can safely set here as no other task remains to run.
                        if (Interlocked.Decrement(ref remaining) == 0)
                        {
                            // get the results into an array.
                            tcs.SetResult();
                        }
                        break;
                }
            };

            foreach (var task in defensive)
            {
                task.ContinueWith(check, default, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
            }

            return tcs.Task;
        }

采用failfast後目前主線程會直接在錯誤時傳回,其餘線程還是繼續執行,需要自行進行canceltoken.cancel或者通過共享變量來取消執行

總結

ShardngCore

目前還在不斷努力成長中,也希望各位多多包涵可以在使用中多多提出響應的意見和建議

demo https://github.com/xuejmnet/ShardingCircuitBreaker

參考資料

https://stackoverflow.com/questions/57313252/how-can-i-await-an-array-of-tasks-and-stop-waiting-on-first-exception

下期預告

下一篇我們将講解如何讓流式聚合支援更多的sql查詢,如何将不支援的sql降級為union all

分表分庫元件求贊求star

您的支援是開源作者能堅持下去的最大動力

Github ShardingCore

部落格

QQ群:771630778

個人QQ:326308290(歡迎技術支援提供您寶貴的意見)

個人郵箱:[email protected]

分庫分表如何進行極緻的優化

分庫分表下極緻的優化

題外話

介紹

項目位址

本次優化點

斷路器CircuitBreaker

CircuitBreaker

查詢配置

建立項目

安裝依賴

添加訂單表和訂單表映射

添加DbContext

添加訂單分片路由

啟動配置

編寫查詢

快速失敗FastFail

總結

參考資料

下期預告

分表分庫元件求贊求star

您的支援是開源作者能堅持下去的最大動力

繼續閱讀

配置mongodb分片群集(sharding cluster)

ShardingSphereShardingSphere

ShardingSphere踩坑合集（一）前言ShardingSphere簡介更新分片字段問題between and 解析問題總結

DBContext設定SQLServer連接配接字元串

Sharding-JDBC Demo含源碼

Sharding-jdbc讀寫分離一、binlog日志二、一主一從（讀寫分離）垂直分表實作三、多主多從（讀寫分離）分庫分表實作

springboot整合shardingsphere實作分庫分表實戰

總表與分表互相更新（Excel代碼集團）

Tumblr Architecture - 15 Billion Page Views a Month and Harder to Scale than Twitter

Sharding-JDBC實戰(水準分表、水準分庫、垂直分表、公共表、讀寫分離)Sharding-JDBC實戰

Sharding-Proxy實戰(水準分表、水準分庫、垂直分表、讀寫分離)Sharding-Proxy實戰

Mycat 之垂直拆分（分庫）

Mycat資料庫中間件Mycat

feed流系統重構-架構篇01 應用背景03 分庫分表04 Feed流05 消息隊列寫在最後

【Teams】如何使用sonarqube(docker版)掃描dotnet core版本的Teams項目的代碼品質

Mysql 網站大全