這是一文說通系列的第二篇，裏面有些內容會用到第一篇中間件的部分概念。如果需要，可以參看第一篇：一文說通Dotnet Core的中間件

一、前言

後台任務在一些特殊的應用場合，有相當的需求。

比方，我們需要實現一個定時任務、或周期性的任務、或非API輸出的業務響應、或不允許併發的業務處理，像提現、支付回調等，都需要用到後台任務。

通常，我們在實現後台任務時，有兩種選擇：WebAPI和Console。

下面，我們會用實際的代碼，來理清這兩種工程模式下，後台任務的開發方式。

    為了防止不提供原網址的轉載，特在這裏加上原文鏈接：https://www.cnblogs.com/tiger-wang/p/13081020.html

二、開發環境&基礎工程

這個Demo的開發環境是：Mac + VS Code + Dotnet Core 3.1.2。

$ dotnet --info
.NET Core SDK (reflecting any global.json):
 Version:   3.1.201
 Commit:    b1768b4ae7

Runtime Environment:
 OS Name:     Mac OS X
 OS Version:  10.15
 OS Platform: Darwin
 RID:         osx.10.15-x64
 Base Path:   /usr/local/share/dotnet/sdk/3.1.201/

Host (useful for support):
  Version: 3.1.3
  Commit:  4a9f85e9f8

.NET Core SDKs installed:
  3.1.201 [/usr/local/share/dotnet/sdk]

.NET Core runtimes installed:
  Microsoft.AspNetCore.App 3.1.3 [/usr/local/share/dotnet/shared/Microsoft.AspNetCore.App]
  Microsoft.NETCore.App 3.1.3 [/usr/local/share/dotnet/shared/Microsoft.NETCore.App]

首先，在這個環境下建立工程：

1. 創建Solution

% dotnet new sln -o demo
The template "Solution File" was created successfully.

1. 這次，我們用Webapi創建工程

% cd demo
% dotnet new webapi -o webapidemo
The template "ASP.NET Core Web API" was created successfully.

Processing post-creation actions...
Running 'dotnet restore' on webapidemo/webapidemo.csproj...
  Restore completed in 179.13 ms for demo/demo.csproj.

Restore succeeded.
% dotnet new console -o consoledemo
The template "Console Application" was created successfully.

Processing post-creation actions...
Running 'dotnet restore' on consoledemo/consoledemo.csproj...
  Determining projects to restore...
  Restored consoledemo/consoledemo.csproj (in 143 ms).

Restore succeeded.

1. 把工程加到Solution中

% dotnet sln add webapidemo/webapidemo.csproj
% dotnet sln add consoledemo/consoledemo.csproj

基礎工程搭建完成。

三、在WebAPI下實現一個後台任務

WebAPI下後台任務需要作為託管服務來實現，而託管服務，需要實現IHostedService接口。

首先，我們需要引入一個庫：

% cd webapidemo
% dotnet add package Microsoft.Extensions.Hosting

引入后，我們就有了IHostedService。

下面，我們來做一個IHostedService的派生託管類：

namespace webapidemo
{
    public class DemoService : IHostedService
    {
        public DemoService()
        {
        }

        public Task StartAsync(CancellationToken cancellationToken)
        {
            throw new NotImplementedException();
        }

        public Task StopAsync(CancellationToken cancellationToken)
        {
            throw new NotImplementedException();
        }
    }
}

IHostedService需要實現兩個方法：StartAsync和StopAsync。其中：

StartAsync： 用於啟動後台任務；

StopAsync：主機Host正常關閉時觸發。

如果派生類中有任何非託管資源，那還可以引入IDisposable，並通過實現Dispose來清理非託管資源。

這個類生成后，我們將這個類注入到ConfigureServices中，以使這個類在Startup.Configure調用之前被調用：

public void ConfigureServices(IServiceCollection services)
{
    services.AddControllers();

    services.AddHostedService<DemoService>();
}

下面，我們用一個定時器的後台任務，來加深理解：

namespace webapidemo
{
    public class TimerService : IHostedService, IDisposable
    {
          /* 下面這兩個參數是演示需要，非必須 */
        private readonly ILogger _logger;
        private int executionCount = 0;

          /* 這個是定時器 */
        private Timer _timer;

        public TimerService(ILogger<TimerService> logger)
        {
            _logger = logger;
        }

        public void Dispose()
        {
            _timer?.Dispose();

        }

        private void DoWork(object state)
        {
            var count = Interlocked.Increment(ref executionCount);

            _logger.LogInformation($"Service proccessing {count}");
        }

        public Task StartAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("Service starting");

            _timer = new Timer(DoWork, null, TimeSpan.Zero, TimeSpan.FromSeconds(5));
            return Task.CompletedTask;
        }

        public Task StopAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("Service stopping");

            _timer?.Change(Timeout.Infinite, 0);
            return Task.CompletedTask;
        }
    }
}

注入到ConfigureServices中：

public void ConfigureServices(IServiceCollection services)
{
    services.AddControllers();

    services.AddHostedService<TimerService>();
}

就OK了。代碼比較簡單，就不解釋了。

四、WebAPI後台任務的依賴注入變形

上一節的示例，是一個簡單的形態。

下面，我們按照標準的依賴注入，實現一下這個定時器。

依賴注入的簡單樣式，請參見一文說通Dotnet Core的中間件。

首先，我們創建一個接口IWorkService：

namespace webapidemo
{
    public interface IWorkService
    {
        Task DoWork();
    }
}

再根據IWorkService，建立一個實體類：

namespace webapidemo
{
    public class WorkService : IWorkService
    {
        private readonly ILogger _logger;
        private Timer _timer;
        private int executionCount = 0;

        public WorkService(ILogger<WorkService> logger)
        {
            _logger = logger;
        }

        public async Task DoWork()
        {
            var count = Interlocked.Increment(ref executionCount);

            _logger.LogInformation($"Service proccessing {count}");
        }
    }
}

這樣就建好了依賴的全部內容。

下面，創建託管類：

namespace webapidemo
{
    public class HostedService : IHostedService, IDisposable
    {
        private readonly ILogger<HostedService> _logger;
        public IServiceProvider Services { get; }
        private Timer _timer;

        public HostedService(IServiceProvider services, ILogger<HostedService> logger)
        {
            Services = services;
            _logger = logger;
        }

          public void Dispose()
        {
            _timer?.Dispose();
        }

        private void DoWork(object state)
        {
            _logger.LogInformation("Service working");

            using (var scope = Services.CreateScope())
            {
                var scopedProcessingService =
                    scope.ServiceProvider
                        .GetRequiredService<IWorkService>();

                scopedProcessingService.DoWork().GetAwaiter().GetResult();
            }
        }

        public Task StartAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("Service starting");

            _timer = new Timer(DoWork, null, TimeSpan.Zero, TimeSpan.FromSeconds(5));
            return Task.CompletedTask;
        }


        public Task StopAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("Service stopping");

            _timer?.Change(Timeout.Infinite, 0);
            return Task.CompletedTask;
        }
    }
}

把託管類注入到ConfigureServices中：

public void ConfigureServices(IServiceCollection services)
{
    services.AddControllers();

    services.AddHostedService<HostedService>();
    services.AddSingleton<IWorkService, WorkService>();
}

這樣就完成了。

這種模式下，可以根據注入的內容切換應用的執行內容。不過，這種模式需要注意services.AddSingleton、services.AddScoped和services.AddTransient的區別。

五、Console下的後台任務

Console應用本身就是後台運行，所以區別於WebAPI，它不需要託管運行，也不需要Microsoft.Extensions.Hosting庫。

我們要做的，就是讓程序運行，就OK。

下面是一個簡單的Console模板：

namespace consoledemo
{
    class Program
    {
        private static AutoResetEvent _exitEvent;

        static async Task Main(string[] args)
        {
                /* 確保程序只有一個實例在運行 */
            bool isRuned;
            Mutex mutex = new Mutex(true, "OnlyRunOneInstance", out isRuned);
            if (!isRuned)
                return;

            await DoWork();

                        /* 後台等待 */
            _exitEvent = new AutoResetEvent(false);
            _exitEvent.WaitOne();
        }

        private static async Task DoWork()
        {
            throw new NotImplementedException();
        }
    }
}

這個模板有兩個關鍵的內容：

1. 單實例運行：通常後台任務，只需要有一個實例運行。所以，第一個小段，是解決單實例運行的。多次啟動時，除了第一個實例外，其它的實例會自動退出；
2. 後台等待：看過很多人寫的，在這兒做後台等待時，用了一個無限的循環。類似於下面的：

while(true)
{
    Thread.Sleep(1000);
}

這種方式也沒什麼太大的問題。不過，這段代碼總是要消耗CPU的計算量，雖然很少，但做為後台任務，或者說Service，畢竟是一種消耗，而且看着不夠高大上。

當然如果我們需要中斷，我們也可以把這個模板改成這樣：

namespace consoledemo
{
    class Program
    {
        private static AutoResetEvent _exitEvent;

        static async Task Main(string[] args)
        {
            bool isRuned;
            Mutex mutex = new Mutex(true, "OnlyRunOneInstance", out isRuned);
            if (!isRuned)
                return;

            _exitEvent = new AutoResetEvent(false);
            await DoWork(_exitEvent);
            _exitEvent.WaitOne();
        }

        private static async Task DoWork(AutoResetEvent _exitEvent)
        {
            /* Your Code Here */

            _exitEvent.Set();
        }
    }
}

這樣就可以根據需要，來實現中斷程序並退出。

六、Console應用的其它運行方式

上一節介紹的Console，其實是一個應用程序。

在實際應用中，Console程序跑在Linux服務器上，我們可能會有一些其它的要求：

1. 定時運行

Linux上有一個Service，叫cron，是一個用來定時執行程序的服務。

這個服務的設定，需要另一個命令：crontab，位置在/usr/bin下。

具體命令格式這兒不做解釋，網上隨便查。

1. 運行到後台

命令後邊加個&字符即可：

$ ./command &

1. 運行為Service

需要持續運行的應用，如果以Console的形態存在，則設置為Service是最好的方式。

Linux下，設置一個應用為Service很簡單，就這麼簡單三步：

第一步：在/etc/systemd/system下面，創建一個service文件，例如command.service：

[Unit]
# Service的描述，隨便寫
Description=Command

[Service]
RestartSec=2s
Type=simple
# 執行應用的默認用戶。應用如果沒有特殊要求，最好別用root運行
User=your_user_name
Group=your_group_name
# 應用的目錄，絕對路徑
WorkingDirectory=your_app_folder
# 應用的啟動路徑
ExecStart=your_app_folder/your_app
Restart=always

[Install]
WantedBy=multi-user.target

差不多就這麼個格式。參數的詳細說明可以去網上查，實際除了設置，就是運行了一個腳本。

第二步：把這個command.service加上運行權限：

# chmod +x ./command.service

第三步：註冊為Service：

# systemctl enable command.service

完成。

為了配合應用，還需要記住兩個命令：啟動和關閉Service

# #啟動Service
# systemctl start command.service
# #關閉Service
# systemctl stop command.service

七、寫在後邊的話

今天這個文章，是因為前两天，一個兄弟跑過來問我關於數據總線的實現方式，而想到的一個點。

很多時候，大家在寫代碼的時候，會有一種固有的思想：寫WebAPI，就想在這個框架中把所有的內容都實現了。這其實不算是一個很好的想法。WebAPI，在業務層面，就應該只是實現簡單的處理請求，返回結果的工作，而後台任務跟這個內容截然不同，通常它只做處理，不做返回 — 事實上也不太好返回，要麼客戶端等待時間太長，要麼客戶端已經斷掉了。換句話說，用WebAPI實現總線，絕不是一個好的方式。

不過，Console運行為Service，倒是一個總線應用的絕好方式。如果需要按序執行，可以配合MQ服務器，例如RabbitMQ，來實現消息的按序處理。

再說代碼。很多需求，本來可以用很簡單的方式實現。模式這個東西，用來面試，用來講課，都是很好的內容，但實際開發中，如果有更簡單更有效的方式，用起來！Coding的工作是實現，而不是秀技術。當然，能否找到簡單有效的方式，這個可能跟實際的技術面有關係。但這並不是一個不能跨越的坎。

多看，多想，每天成長一點點！

今天的代碼，在：https://github.com/humornif/Demo-Code/tree/master/0012/demo

（全文完）

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※網頁設計公司推薦不同的風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※南投搬家公司費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

※超省錢租車方案

　　相信絕大多數.NET玩家和我一樣，常常使用Timer這個對象，而在WPF中使用DispatcherTimer的人也是很多，DispatcherTimer是在UI線程跑的。我們的程序中大多數都會充斥很多Timer，可以理解它是一個線程，它繼承自 System.Windows.Threading 。

　　程序中也許會有一些靜態變量或是單例模式的對象來讓不同的頁面進行交互，但也就是這樣讓每個線程之間打架提供了基礎。因為資源是單獨的，就像是腳踩兩隻船的人，必定會翻車。例如一個List集合，你在一個線程中對它進行了操作，在同步瞬間的另線程中，如果不對它謹慎處理，就會造成 “集合已修改；可能無法執行枚舉操作”。當然我們說的不是關於集合的相關問題，而是關於資源分配的，當然在資源搶奪上，是在耗時的線程中才會出現的，例如下面的這張圖。

　　這種耗時的操作，並且在同步線程中，沒有對線程進行封裝，很容易造成資源搶奪問題，假如Object是個集合，我在中間把它改了，下一秒的其它線程對它進行臟讀了，就會產生錯誤，我們可以通過Lock關鍵字。

　　首先在Microsoft文檔中對Lock的說明是，lock 關鍵字可以用來確保代碼塊完成運行，而不會被其他線程中斷。這是通過在代碼塊運行期間為給定對象獲取互斥鎖來實現的。

　　不過我們需要注意的是Lock本質上Monitor.Enter，Monitor.Enter會使值類型裝箱，每次Lock的是裝箱后的對象。Lock其實是類似編譯器的語法糖，因此編譯器直接限制住不能lock值類型，為啥呢，你仔細想想，每次裝箱后都是不同的對象，我怎麼判斷？ object.ReferenceEquals 每次都是false…還有就是千萬不要Lock 字符串，簡單來說Lock字符串之後，只要是你以後有字符串匹配和你Lock里的內容有一樣的，那個該字符串也會被鎖定，相當於死鎖了。

　　Lock和Monitor的區別不是很大，具體看以下代碼。

private static object obj = new object();
        public void LockSomething()
        {
            lock (obj)
            {
                dosomething();
            }
        }
        public void MonitorSomeThing()
        {
                Monitor.Enter(obj);
                dosomething();
                Monitor.Exit(obj);
        }
        public void dosomething()
        { 
            //做具體的事情
        }

 　　lock和Monitor是.NET用一個特殊結構實現的，Monitor對象是完全託管的、完全可移植的，並且在操作系統資源要求方面可能更為有效，同步速度較快，但不能跨進程同步。主要作用是鎖定臨界區，使臨界區代碼只能被獲得鎖的線程執行。Monitor.Wait和Monitor.Pulse用於線程同步，類似信號操作，個人感覺使用比較複雜，容易造成死鎖。

　　lock就是封裝了Monitor.Enter和Monitor.Exit方法其實非常不難理解，只要確定Lock在啥時候用，該怎麼用就可以了，總結一句話。經常會應用於防止多線程操作導致公用變量值出現不確定的異常，用於確保操作的安全性。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※網頁設計公司推薦不同的風格，搶佔消費者視覺第一線

※Google地圖已可更新顯示潭子電動車充電站設置地點!!

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※別再煩惱如何寫文案,掌握八大原則!

※網頁設計最專業,超強功能平台可客製化

redis源碼分析系列文章

[Redis源碼系列]在Liunx安裝和常見API 

為什麼要從Redis源碼分析 

String底層實現——動態字符串SDS 

Redis的雙向鏈表一文全知道

前言

hello，各位小可愛們，又見面了。今天這篇文章來自去年面試閱文的面試題，結果被虐了。這一part不說了，下次專門開一篇，寫下我面試被虐的名場面，尷尬的不行，全程尬聊。哈哈哈哈，話不多說，開始把。

今天要寫Redis的Hash類型，如果有對Redis不熟悉，或者對其他數據類型感興趣的，可以移步上面的系列文章。（最上面的最上面最上面，重要的事情說三遍）

在Redis中Hash類型的應用非常廣泛，其中key到value的映射就通過字典結構來維護的。記筆記，此處要考。

​

API使用

API的使用比較簡單，所以以下就粗略的寫了。

插入數據hset

使用hset命令往myhash中插入兩個key，value的鍵值對，分別是（name,zhangsan）和（age,20），返回值當前的myhash的長度。

獲取數據hget

使用hget命令獲取myhash中key為name的value值。

獲取所有數據hgetall

使用hgetall命令獲取myhash中所有的key和value值。

獲取所有key

使用hkeys命令獲取myhash中所有的key值。

獲取長度

使用hlen命令獲取myhash的長度。

獲取所有value

使用hvals命令獲取myhash中所有的value值。

具體邏輯圖

hash的底層主要是採用字典dict的結構，整體呈現層層封裝。

首先dict有四個部分組成，分別是dictType(類型，不咋重要),dictht（核心），rehashidx(漸進式hash的標誌)，iterators（迭代器），這裏面最重要的就是dictht和rehashidx。

接下來是dictht，其有兩個數組構成，一個是真正的數據存儲位置，還有一個用於hash過程，包括的變量分別是真正的數據table和一些常見變量。

最後數據節點，和上篇說的雙向鏈表一樣，每個節點都有next指針，方便指向下一個節點，這樣目的是為了解決hash碰撞。具體的可以看下圖。

這邊看不懂沒關係，後面會針對每個模塊詳細說明。（千萬不要看到這裏就跳過啦）

​

雙向鏈表的定義

字典結構體dict

我們先看字典結構體dict，其包括四個部分，重點是dictht[2]（真正的數據）和rehashidx（漸進式hash的標誌）。具體圖如下。

具體代碼如下：

//字典結構體 
 typedef struct dict {
    dictType *type;//類型，包括一些自定義函數，這些函數使得key和value能夠存儲 
    void *privdata;//私有數據 
    dictht ht[2];//兩張hash表 
    long rehashidx; //漸進式hash標記，如果為-1，說明沒在進行hash
    unsigned long iterators; //正在迭代的迭代器數量
} dict;

數組結構體dictht

dictht主要包括四個部分，1是真正的數據dictEntry類型的數組，裏面存放的是數據節點；2是數組長度size；3是進行hash運算的參數sizemask，這個不咋重要，只要記住等於size-1；4是數據節點數量used，當前有多少個數據節點。

具體代碼如下：

//hash結構體 
typedef struct dictht {
    dictEntry **table;//真正數據的數組 
    unsigned long size;//數組的大小 
    unsigned long sizemask;//用戶將hash映射到table的位置索引，他的值總是等於size-1 
    unsigned long used;//已用節點數量 
} dictht;

數據節點dictEntry

dictEntry為真正的數據節點，包括key，value和next節點。

//每個節點的結構體  
typedef struct dictEntry {
    void *key; //key
    union {
        void *val;
        uint64_t u64;
        int64_t s64;
        double d;
    } v;//value
    struct dictEntry *next; //下一個數據節點的地址
} dictEntry;

擴容過程和漸進式Hash圖解

我們先來第一個部分，dictht[2]為什麼會要2個數組存放，真正的數據只要一個數組就夠了？

這其實和Java的HashMap相似，都是數據加鏈表的結構，隨着數據量的增加，hash碰撞發生的就越頻繁，每個數組後面的鏈表就越長，整個鏈表顯得非常累贅。如果業務需要大量查詢操作，因為是鏈表，只能從頭部開始查詢，等一個數組的鏈表全部查詢完才能開始下一個數組，這樣查詢時間將無線拉長。

這無疑是要進行擴容，所以第一個數組存放真正的數據，第二個數組用於擴容用。第一個數組中的節點經過hash運算映射到第二個數組上，然後依次進行。那麼過程中還能對外提供服務嗎？答案是可以的，因為他可以隨時停止，這就到了下一個變量rehashidx。（一點都不生硬的轉場，哈哈哈）

rehashidx其實是一個標誌量，如果為-1說明當前沒有擴容，如果不為-1則表示當前擴容到哪個下標位置，方便下次進行從該下標位置繼續擴容。

這樣說是不是太抽象了，還是一臉懵逼，貼心的送上擴容過程全解，一定要點贊評論多誇誇我哦。（越來越不要臉了。。。）

步驟1

首先是未擴容前，rehashidx為-1，表示未擴容，第一個數組的dictEntry長度為4，一共有5個節點，所以used為5。

步驟2

當發生擴容了，rahashidx為第一個數組的第一個下標位置，即0。擴容之後的大小為大於used*2的2的n次方的最小值，即能包含這些節點*2的2的倍數的最小值。因為當前為5個數據節點，所以used*2=10，擴容后的數組大小為大於10的2的次方的最小值，為16。從第一個數組0下標位置開始，查找第一個元素，找到key為name，value為張三的節點，將其hash過，找到在第二個數組的下標為1的位置，將節點移過去，其實是指針的移動。這邊就簡單說了。

步驟3

key為name，value為張三的節點移動結束后，繼續移動第一個數組dictht[0]的下標為0的後續節點，移動步驟和上面相同。

步驟4

繼續移動第一個數組dictht[0]的下標為0的後續節點都移動完了，開始移動下標為1的節點，發現其沒有數據，所以移動下標為2的節點，同時修改rehashidx為2，移動步驟和上面相同。

整個過程的重點在於rehashidx，其為第一個數組正在移動的下標位置，如果當前內存不夠，或者操作系統繁忙，擴容的過程可以隨時停止。

停止之後如果對該對象進行操作，那是什麼樣子的呢？

• 如果是新增，則直接新增后第二個數組，因為如果新增到第一個數組，以後還是要移過來，沒必要浪費時間
• 如果是刪除，更新，查詢，則先查找第一個數組，如果沒找到，則再查詢第二個數組。

字典的實現（源碼分析）

創建並初始化字典

首先分配內存，接着調用初始化方法_dictInit，主要是賦值操作，重點看下rehashidx賦值為-1（這驗證了剛才的圖解，-1表示未進行hash擴容），最後返回是否創建成功。

/* 創建並初始化字典 */
dict *dictCreate(dictType *type,
        void *privDataPtr)
{
    dict *d = zmalloc(sizeof(*d));
    _dictInit(d,type,privDataPtr);
    return d;
}

/* Initialize the hash table */
int _dictInit(dict *d, dictType *type,
        void *privDataPtr)
{
    _dictReset(&d->ht[0]);
    _dictReset(&d->ht[1]);
    d->type = type;
    d->privdata = privDataPtr;
    d->rehashidx = -1;//賦值為-1，表示未進行hash
    d->iterators = 0;
    return DICT_OK;
}

擴容

dict裏面有一個靜態方法_dictExpandIfNeed，判斷是否需要擴容。

首先判斷通過dictIsRehashing方法，判斷是否處於hash狀態，其調用的是宏常量#define dictIsRehashing(d) ((d)->rehashidx != -1)，即判斷rehashidx是否為-1，如果為-1，即不處於hash狀態，if條件為false，可以進行擴容，如果不為-1，即處於hash狀態，if條件為true，不可以進行擴容，直接返回常量DICT_OK。

接着判斷第一個數組的size是否為0，如果為0，則擴容為默認大小4，如果不為0，則執行下面的代碼。

再接着判斷是否需要擴容，if中有三個條件，具體的分析如下。

最後就是調用dictExpand擴容方法了，參數為數據節點的雙倍大小ht[0].used*2。此處驗證了上面擴容過程的數組大小16。

擴容方法比較簡單點，獲取擴容后的大小，將第二個設置新的大小。

這樣講感覺有點空，看下流程圖。

擴容流程圖

具體代碼：

static int _dictExpandIfNeeded(dict *d)
{
    //判斷是否處於擴容狀態中，通過調用宏常量#define dictIsRehashing(d) ((d)->rehashidx != -1)
    //來判斷是否可以擴容
    if (dictIsRehashing(d)) return DICT_OK;

    //判斷第一個數組size是否為0,如果為0，則調用擴容方法，大小為宏常量
    //#define DICT_HT_INITIAL_SIZE     4
    if (d->ht[0].size == 0) return dictExpand(d, DICT_HT_INITIAL_SIZE);

    //下面先列出if條件中所使用到的參數 
    // static int dict_can_resize = 1;數值為1表示可以擴容
    //static unsigned int dict_force_resize_ratio = 5;
    //我們來分析if條件，如果第一個數組的所有節點數量大於等於第一個數組的大小（表示節點數據已經有些多）
    //並且可用擴容（數值為1）或者所有節點數量除以數組大小大於5
    //這個條件表示擴容那個的條件，第一個就是節點必要大於等於數組長度，
    //第二點就再可以擴容和數據太多，超過5兩个中選其一
    if (d->ht[0].used >= d->ht[0].size &&
        (dict_can_resize ||
         d->ht[0].used/d->ht[0].size > dict_force_resize_ratio))
    {
        //調用擴容方法
        return dictExpand(d, d->ht[0].used*2);
    }
    return DICT_OK;
}

int dictExpand(dict *d, unsigned long size)
{
    dictht n;
    //獲取擴容后真正的大小，找到比size大的最小值，且是2的倍數
    unsigned long realsize = _dictNextPower(size);

    //一些判斷條件
    if (dictIsRehashing(d) || d->ht[0].used > size)
        return DICT_ERR;

    if (realsize == d->ht[0].size) return DICT_ERR;

    n.size = realsize;
    n.sizemask = realsize-1;
    n.table = zcalloc(realsize*sizeof(dictEntry*));
    n.used = 0;

    //第一個hash為null，說明在初始化 
    if (d->ht[0].table == NULL) {
        d->ht[0] = n;
        return DICT_OK;
    }
    //正在hash，給第二個hash的長度設置新的， 
    d->ht[1] = n;
    d->rehashidx = 0;//設置當前正在hash 
    return DICT_OK;
}

/* 找到比size大的最小值，且是2的倍數 */
static unsigned long _dictNextPower(unsigned long size)
{
    unsigned long i = DICT_HT_INITIAL_SIZE;

    if (size >= LONG_MAX) return LONG_MAX;
    while(1) {
        if (i >= size)
            return i;
        i *= 2;
    }
}

漸進式hash

漸進式hash過程已經通過上面圖解說明，以下主要看下代碼是如何實現的，以及過程是不是對的。

擴容之後就是執行dictRehash方法，參數包括待移動的哈希表d和步驟数字n。

首先判斷標誌量rehashidx是否等於-1，如果等於-1，則表示hash完成，如果不等於-1，則執行下面的代碼。

接着進行循環，遍歷第一個數組上的每個下標，每次移動下標位置，都需要更新rehashidx值，每次加1。

再接着進行第二個循環，遍歷下標的鏈表每個節點，完成數據的遷移，主要是指針的移動和一些參數的修改。

最後，返回int數值，如果為0表示整個數據全部hash完成，如果返回1則表示部分hash結束，並沒有全部完成，下次可以通過rehashidx值繼續hash。

具體代碼如下：

//重新hash這個哈希表
 // Redis的哈希表結構公有兩個table數組，t0和t1，平常只使用一個t0，當需要重hash時則重hash到另一個table數組中
 //參數列表
 // 1. d: 待移動的哈希表，結構中存有目前已經重hash到哪個桶了
  //  2. n: N步進行rehash 
// 返回值 返回0說明整個表都重hash完成了，返回1代表未完成
int dictRehash(dict *d, int n) {
    int empty_visits = n*10; 
    //如果當前rehashidx=-1,則返回0，表示hash完成 
    if (!dictIsRehashing(d)) return 0;
    //分n步，而且ht[0]還有沒有移動的節點 
    while(n-- && d->ht[0].used != 0) {
        dictEntry *de, *nextde;
        assert(d->ht[0].size > (unsigned long)d->rehashidx);
        //第一個循環用來更新 rehashidx 的值，因為有些桶為空，所以 rehashidx並非每次都比原來前進一個位置，而是有可能前進幾個位置，但最多不超過 10。
        //將rehashidx移動到ht[0]有節點的下標，也就是table[d->rehashidx]非空
        while(d->ht[0].table[d->rehashidx] == NULL) {
            d->rehashidx++;
            if (--empty_visits == 0) return 1;
        }
        de = d->ht[0].table[d->rehashidx];  //第二個循環用來將ht[0]表中每次找到的非空桶中的鏈表（或者就是單個節點）拷貝到ht[1]中

        /* 利用循環講數據節點移過去 */
        while(de) {
            unsigned int h;

            nextde = de->next;
            /* Get the index in the new hash table */
            h = dictHashKey(d, de->key) & d->ht[1].sizemask;
            de->next = d->ht[1].table[h];
            d->ht[1].table[h] = de;
            d->ht[0].used--;
            d->ht[1].used++;
            de = nextde;
        }
        d->ht[0].table[d->rehashidx] = NULL;
        d->rehashidx++;
    }

    if (d->ht[0].used == 0) {
        zfree(d->ht[0].table);
        d->ht[0] = d->ht[1];
        _dictReset(&d->ht[1]);
        d->rehashidx = -1;
        return 0;
    }

    return 1;
}

總結

該篇主要講了Redis的Hash數據類型的底層實現字典結構Dict，先從Hash的一些API使用，引出字典結構Dict，剖析了其三個主要組成部分，字典結構體Dict，數組結構體Dictht，數據節點結構體DictEntry，進而通過多幅過程圖解釋了擴容過程和rehash過程，最後結合源碼對字典進行描述，如創建過程，擴容過程，漸進式hash過程，中間穿插流程圖講解。

如果覺得寫得還行，麻煩給個贊，您的認可才是我寫作的動力！

如果覺得有說的不對的地方，歡迎評論指出。

好了，拜拜咯。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※自行創業缺乏曝光? 網頁設計幫您第一時間規劃公司的形象門面

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※想知道最厲害的網頁設計公司"嚨底家"!

※別再煩惱如何寫文案,掌握八大原則!

※產品缺大量曝光嗎?你需要的是一流包裝設計!

從linux源碼看socket(tcp)的timeout

前言

網絡編程中超時時間是一個重要但又容易被忽略的問題,對其的設置需要仔細斟酌。在經歷了數次物理機宕機之後,筆者詳細的考察了在網絡編程(tcp)中的各種超時設置，於是就有了本篇博文。本文大部分討論的是socket設置為block的情況,即setNonblock(false)，僅在最後提及了nonblock socket(本文基於linux 2.6.32-431內核)。

connectTimeout

在討論connectTimeout之前，讓我們先看下java和C語言對於socket connect調用的函數簽名:

java:
 // 函數調用中攜帶有超時時間
 public void connect(SocketAddress endpoint, int timeout) ;
C語言:
 // 函數調用中並不攜帶超時時間
 int connect(int sockfd, const struct sockaddr * sockaddr, socklen_t socklent) 	 

操作系統提供的connect系統調用並沒有提供timeout的參數設置而java卻有，我們先考察一下原生系統調用的超時策略。

connect系統調用

我們觀察一下此系統調用的kernel源碼，調用棧如下所示:

connect[用戶態]
	|->SYSCALL_DEFINE3(connect)[內核態]
			|->sock->ops->connect

由於我們考察的是tcp的connect,其socket的內部結構如下圖所示:

最終調用的是tcp_connect,代碼如下所示:

int tcp_connect(struct sock *sk) {
	......
	// 發送SYN
	err = tcp_transmit_skb(sk, buff, 1, sk->sk_allocation);
	...
	/* Timer for repeating the SYN until an answer. */
	// 由於是剛建立連接，所以其rto是TCP_TIMEOUT_INIT
	inet_csk_reset_xmit_timer(sk, ICSK_TIME_RETRANS,
				inet_csk(sk)->icsk_rto, TCP_RTO_MAX);
	return 0;	
}

又上面代碼可知，在tcp_connect設置了重傳定時器之後return回了tcp_v4_connect再return到inet_stream_connect。我們繼續考察:

int inet_stream_connect(struct socket *sock, struct sockaddr *uaddr,
			int addr_len, int flags)
{
	......
	// tcp_v4_connect=>tcp_connect
	err = sk->sk_prot->connect(sk, uaddr, addr_len);
	// 這邊用的是sk->sk_sndtimeo
	timeo = sock_sndtimeo(sk, flags & O_NONBLOCK);
	......
	inet_wait_for_connect(sk, timeo));
	......
out:
	release_sock(sk);
	return err;

sock_error:
	err = sock_error(sk) ? : -ECONNABORTED;
	sock->state = SS_UNCONNECTED;
	if (sk->sk_prot->disconnect(sk, flags))
		sock->state = SS_DISCONNECTING;
	goto out
}

由上面代碼可見，可以採用設置SO_SNDTIMEO來控制connect系統調用的超時,如下所示:

setsockopt(sockfd, SOL_SOCKET, SO_SNDTIMEO, &timeout, len);

不設置SO_SNDTIMEO

如果不設置SO_SNDTIMEO,那麼會由tcp重傳定時器在重傳超過設置的時候后超時,如下圖所示:

這個syn重傳的次數由:

cat /proc/sys/net/ipv4/tcp_syn_retries 筆者機器上是5 

來決定。那麼我們就來看一下這個重傳到底是多長時間:

tcp_connect中:
		// 設置的初始超時時間為icsk_rto=TCP_TIMEOUT_INIT為1s
		inet_csk_reset_xmit_timer(sk, ICSK_TIME_RETRANS,
				inet_csk(sk)->icsk_rto, TCP_RTO_MAX);

其重傳定時器的回掉函數為tcp_retransmit_timer:

void tcp_retransmit_timer(struct sock *sk)
{
	......
	// 檢測是否超時
	if (tcp_write_timeout(sk))
		goto out;
	......
	// icsk_rto = icsk_rto * 2,由於syn階段，所以isck_rto不會由於網絡傳輸而改變
	// 重傳的時候會以1,2,4,8指數遞增
	icsk->icsk_rto = min(icsk->icsk_rto << 1, TCP_RTO_MAX);
	// 重設timer
	inet_csk_reset_xmit_timer(sk, ICSK_TIME_RETRANS, icsk->icsk_rto, TCP_RTO_MAX);
out:;		
}

而計算tcp_write_timeout的邏輯則是在這篇blog中已經詳細描述過,

https://my.oschina.net/alchemystar/blog/1936433

只不過在connect時刻，重傳的計算以TCP_TIMEOUT_INIT為單位進行計算。而ESTABLISHED(read/write)時刻，重傳以TCP_RTO_MIN進行計算。那麼根據這段重傳邏輯，我們就可以計算出不同tcp_syn_retries最終表現的超時時間。如下圖所示:

那麼整理下錶格，對於系統調用，connect的超時時間為:

上述超時時間和筆者的實測一致。

kernel代碼版本細微變化

值得注意的是，linux本身官方發布的2.6.32源碼對於tcp_syn_retries2的解釋和RFC並不一致(至少筆者閱讀的代碼如此，這個細微的變化困擾了筆者好久，筆者下載了和機器對應的內核版本后才發現代碼改了)。而redhat發布的2.6.32-431已經修復了這個問題(不清楚具體哪個小版本修改的)，並將初始RTO設置為1s(官方2.6.32為3s)。這也是，不同內核小版本上的實驗會有不同的connect timeout表現的原因(有的抓包到的重傳SYN時間間隔為3,6,12……)。以下為代碼對比:

========================>linux 內核版本2.6.32-431<========================
#define TCP_TIMEOUT_INIT ((unsigned)(1*HZ))	/* RFC2988bis initial RTO value	*/

static inline bool retransmits_timed_out(struct sock *sk,
					 unsigned int boundary,
					 unsigned int timeout,
					 bool syn_set)
{
	......
	unsigned int rto_base = syn_set ? TCP_TIMEOUT_INIT : TCP_RTO_MIN;
	......
	timeout = ((2 << boundary) - 1) * rto_base;
	......

}
========================>linux 內核版本2.6.32.63<========================
#define TCP_TIMEOUT_INIT ((unsigned)(3*HZ))	/* RFC 1122 initial RTO value	*/

static inline bool retransmits_timed_out(struct sock *sk,
					 unsigned int boundary
{
	......
	timeout = ((2 << boundary) - 1) * TCP_RTO_MIN;
	......
}

另外，tcp_syn_retries重傳次數可以在單個socket中通過setsockopt設置。

JAVA connect API

現在我們考察下java的connect api,其connect最終調用下面的代碼:

Java_java_net_PlainSocketImpl_socketConnect(...){

    if (timeout <= 0) {
    	 ......
        connect_rv = NET_Connect(fd, (struct sockaddr *)&him, len);
    	 .....
    }else{
    	 // 如果timeout > 0 ，則設置為nonblock模式
        SET_NONBLOCKING(fd);
        /* no need to use NET_Connect as non-blocking */
        connect_rv = connect(fd, (struct sockaddr *)&him, len);
        /*
         * 這邊用系統調用select來模擬阻塞調用超時
         */
        while (1) {
            ......
            struct timeval t;
            t.tv_sec = timeout / 1000;
            t.tv_usec = (timeout % 1000) * 1000;
            connect_rv = NET_Select(fd+1, 0, &wr, &ex, &t);
            ......
        }
        ......
        // 重新設置為阻塞模式
        SET_BLOCKING(fd);
        ......
    }
}

其和connect系統調用的不同點是，在timeout為0的時候，走默認的系統調用不設置超時時間的邏輯。在timeout>0時，將socket設置為非阻塞，然後用select系統調用去模擬超時,而沒有走linux本身的超時邏輯，如下圖所示:

由於沒有java並沒有設置so_sndtimeo的選項，所以在timeout為0的時候，直接就通過重傳次數來控制超時時間。而在調用connect時設置了timeout(不為0)的時候,超時時間如下錶格所示:

socketTimeout

write系統調用的超時時間

socket的write系統調用最後調用的是tcp_sendmsg,源碼如下所示:

int tcp_sendmsg(struct kiocb *iocb, struct socket *sock, struct msghdr *msg,
		size_t size){
	......
	timeo = sock_sndtimeo(sk, flags & MSG_DONTWAIT);
	......
	while (--iovlen >= 0) {
		......
		// 此種情況是buffer不夠了
		if (copy <= 0) {
	new_segment:
		  ......
		  if (!sk_stream_memory_free(sk))
			  goto wait_for_sndbuf;

		  skb = sk_stream_alloc_skb(sk, select_size(sk),sk->sk_allocation);
		  if (!skb)
			  goto wait_for_memory;
		}
		......
	}
	......
	// 這邊等待write buffer有空間
wait_for_sndbuf:
		set_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
wait_for_memory:
		if (copied)
			tcp_push(sk, flags & ~MSG_MORE, mss_now, TCP_NAGLE_PUSH);
			// 這邊等待timeo長的時間
		if ((err = sk_stream_wait_memory(sk, &timeo)) != 0)
			goto do_error;
		......
out:
	// 如果拷貝了數據，則返回
	if (copied)
		tcp_push(sk, flags, mss_now, tp->nonagle);
	TCP_CHECK_TIMER(sk);
	release_sock(sk);
	return copied;		
out_err:
	// error的處理
	err = sk_stream_error(sk, flags, err);
	TCP_CHECK_TIMER(sk);
	release_sock(sk);
	return err;		
}

從上面的內核代碼看出，如果socket的write buffer依舊有空間的時候，會立馬返回，並不會有timeout。但是write buffer不夠的時候，會等待SO_SNDTIMEO的時間(nonblock時候為0)。但是如果SO_SNDTIMEO沒有設置的時候,默認初始化為MAX_SCHEDULE_TIMEOUT,可以認為其超時時間為無限。那麼其超時時間會有另一個條件來決定，我們看下sk_stream_wait_memory的源碼:

int sk_stream_wait_memory(struct sock *sk, long *timeo_p){
		// 等待socket shutdown或者socket出現err
		sk_wait_event(sk, &current_timeo, sk->sk_err ||
						  (sk->sk_shutdown & SEND_SHUTDOWN) ||
						  (sk_stream_memory_free(sk) &&
						  !vm_wait));
}						 

在write等待的時候，如果出現socket被shutdown或者socket出現錯誤的時候，則會跳出wait進而返回錯誤。在不考慮對端shutdown的情況下,出現sk_err的時間其實就是其write的timeout時間,那麼我們看下什麼時候出現sk->sk_err。

SO_SNDTIMEO不設置,write buffer滿之後ack一直不返回的情況(例如，物理機宕機)

物理機宕機后，tcp發送msg的時候,ack不會返回，則會在重傳定時器tcp_retransmit_timer到期后timeout,其重傳到期時間通過tcp_retries2以及TCP_RTO_MIN計算出來。其源碼可見筆者的blog:

https://my.oschina.net/alchemystar/blog/1936433

tcp_retries2的設置位置為:

cat /proc/sys/net/ipv4/tcp_retries2 筆者機器上是5,默認是15

SO_SNDTIMEO不設置,write buffer滿之後對端不消費，導致buffer一直滿的情況

和上面ack超時有些許不一樣的是，一個邏輯是用TCP_RTO_MIN通過tcp_retries2計算出來的時間。另一個是真的通過重傳超過tcp_retries2次數來time_out，兩者的區別和rto的動態計算有關。但是可以大致認為是一致的。

上述邏輯如下圖所示:

write_timeout表格

java的SocketOutputStream的sockWrite0超時時間

java的sockWrite0沒有設置超時時間的地方，同時也沒有設置過SO_SNDTIMEOUT，其直接調用了系統調用,所以其超時時間和write系統調用保持一致。

readTimeout

ReadTimeout可能是最容易導致問題的地方。我們先看下系統調用的源碼:

read系統調用

socket的read系統調用最終調用的是tcp_recvmsg, 其源碼如下:

int tcp_recvmsg(struct kiocb *iocb, struct sock *sk, struct msghdr *msg,
		size_t len, int nonblock, int flags, int *addr_len)
{
	......
	// 這邊timeo=SO_RCVTIMEO
	timeo = sock_rcvtimeo(sk, nonblock);
	......
	do{
		......
		// 下面這一堆判斷表明，如果出現錯誤,或者已經被CLOSE/SHUTDOWN則跳出循環
		if(copied) {
			if (sk->sk_err ||
			    sk->sk_state == TCP_CLOSE ||
			    (sk->sk_shutdown & RCV_SHUTDOWN) ||
			    !timeo ||
			    signal_pending(current))
				break;
		} else {
			if (sock_flag(sk, SOCK_DONE))
				break;

			if (sk->sk_err) {
				copied = sock_error(sk);
				break;
			}
			// 如果socket shudown跳出
			if (sk->sk_shutdown & RCV_SHUTDOWN)
				break;
			// 如果socket close跳出
			if (sk->sk_state == TCP_CLOSE) {
				if (!sock_flag(sk, SOCK_DONE)) {
					/* This occurs when user tries to read
					 * from never connected socket.
					 */
					copied = -ENOTCONN;
					break;
				}
				break;
			}
			.......
		}
		.......

		if (copied >= target) {
			/* Do not sleep, just process backlog. */
			release_sock(sk);
			lock_sock(sk);
		} else /* 如果沒有讀到target自己數(和水位有關,可以暫認為是1)，則等待SO_RCVTIMEO的時間 */
			sk_wait_data(sk, &timeo);	
	} while (len > 0);
	......
}

上面的邏輯如下圖所示:

重傳以及探測定時器timeout事件的觸發時機如下圖所示:

如果內核層面ack正常返回而且對端窗口不為0，僅僅應用層不返回任何數據,那麼就會無限等待，直到對端有數據或者socket close/shutdown為止，如下圖所示:

很多應用就是基於這個無限超時來設計的,例如activemq的消費者邏輯。

java的SocketInputStream的sockRead0超時時間

java的超時時間由SO_TIMOUT決定，而linux的socket並沒有這個選項。其sockRead0和上面的java connect一樣，在SO_TIMEOUT>0的時候依舊是由nonblock socket模擬,在此就不再贅述了。

ReadTimeout超時表格

C系統調用:

Java系統調用

對端物理機宕機之後的timeout

對端物理機宕機后還依舊有數據發送

對端物理機宕機時對端內核也gg了(不會發出任何包通知宕機)，那麼本端發送任何數據給對端都不會有響應。其超時時間就由上面討論的
min(設置的socket超時[例如SO_TIMEOUT],內核內部的定時器超時來決定)。

對端物理機宕機后沒有數據發送，但在read等待

這時候如果設置了超時時間timeout，則在timeout后返回。但是，如果僅僅是在read等待，由於底層沒有數據交互，那麼其無法知道對端是否宕機，所以會一直等待。但是，內核會在一個socket兩個小時都沒有數據交互情況下(可設置)啟動keepalive定時器來探測對端的socket。如下圖所示:

大概是2小時11分鐘之後會超時返回。keepalive的設置由內核參數指定：

cat /proc/sys/net/ipv4/tcp_keepalive_time 7200 即兩個小時后開始探測
cat /proc/sys/net/ipv4/tcp_keepalive_intvl 75 即每次探測間隔為75s
cat /proc/sys/net/ipv4/tcp_keepalve_probes 9 即一共探測9次

可以在setsockops中對單獨的socket指定是否啟用keepalive定時器(java也可以)。

對端物理機宕機后沒有數據發送，也沒有read等待

和上面同理，也是在keepalive定時器超時之後，將連接close。所以我們可以看到一個不活躍的socket在對端物理機突然宕機之後,依舊是ESTABLISHED狀態，過很長一段時間之後才會關閉。

進程宕后的超時

如果僅僅是對端進程宕機的話(進程所在內核會close其所擁有的所有socket)，由於fin包的發送，本端內核可以立刻知道當前socket的狀態。如果socket是阻塞的，那麼將會在當前或者下一次write/read系統調用的時候返回給應用層相應的錯誤。如果是nonblock，那麼會在select/epoll中觸發出對應的事件通知應用層去處理。
如果fin包沒發送到對端，那麼在下一次write/read的時候內核會發送reset包作為回應。

nonblock

設置為nonblock=true后，由於read/write都是立刻返回，且通過select/epoll等處理重傳超時/probe超時/keep alive超時/socket close等事件，所以根據應用層代碼決定其超時特性。定時器超時事件發生的時間如上面幾小節所述，和是否nonblock無關。nonblock的編程模式可以讓應用層對這些事件做出響應。

總結

網絡編程中超時時間是個重要但又容易被忽略的問題，這個問題只有在遇到物理機宕機等平時遇不到的現象時候才會凸顯。筆者在經曆數次物理機宕機之後才好好的研究了一番，希望本篇文章可以對讀者在以後遇到類似超時問題時有所幫助。

公眾號

關注筆者公眾號，獲取更多乾貨文章:

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※別再煩惱如何寫文案,掌握八大原則!

※教你寫出一流的銷售文案?

※超省錢租車方案

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※產品缺大量曝光嗎?你需要的是一流包裝設計!

前幾天同事不小心誤操作，將SQLServer庫的一張表的一個狀態字段給刷成了一個統一狀態，由於是update執行所以原來的相關狀態無法確定。發生這種事情的時候我的小夥伴背後 一涼。

由於是在開發試運行中的項目，還沒來得及進行備份處理，所以從備份恢復宣告失敗。就算有備份那麼恢復的也是備份時間節點的數據，意味着使用平台做的數據需要從備份時間重新做過，而且有可能有遺漏。

小夥伴問我這咋辦，首先沒有備份，那麼只有從數據庫日誌查找，然後看能不能通過日誌找回之前的數據，再還原到刷狀態之前的數據。然後就找到了ApexSQLLog工具，接下來我介紹下這款工具的使用和如何恢複數據。ApexSQLLog有幾個版本，我是用的是ApexSQLLog2014支持SqlServer更高的版本，數據庫使用的是SqlSerVer2014。

ApexSQLLog2014
提取碼: np4f

• 首先建一個測試庫，和一張測試表。
  測試庫ApexSQLLogTest和測試用的表TestUser，然後我手動編輯了三條數據進去，保存編輯的數據。

• 用ApexSQLLog打開測試庫日誌
  選擇要連接的數據庫，也可以從最近的session中打開，打開篩選過的記錄可以保存未session。
  然後選中要篩選的日誌文件，如果有備份數據庫文件也會自動查找到並在這裏羅列出來，自己按情況選擇。

• 條件篩選
  我們選擇日誌文件后就進入到篩選條件選擇，可以在篩選條件裏面自由組合。
  可以選擇時間段（Time range）、操作（operations）、表（tables）。

  高級選項（advanced options）裏面還有用戶、字段條件等可以選擇。

• 查看日誌數據
  當我們組合完篩選條件后，就進入到日誌分析界面，可以看到我們之前手動插入的三條數據實際已經在日誌裏面了，分成了三條insert語句。在選中其中一條日誌的時候在下面可以看到執行的各字段值的修改情況。

  可以點擊 下面的Row history查看記錄，Redo script可以生成執行的操作， Undo script可以還原到之前的數據。我們恢複數據就是使用Undo script。

• 恢複數據測試。
  我們使用update語句將Status狀態全都重置為3。

update TestUser set Status=3 

然後刷新下日誌，會看到多出了三條Update日誌記錄，點擊第一條看到下面的Status字段從0變為了3。

我們選中這三條記錄右鍵或者上面的菜單欄功能，用create undo script 生成恢復sql。

--	This UNDO script was generated with ApexSQL Log 2014.04.1133 on 2020-06-10 11:18:47.601
--	NOTE: Operations in UNDO scripts are always output in descending order.
--	SERVER VIP-966\SQLEXPRESS
--	DATABASE ApexSQLLogTest

--	UPDATE (00000024:000000A0:0004) done at 2020-06-10 11:09:36.293 by VIP-966\Administrator in transaction 0000:0000034B (Committed)
BEGIN TRANSACTION 
UPDATE [dbo].[TestUser] SET [Status] = 2 WHERE [Id] = 3
IF @@ROWCOUNT <= 1 COMMIT TRANSACTION ELSE BEGIN ROLLBACK TRANSACTION; PRINT 'ERROR: STATEMENT AFFECTED MORE THAN ONE ROW. ALL THE CHANGES WERE ROLLED BACK.' END
--	UPDATE (00000024:000000A0:0003) done at 2020-06-10 11:09:36.293 by VIP-966\Administrator in transaction 0000:0000034B (Committed)
BEGIN TRANSACTION 
UPDATE [dbo].[TestUser] SET [Status] = 1 WHERE [Id] = 2
IF @@ROWCOUNT <= 1 COMMIT TRANSACTION ELSE BEGIN ROLLBACK TRANSACTION; PRINT 'ERROR: STATEMENT AFFECTED MORE THAN ONE ROW. ALL THE CHANGES WERE ROLLED BACK.' END
--	UPDATE (00000024:000000A0:0002) done at 2020-06-10 11:09:36.293 by VIP-966\Administrator in transaction 0000:0000034B (Committed)
BEGIN TRANSACTION 
UPDATE [dbo].[TestUser] SET [Status] = 0 WHERE [Id] = 1
IF @@ROWCOUNT <= 1 COMMIT TRANSACTION ELSE BEGIN ROLLBACK TRANSACTION; PRINT 'ERROR: STATEMENT AFFECTED MORE THAN ONE ROW. ALL THE CHANGES WERE ROLLED BACK.' END
GO

--	FINISHED ON 2020-06-10 11:18:47.697
--	TOTAL OPERATIONS PROCESSED 3
--	END OF FILE

最後我們就可以使用這個腳本去恢複數據了。

注意

我們在使用日誌恢復的時候如果表有主鍵會根據主鍵生成sql，如上圖sql中 where後面的條件。如果表沒有主鍵那麼生成的sql後面的where條件會帶上所有的字段。在我幫小夥伴恢複數據的時候發現他的表沒有設置主鍵，而且字段有20多個，3萬多條數據生成的sql都是100多M，還要拆分執行。
比如我把Id主鍵去了再更新下Status狀態到4，生成的sql如下，會提示沒有主鍵。

以上就是一次數據恢復的分享，如果下次你也遇到這種情況希望能幫到你。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※教你寫出一流的銷售文案?

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※回頭車貨運收費標準

※別再煩惱如何寫文案,掌握八大原則!

※超省錢租車方案

※產品缺大量曝光嗎?你需要的是一流包裝設計!