在Asp.NET Core中如何優雅的管理用戶機密數據

背景

回顧

在軟件開發過程中，使用配置文件來管理某些對應用程序運行中需要使用的參數是常見的作法。在早期VB/VB.NET時代，經常使用.ini文件來進行配置管理；而在.NET FX開發中，我們則傾向於使用web.config文件，通過配置appsetting的配置節來處理；而在.NET Core開發中，我們有了新的基於json格式的appsetting.json文件。

無論採用哪種方式，其實配置管理從來都是一件看起來簡單，但影響非常深遠的基礎性工作。尤其是配置的安全性，貫穿應用程序的始終，如果沒能做好安全性問題，極有可能會給系統帶來不可控的風向。

源代碼比配置文件安全么？

有人以為把配置存放在源代碼中，可能比存放在明文的配置文件中似乎更安全，其實是“皇帝的新裝”。

在前不久，筆者的一位朋友就跟我說了一段故事：他說一位同事在離職后，直接將曾經寫過的一段代碼上傳到github的公共倉庫，而這段代碼中包含了某些涉及到原企業的機密數據，還好被github的安全機制提前發現而及時終止了該行為，否則後果不堪設想。

於是，筆者順手查了一下由於有意或無意泄露企業機密，造成企業損失的案例，發現還真不少。例如大疆前員工通過 Github 泄露公司源代碼，被罰 20 萬、獲刑半年 這起案件，也是一個典型的案例。

該員工離職后，將包含關鍵配置信息的源代碼上傳到github的公共倉庫，被黑客利用，使得大量用戶私人數據被黑客獲取，該前員工最終被刑拘。

圖片來源： http://www.digitalmunition.com/WhyIWalkedFrom3k.pdf

大部分IT公司都會在入職前進行背景調查，而一旦有案底，可能就已經與許多IT公司無緣；即便是成為創業者，也可能面臨無法跟很多正規企業合作的問題。

小結

所以，安全性問題不容小覷，哪怕時間再忙，也不要急匆匆的就將數據庫連接字符串或其他包含敏感信息的內容輕易的記錄在源代碼或配置文件中。在這個點上，一旦出現問題，往往都是非常嚴重的問題。

如何實現

在.NET FX時代，我們可以使用對web.config文件的關鍵配置節進行加密的方式，來保護我們的敏感信息，在.NET Core中，自然也有這些東西，接下來我將簡述在開發環境和生產環境下不同的配置加密手段，希望能夠給讀者帶來啟迪。

開發環境

在開發環境下，我們可以使用visual studio 工具提供的用戶機密管理器，只需0行代碼，即可輕鬆完成關鍵配置節的處理。

機密管理器概述

根據微軟官方文檔 的描述：

ASP.NET Core 機密管理器工具提供了開發過程中在源代碼外部保存機密的另一種方法 。 若要使用機密管理器工具，請在項目文件中安裝包 Microsoft.Extensions.Configuration.SecretManager 。 如果該依賴項存在並且已還原，則可以使用 dotnet user-secrets 命令來通過命令行設置機密的值。 這些機密將存儲在用戶配置文件目錄中的 JSON 文件中（詳細信息隨操作系統而異），與源代碼無關。

機密管理器工具設置的機密是由使用機密的項目的 UserSecretsId 屬性組織的。 因此，必須確保在項目文件中設置 UserSecretsId 屬性，如下面的代碼片段所示。 默認值是 Visual Studio 分配的 GUID，但實際字符串並不重要，只要它在計算機中是唯一的。

<PropertyGroup>
   <UserSecretsId>UniqueIdentifyingString</UserSecretsId>
</PropertyGroup> 

Secret Manager工具允許開發人員在開發ASP.NET Core應用程序期間存儲和檢索敏感數據。敏感數據存儲在與應用程序源代碼不同的位置。由於Secret Manager將秘密與源代碼分開存儲，因此敏感數據不會提交到源代碼存儲庫。但機密管理器不會對存儲的敏感數據進行加密，因此不應將其視為可信存儲。敏感數據作為鍵值對存儲在JSON文件中。最好不要在開發和測試環境中使用生產機密。查看引文。

存放位置

在windows平台下，機密數據的存放位置為：

%APPDATA%\Microsoft\UserSecrets\\secrets.json

而在Linux/MacOs平台下，機密數據的存放位置為：

 ~/.microsoft/usersecrets/<user_secrets_id>/secrets.json 

在前面的文件路徑中， “將替換UserSecretsId為 .csproj文件中指定的值。

在Windows環境下使用機密管理器

在windows下，如果使用Visual Studio2019作為主力開發環境，只需在項目右鍵單擊，選擇菜單【管理用戶機密】，即可添加用戶機密數據。

在管理用戶機密數據中，添加的配置信息和傳統的配置信息沒有任何區別。

{
“ConnectionStrings”: {
“Default”: “Server=xxx;Database=xxx;User ID=xxx;Password=xxx;”
}
}

我們同樣也可以使用IConfiguration的方式、IOptions 的方式，進行配置的訪問。

在非Windows/非Visual Studio環境下使用機密管理器

完成安裝dotnet-cli后，在控制台輸入

dotnet user-secrets init 

前面的命令將在UserSecretsId .csproj 文件的PropertyGroup中添加 .csproj一個元素。 UserSecretsId是對項目是唯一的Guid值。

 <PropertyGroup>  
 	<TargetFramework>netcoreapp3.1</TargetFramework>
    <UserSecretsId>79a3edd0-2092-40a2-a04d-dcb46d5ca9ed</UserSecretsId> 
 </PropertyGroup> 

設置機密

 dotnet user-secrets set "Movies:ServiceApiKey" "12345" 

列出機密

 dotnet user-secrets list 

刪除機密

 dotnet user-secrets remove "Movies:ConnectionString" 

清除所有機密

 dotnet user-secrets clear 

生產環境

機密管理器為開發者在開發環境下提供了一種保留機密數據的方法，但在開發環境下是不建議使用的，如果想在生產環境下，對機密數據進行保存該怎麼辦？

按照微軟官方文檔的說法，推薦使用Azure Key Vault 來保護機密數據，但。。我不是貴雲的用戶（當然，買不起貴雲不是貴雲太貴，而是我個人的問題[手動狗頭]）。

其次，與Azure Key Valut類似的套件，例如其他雲，差不多都有，所以都可以為我們所用。

但。。如果您如果跟我一樣，不想通過第三方依賴的形式來解決這個問題，那不如就用最簡單的辦法，例如AES加密。

使用AES加密配置節

該方法與平時使用AES對字符串進行加密和解密的方法並無區別，此處從略。

使用數據保護Api（DataProtect Api實現）

在平時開發過程中，能夠動手擼AES加密是一種非常好的習慣，而微軟官方提供的數據保護API則將這個過程進一步簡化，只需調Api即可完成相應的數據加密操作。

關於數據保護api， Savorboard 大佬曾經寫過3篇博客討論這個技術問題，大家可以參考下面的文章來獲取信息。

ASP.NET Core 數據保護（Data Protection 集群場景）【上】

ASP.NET Core 數據保護（Data Protection 集群場景）【中】

ASP.NET Core 數據保護（Data Protection 集群場景）【下】

(接下來我要貼代碼了，如果沒興趣，請出門左拐，代碼不能完整運行，查看代碼）

首先，注入配置項

 public static IServiceCollection AddProtectedConfiguration(this IServiceCollection services, string directory)
        {
            services
                .AddDataProtection()
                .PersistKeysToFileSystem(new DirectoryInfo(directory))
                .UseCustomCryptographicAlgorithms(new ManagedAuthenticatedEncryptorConfiguration
                {
                    EncryptionAlgorithmType = typeof(Aes),
                    EncryptionAlgorithmKeySize = 256,
                    ValidationAlgorithmType = typeof(HMACSHA256)
                });
            ;

            return services;
        }

其次，實現對配置節的加/解密。（使用AES算法的數據保護機制）

public class ProtectedConfigurationSection : IConfigurationSection
    {
        private readonly IDataProtectionProvider _dataProtectionProvider;
        private readonly IConfigurationSection _section;
        private readonly Lazy<IDataProtector> _protector;

        public ProtectedConfigurationSection(
            IDataProtectionProvider dataProtectionProvider,
            IConfigurationSection section)
        {
            _dataProtectionProvider = dataProtectionProvider;
            _section = section;

            _protector = new Lazy<IDataProtector>(() => dataProtectionProvider.CreateProtector(section.Path));
        }

        public IConfigurationSection GetSection(string key)
        {
            return new ProtectedConfigurationSection(_dataProtectionProvider, _section.GetSection(key));
        }

        public IEnumerable<IConfigurationSection> GetChildren()
        {
            return _section.GetChildren()
                .Select(x => new ProtectedConfigurationSection(_dataProtectionProvider, x));
        }

        public IChangeToken GetReloadToken()
        {
            return _section.GetReloadToken();
        }

        public string this[string key]
        {
            get => GetProtectedValue(_section[key]);
            set => _section[key] = _protector.Value.Protect(value);
        }

        public string Key => _section.Key;
        public string Path => _section.Path;

        public string Value
        {
            get => GetProtectedValue(_section.Value);
            set => _section.Value = _protector.Value.Protect(value);
        }

        private string GetProtectedValue(string value)
        {
            if (value == null)
                return null;

            return _protector.Value.Unprotect(value);
        }
    }

再次，在使用前，先將待加密的字符串轉換成BASE64純文本，然後再使用數據保護API對數據進行處理，得到處理后的字符串。

private readonly IDataProtectionProvider _dataProtectorTokenProvider;
public TokenAuthController( IDataProtectionProvider dataProtectorTokenProvider)
{
}
[Route("encrypt"), HttpGet, HttpPost]
public string Encrypt(string section, string value)
{
     var protector = _dataProtectorTokenProvider.CreateProtector(section);
     return protector.Protect(value);
}

再替換配置文件中的對應內容。

{
  "ConnectionStrings": {
    "Default": "此處是加密后的字符串"
  }
}

然後我們就可以按照平時獲取IOptions 的方式來獲取了。

問題

公眾號【DotNET騷操作】號主【周傑】同學提出以下觀點：

1、在生產環境下，使用AES加密，其實依然是一種不夠安全的行為，充其量也就能忽悠下產品經理，畢竟幾條簡單的語句，就能把機密數據dump出來。

也許在這種情況下，我們應該優先考慮accessKeyId/accessSecret，盡量通過設置多級子賬號，通過授權Api的機制來管理機密數據，而不是直接暴露類似於數據庫連接字符串這樣的關鍵配置信息。另外，應該定期更換數據庫的密碼，盡量將類似的問題可能造成的風險降到最低。數據保護api也提供的類似的機制，使得開發者能夠輕鬆的管理機密數據的時效性問題。

2、配置文件放到CI/CD中，發布的時候在CI/CD中進行組裝，然後運維只是負責管理CI/CD的賬戶信息，而最高機密數據，則由其他人負責配置。

嗯，我完全同意他的第二種做法，另外考慮到由於運維同樣有可能會有意無意泄露機密數據，所以如果再給運維配備一本《刑法》，並讓他日常補習【侵犯商業秘密罪】相關條款，這個流程就更加閉環了。

結語

本文簡述了在.NET Core中，如何在開發環境下使用用戶機密管理器、在生產環境下使用AES+IDataProvider的方式來保護我們的用戶敏感數據。由於時間倉促，如有考慮不周之處，還請各位大佬批評指正。

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

【其他文章推薦】

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

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

※超省錢租車方案

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

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

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

我們處於的位置

我們要清楚現在的情況.

現在我們已經調用了SqlSessionFactoryBuilder的build方法生成了SqlSessionFactory 對象.

但是如標題所說,要想生成sqlsession還要另一步SqlSessionFactory 調用openSession()方法生成sqlsession;

這就要從上一部分代碼講起

上文講到

我們創建的實際上是一個叫做DefaultSqlSessionFactory的類,實際上他是一個SqlSessionFactory接口(沒錯,這玩應是接口)的實現類.

既然sqlsession是由opensession產生的,那我們就先看這個方法.

說一嘴題外話就是自動提交也是在這個部分設置的,下面是如果你設置了autocommit的情況.

public SqlSession openSession(boolean autoCommit) {
  //this.configuration.getDefaultExecutorType()值為 ExecutorType.SIMPLE;
    return this.openSessionFromDataSource(this.configuration.getDefaultExecutorType(), (TransactionIsolationLevel)null, autoCommit);
}

參數中 configuration 獲取了默認的執行器 “SIMPLE”.

DefaultSqlSessionFactory類

調用了一個同一個類中openSessionFromDataSource方法.

在這個類中是如下執行流程

所要知道的一部分知識.

environments運行環境

MyBatis 核心配置綜述之 Configuration詳解

其實就是數據庫連接那個部分.

private SqlSession openSessionFromDataSource(ExecutorType execType, TransactionIsolationLevel level, boolean autoCommit) {
  Transaction tx = null;
  try {
    //從configuration對象中得到環境配置的對象
    final Environment environment = configuration.getEnvironment();
    //這個對象被用來創建一個事務工廠->一號分支
    final TransactionFactory transactionFactory = getTransactionFactoryFromEnvironment(environment);
  //事務工廠創建一個事務對象->二號分支
    tx = transactionFactory.newTransaction(environment.getDataSource(), level, autoCommit);
    //而 configurationye 則會根據事務對象和執行器類型創建一個執行器。
    ->三號分支
    final Executor executor = configuration.newExecutor(tx, execType);
    //返回一個默認的DefaultSqlSession對象
    ->四號分支
    return new DefaultSqlSession(configuration, executor, autoCommit);
  } catch (Exception e) {
    closeTransaction(tx); // may have fetched a connection so lets call close()
    throw ExceptionFactory.wrapException("Error opening session.  Cause: " + e, e);
  } finally {
    ErrorContext.instance().reset();
  }
}

現在我們要從一號分支開始

一號分支

final TransactionFactory transactionFactory = getTransactionFactoryFromEnvironment(environment);

這個代碼如下:

我們發現有兩種可能性.

如果傳進來的值沒有設置 標籤那麼他會執行 ManagedTransactionFactory()而反之則會執行 environment.getTransactionFactory()

這兩者產生的對象都實現了 TransactionFactory接口.

這裏ManagedTransactionFactory()是沒有標籤時生成的對象.其核心就是一句

private boolean closeConnection = true;的屬性.

我們不必過於關注這個部分.

private TransactionFactory getTransactionFactoryFromEnvironment(Environment environment) {
  if (environment == null || environment.getTransactionFactory() == null) {
   //如果沒有目標標籤
    return new ManagedTransactionFactory();
  }
  //如果有目標標籤
  return environment.getTransactionFactory();
}

environment.getTransactionFactory()產生的東西才是重點.

調用環境對象的getTransactionFactory方法，該方法和我們配置的一樣返回了一個 JdbcTransactionFactory，而實際上，TransactionFactory 只有2個實現類，一個是 ManagedTransactionFactory (沒有標籤時返回的)，一個是 JdbcTransactionFactory(有標籤時返回的)。

至此一號分支結束,從此看來,一號分支實際上是將environment對象包裝成一個工廠對象.

請返回一號分支之前部分繼續.

分支二

tx = transactionFactory.newTransaction(environment.getDataSource(), level, autoCommit);

我們回到openSessionFromDataSource方法，獲取了 JdbcTransactionFactory 后，調用 JdbcTransactionFactory 的newTransaction方法創建一個事務對象.

當然因為代碼中採用TransactionFactory 接口作為聲明對象.所以無論分之一傳回來的是哪個工廠對象.在分支二中都可以執行.

我們先講 JdbcTransactionFactory的情況.

分支二中調用的是這個newTransaction方法.(還有一個重載的)

public Transaction newTransaction(Connection conn) {
  return new JdbcTransaction(conn);
}

這就到了另一個類中JdbcTransaction中.

JdbcTransaction

我刪掉其中的實現代碼

public class JdbcTransaction implements Transaction {

  private static final Log log = LogFactory.getLog(JdbcTransaction.class);

  protected Connection connection;
  protected DataSource dataSource;
  protected TransactionIsolationLevel level;
  protected boolean autoCommmit;

  public JdbcTransaction(DataSource ds, TransactionIsolationLevel desiredLevel, boolean desiredAutoCommit) {
    dataSource = ds;
    level = desiredLevel;
    autoCommmit = desiredAutoCommit;
  }

  public JdbcTransaction(Connection connection) {
    this.connection = connection;
  }

  public Connection getConnection() throws SQLException {
  
  }

  public void commit() throws SQLException {
   
  }

  public void rollback() throws SQLException {
    
  }

  public void close() throws SQLException {
    
  }

  protected void setDesiredAutoCommit(boolean desiredAutoCommit) {
   
  }

  protected void resetAutoCommit() {
    
  }

  protected void openConnection() throws SQLException {
   
  }

}

其實只要看了代碼你就會發現,這個類中的方法,和我們調用session的方法高度重合.比如commit,rollback等等.而且還能設置事務的隔離級別

所以我們有理由認為,這個類就是對jdbc連接部分的封裝.

總結

至此分支二結束,我們對於 標籤在xml中的存在情況,會返回兩種截然不同對象.一種是作為jdbc連接封裝的 JdbcTransaction對象.另一個則是 ManagedTransaction對象(這個沒講….)

分支三

第三分支我們將回到Configuration對象.

Configuration對象

法此時已經創建好事務對象。接下來將事務對象執行器作為參數執行 configuration 的 newExecutor 方法來獲取一個 執行器類。我們看看該方法實現：

首先第一句將判斷是否傳入了一個excutorType參數,如果沒有就用默認的參數.

也就是 ExecutorType.SIMPLE(前面出現過)，然後根據執行的類型來創建不同的執行器，默認是 SimpleExecutor 執行器.

Mybatis有三種基本的Executor執行器:

• SimpleExecutor：每執行一次update或select，就開啟一個Statement對象，用完立刻關閉Statement對象。

• ReuseExecutor：執行update或select，以sql作為key查找Statement對象，存在就使用，不存在就創建，用完后，不關閉Statement對象，而是放置於Map<String, Statement>內，供下一次使用。簡言之，就是重複使用Statement對象。

• BatchExecutor：執行update（沒有select，JDBC批處理不支持select），將所有sql都添加到批處理中（addBatch()），等待統一執行（executeBatch()），它緩存了多個Statement對象，每個Statement對象都是addBatch()完畢后，等待逐一執行executeBatch()批處理。與JDBC批處理相同。

作用範圍：Executor的這些特點，都嚴格限制在SqlSession生命周期範圍內。

然後我們看下一句部分

Executor executor;
//看看上文.這是根據傳入的內容不同,最終結果是
if (ExecutorType.BATCH == executorType) {
  executor = new BatchExecutor(this, transaction);
} else if (ExecutorType.REUSE == executorType) {
  executor = new ReuseExecutor(this, transaction);
} else {
  executor = new SimpleExecutor(this, transaction);
}

我們先將 BatchExecutor執行器.

該類包裝了事務對象，延遲加載的隊列，本地緩存，永久緩存，配置對象，還包裝了自己。

傳入的兩個參數分別為存儲了配置信息的Configuration對象,以及封裝了jdbc中連接數據庫部分代碼的JdbcTransaction對象.

回到 newExecutor 方法，判斷是否使用緩存，默認是true， 則將剛剛的執行器包裝到新的 CachingExecutor 緩存執行器中。最後將執行器添加到所有的攔截器中（如果配置了話），我們這裏沒有配置。

到此分支三結束

總結:

我們從用從分支二得到的對象,構建了一個執行器.這個執行對象,包括事務對象(即連jdbc連接部分的控制封裝.JdbcTransaction)，延遲加載的隊列，本地緩存，永久緩存，配置對象(Configuration)，還包裝了自己。

四號分支

我們已經有了執行器，此時創建 DefaultSqlSession 對象，攜帶 configuration, executor, autoCommit 三個參數，該構造器就是簡單的賦值過程。我們有必要看看該類的結構：

該類包含了常用的所有方法，包括事務方法，可以說，該類封裝了執行器和事務類。而執行器才是具體的執行工作人員。

至此，我們已經完成了 SqlSession 的創建過程。

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

【其他文章推薦】

※為什麼 USB CONNECTOR 是電子產業重要的元件?

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

※台北網頁設計公司全省服務真心推薦

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

※推薦評價好的iphone維修中心

1. 知識點思維導圖

(圖比較大，可以右鍵在新窗口打開)

2. 經典的wait()/notify()/notifyAll()實現生產者/消費者編程範式深入分析 & synchronized

注：本節代碼和部分分析參考了你真的懂wait、notify和notifyAll嗎。

看下面一段典型的wait()/notify()/notifyAll()代碼，對於值得注意的細節，用註釋標出。

import java.util.ArrayList;
import java.util.List;

public class Something {
    private Buffer mBuf = new Buffer(); // 共享的池子

    public void produce() {
        synchronized (this) { // 注1、注2
            while (mBuf.isFull()) { // 注3
                try {
                    wait(); // 注4
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            mBuf.add();
            notifyAll();  // 注5、注6
        }
    }

    public void consume() {
        synchronized (this) { // 見注1、注2
            while (mBuf.isEmpty()) { // 注3
                try {
                    wait(); // 注4
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            mBuf.remove();
            notifyAll(); // 注5、注6
        }
    }

    private class Buffer {
        private static final int MAX_CAPACITY = 1;
        private List innerList = new ArrayList<>(MAX_CAPACITY);

        void add() {
            if (isFull()) {
                throw new IndexOutOfBoundsException();
            } else {
                innerList.add(new Object());
            }
            System.out.println(Thread.currentThread().toString() + " add");

        }

        void remove() {
            if (isEmpty()) {
                throw new IndexOutOfBoundsException();
            } else {
                innerList.remove(MAX_CAPACITY - 1);
            }
            System.out.println(Thread.currentThread().toString() + " remove");
        }

        boolean isEmpty() {
            return innerList.isEmpty();
        }

        boolean isFull() {
            return innerList.size() == MAX_CAPACITY;
        }
    }

    public static void main(String[] args) {
        Something sth = new Something();
        Runnable runProduce = new Runnable() {
            int count = 4;

            @Override
            public void run() {
                while (count-- > 0) {
                    sth.produce();
                }
            }
        };
        Runnable runConsume = new Runnable() {
            int count = 4;

            @Override
            public void run() {
                while (count-- > 0) {
                    sth.consume();
                }
            }
        };
        for (int i = 0; i < 2; i++) {
            new Thread(runConsume).start();
        }
        for (int i = 0; i < 2; i++) {
            new Thread(runProduce).start();
        }
    }
}

• 注1：wait()/notify()/notifyAll()必須在synchronized塊中使用
• 注2：使用synchronized(this)的原因是，這段代碼的main()，是通過實例化Something的對象，並使用它的方法來進行生產/消費的，因此是一個指向this的對象鎖。不同的場景，需要注意同步的對象的選擇。
• 注3：必須使用while循環來包裹wait()。設想一種場景：存在多個生產者或多個消費者消費者，以多個生成者為例，在緩衝區滿的情況下，如果生產者通過notify()喚醒的線程仍是生產者，如果不使用while，那麼獲取鎖的線程無法重新進入睡眠，鎖也不能釋放，造成死鎖。
• 注4：wait()會釋放鎖
• 注5：notfiy()、notifyAll()會通知其他在wait的線程來獲取鎖，但是獲取鎖的真正時機是鎖的原先持有者退出synchronized塊的時候。
• 注6：使用notifyAll()而不是notfiy()的原因是，仍考慮注3的場景，假如生產者喚醒的也是生產者，後者發現緩衝區滿重新進入阻塞，此時沒有辦法再喚醒在等待的消費者線程了，也會造成死鎖。

擴展知識點1：synchronized塊的兩個隊列

synchronized入口是將線程放入同步隊列，wait()是將線程放入阻塞隊列。notify()/notifyAll()實際上是把線程從阻塞隊列放入同步隊列。wait/notify/notifyAll方法需不需要被包含在synchronized塊中，為什麼？

擴展知識點2：synchronized重入原理

synchronized是可重入的，原理是它內部包含了一個計數器，進入時+1，退出時-1。 Java多線程：synchronized的可重入性

擴展知識點3：作用範圍

synchronized支持三種用法：修飾靜態方法、修飾實例方法、修飾代碼塊，前兩種分別鎖類對象、鎖對象實例，最後一種根據傳入的值來決定鎖什麼。
synchronized是基於java的對象頭實現的，從字節碼可以看出包括了一對進入&退出的監視器。
深入理解Java併發之synchronized實現原理

擴展知識點4：分佈式環境synchronized的意義

單看應用所運行的的單個宿主機，仍然可能有多線程的處理模式，在這個前提下使用併發相關技術是必須的。

擴展知識點5：哪些方法釋放資源，釋放鎖

所謂資源，指的是系統資源。

wait(): 線程進入阻塞狀態，釋放資源，釋放鎖，Object類final方法（notify/notifyAll一樣，不可改寫）。
sleep(): 線程進入阻塞態，釋放資源，（如果在synchronized中）不釋放鎖，進入阻塞狀態，喚醒隨機線程，Thread類靜態native方法。
yield(): 線程進入就緒態，釋放資源，（如果在synchronized中）不釋放鎖，進入可執行狀態，選擇優先級高的線程執行，Thread類靜態native方法。
如果線程產生的異常沒有被捕獲，會釋放鎖。
sleep和yield的比較

可以進一步地將阻塞劃分為同步阻塞——進入synchronized時沒獲取到鎖、等待阻塞——wait()、其他阻塞——sleep()/join()，可以參考線程的狀態及sleep、wait等方法的區別

再進一步地，Java線程狀態轉移可以用下圖表示（圖源《Java 併發編程藝術》4.1.4 節）

WAITING狀態的線程是不會消耗CPU資源的。

3. 線程數調優

理論篇

本節參考了《Java併發編程實戰》8.2節，也可以結合面試問我，創建多少個線程合適？我該怎麼說幫助理解，其中的計算題比較有價值。

前置知識

I/O密集型任務：I/O任務執行時CPU空閑。
CPU密集型任務：進行計算
有的任務是二者兼備的。為了便於分析，不考慮。

定性分析

場景：單核單線程/單核多線程/多核多線程。單核多線程+CPU密集型不能提升執行效率，多核+CPU密集型任務可以；單核多線程+I/O密集型可以提升執行效率。
因此，I/O耗時越多，線程也傾向於變多來充分利用IO等待時間。

定量分析

對於CPU密集型，線程數量=CPU 核數（邏輯）即可。特別的，為了防止線程在程序運行異常時不空轉，額外多設一個線程線程數量 = CPU 核數（邏輯）+ 1
對於I/O密集型，最佳線程數 = CPU核數 * (1/CPU利用率) = CPU核數 * (1 + I/O耗時/CPU耗時)
為什麼CPU利用率=1/(1+ I/O耗時/CPU耗時)？簡單推導一下：

1/(1+ I/O耗時/CPU耗時) = 1/((CPU耗時+I/O耗時)/ CPU耗時) = CPU耗時/總耗時 = CPU利用率

如何獲取參數——CPU利用率？

因為利用率不是一成不變的，需要通過全面的系統監控工具（如SkyWalking、CAT、zipkin），並長期進行調整觀測。
可以先取2N即2倍核數，此時即假設I/O耗時/CPU耗時=1:1，再進行調優。

阿姆達爾定律

CPU併發處理時性能提升上限。
S=1/(1-a+a/n)
其中，a為并行計算部分所佔比例，n為并行處理結點個數。
簡單粗暴理解【阿姆達爾定律】

Java線程池篇

基本屬性

/**
 * 使用給定的初始參數和默認線程工廠創建一個新的ThreadPoolExecutor ，並拒絕執行處理程序。 使用Executors工廠方法之一可能更方便，而不是這種通用構造函數。
參數
 *  corePoolSize - 即使空閑時仍保留在池中的線程數，除非設置 allowCoreThreadTimeOut
 *  maximumPoolSize - 池中允許的最大線程數
 *  keepAliveTime - 當線程數大於核心時，這是多餘的空閑線程在終止之前等待新任務的最大時間。
 *  unit - keepAliveTime參數的時間單位
 *  workQueue - 在執行任務之前用於保存任務的隊列。 該隊列將僅保存execute方法提交的Runnable任務。
 * threadFactory - 執行程序創建新線程時使用的工廠
 */
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory)

常見線程池

由java.util.concurrent.Executors創建的線程池比較常用，而不是使用ThreadPoolExecutor的構造方法。

名稱	特性
newFixedThreadPool	線程池大小為固定值
newSingleThreadExecutor	線程池大小固定為1
newCachedThreadPool	線程池大小初始為0，默認最大值為MAX INTEGER
newScheduledExecutor	延遲執行任務或按周期重複執行任務

線程工廠的作用

用來創建線程，統一在創建線程時設置一些參數，如是否守護線程。線程一些特性等，如優先級。
可參考004-多線程-JUC線程池-ThreadFactory線程工廠

4. 併發容器相關

併發容器可以說是一個面試時的高頻問題了，網絡上也有很多介紹，這裏就不重複解讀，將相關的知識整理一下，邊看源碼邊讀文章效果會很好。
先提一句，Vector是線程安全的，為啥現在不推薦用呢？看源碼可以知道，它將大部分方法都加了synchronized，犧牲了性能換取線程安全，是不可取的。如果真的有需要線程安全的容器，可以用Collections.synchronizedList()來手動給list加synchronized。
再補充一句，其實Vector和Collections.synchronizedList()使用複合操作或迭代器Iterator時也不是線程安全的，具體解釋會在下一篇博客Java容器中介紹。

ConcurrentHashMap

先重點介紹Map的兩個實現類HashMap和ConcurrentHashMap

• HashMap和ConcurrentHashMap HashMap？ConcurrentHashMap？相信看完這篇沒人能難住你！
• HashMap擴容原理：HashMap的擴容機制—resize()
• 多線程下HashMap擴容resize可能導致鏈表循環
• 這兩個數據結構在JDK1.7到1.8時，當數目達到一個閾值時，都從鏈表改用了紅黑樹
• HashMap的node重寫了equals方法來比較節點。Objects.equals會調用Object的equals，對於Object實現類則是實現類自己的equals。

 public final boolean equals(Object o) {
     if (o == this)
         return true;
     if (o instanceof Map.Entry) {
         Map.Entry<?,?> e = (Map.Entry<?,?>)o;
         if (Objects.equals(key, e.getKey()) &&
             Objects.equals(value, e.getValue()))
             return true;
     }
     return false;
 }

ConcurrentLinkedQueue

ConcurrentLinkedQueue使用CAS無鎖操作，保證入隊出隊的線程安全，但不保證遍歷時的線程安全。遍歷要想線程安全需要單獨加鎖。
由於算法的特性，這個容器的尾結點是有延遲的，tail不一定是尾節點，但p.next == null的節點一定是尾結點。
入隊出隊操作很抽象，需要畫圖幫助理解源碼，對應的源碼分析可參考併發容器-ConcurrentLinkedQueue詳解。

5. AQS解讀

抽象隊列同步器AbstractQueuedSynchronizer（AQS）是JUC中很多併發工具類的基礎，用來抽象各種併發控制行為，如ReentranLock、Semaphore。
之前試着直接讀源碼，效果不太好，還是建議結合質量較高的文章來讀，這裏推薦一篇：Java併發之AQS詳解，並且作者還在不斷更新。
這裏簡單記錄一下總結的點。

結構特點

• volatile int state標記位，標識當前的同步狀態。具體的用法和使用AQS的工具類有關。同時，在做CAS的時候，state的狀態變更是通過計算該變量在對象的偏移量來設置的。
• CLH隊列。CLH鎖（Craig，Landin andHagersten）是一種在SMP（Symmetric Multi-Processor對稱多處理器）架構下基於單鏈表的高性能的自旋鎖，隊列中每個節點代表一個自旋的線程，每個線程只需在代表前一個線程的節點上的布爾值locked自旋即可，如圖

  圖源和CLH的詳解見算法：CLH鎖的原理及實現

• exclusiveOwnerThread獨佔模式的擁有者，記錄現在是哪個線程佔用這個AQS。

操作特點

• 對state使用>0和<0的判斷，初看代碼很難看懂，這麼寫的原因是負值表示結點處於有效等待狀態，而正值表示結點已被取消。
• 大量的CAS：無論是獲取鎖、入隊、獲取鎖失敗后的自旋，全部是依賴CAS實現的。
• 沒有使用synchronized：不難理解，如果使用了同步塊，那麼其實現ReentranLock就沒有和synchronized比較的價值了。不過這一點很少有文章專門提到。
• LockSupport類的unpark()/park()方法的使用：回憶上文提到的線程狀態，如果線程獲取不到AQS控制的資源，需要將線程置於waiting，對應可選的方法是wait()/join()/park()。在AQS這個場景下，顯然一沒有synchronized，二沒有顯式的在同一個代碼塊中用join處理多線程（藉助隊列來處理線程，線程相互之間不感知），那麼只有park()才能達到目的。

處理流程

獲取資源acquire(int)

1. 嘗試獲取資源(改寫state)，成功則返回
2. CAS（失敗則自旋）加入等待隊列隊尾
3. 在隊列中自旋，嘗試獲取一次資源（前提：隊頭+ tryAcquire()成功），每次失敗都會更改線程狀態為waiting。自旋時會看看前驅有沒有失效的節點（即不再請求資源的），如果有就插隊到最前面並把前面無效節點清理掉便於gc
4. waiting狀態中不響應中斷，獲取資源后才會補一個自我中斷selfInterrupt (調用Thread.currentThread().interrupt())

釋放資源release(int)

1. 嘗試釋放，成功則處理後續動作，失敗直接返回false
2. 喚醒（unpark）等待隊列的下一個線程。如果當前節點沒找到後繼，則從隊尾tail從后往前找。

共享模式獲取資源acquireShared(int)

除了抽象方法tryAcquireShared()以外，基本和acquire(int)一致。
在等待隊列中獲取資源后，會調用獨有的setHeadAndPropagate()方法，將這個節點設為頭結點的同時，檢查後續節點是否可以獲取資源。

共享模式釋放資源releaseShared()

和release(int)區別在於，喚醒後繼時，不要求當前線程節點狀態為0。舉例：當前線程A原先擁有5個資源，釋放1個，後繼的等待線程B剛好需要1個，那麼此時A、B就可以并行了。

未實現的方法

為了便於使用AQS的類更加個性化，AQS有一下方法直接拋UnsupportedOperationException。

• isHeldExclusively()
• tryAcquire()
• tryRelease()
• tryAcquireShared()
• tryReleaseShared()
  不寫成abstract方法的原因是，避免強迫不需要對應方法的類實現這些方法。比如要寫一個獨佔的鎖，那麼就不需要實現共享模式的方法。

AQS小結

讀完源碼總結一下，AQS是一個維護資源和請求資源的線程之間的關係的隊列。對於資源（有序或無序的）獲取和釋放已經提取成了線程的出入隊方法，這個隊列同時維護上線程的自旋狀態和管理線程間的睡眠喚醒。

應用

本節可以看作為《JAVA併發變成實戰》14.6的引申。

ReentrantLock

用內部類Sync實現AQS，Sync實現ReentrantLock的行為。Sync又有FairSync和UnfairSync兩種實現。FairSync，lock對應aquire(1)；UnfairSync，lock先CAS試着獲取一次，不行再aquire(1)。
實際上，ReentrantLock的公平/非公平鎖只在首次lock時有區別，入隊后喚醒仍是按順序的。可以參考reentrantLock公平鎖和非公平鎖源碼解析
Sync只實現了獨佔模式。

注意：CyclicBarrier直接用了ReentrantLock，沒有直接用AQS。

Semaphore

和ReentrantLock類似，Semaphore也有一個內部類Sync，但相反的是這個Sync只實現了共享模式的acquire()/release()。
Semaphore在acquire()/release()時會計算資源余量並設置，其中unfair模式下的acquire會無條件自旋CAS，fair模式下只有在AQS里不存在排隊中的後繼的情況下才會CAS，否則自旋。

CountDownLatch

同樣有一個內部類Sync，但是不再區分fair/unfair，並且是共享模式的。
await()調用的是acquireSharedInterruptibly()，自然也存在自旋的可能，只是編程時一般不這麼用。countDown()時釋放一個資源繼續在releaseShared()里自旋直到全部釋放。

FutureTask

新版的FutureTask已經重寫，不再使用AQS，這裏就不再提了。

ReentrantReadWriteLock

可重入讀寫鎖，涉及到鎖升級，這裏沒有研究的很透徹，有興趣可以自行了解。
注意到讀鎖和寫鎖是共用同一個Sync的。

6 JMM到底是個啥？

The Java memory model specifies how the Java virtual machine works with the computer’s memory (RAM)。
—— Java Memory Model
雖然被冠以”模型“，JMM實際上是定義JVM如何與計算機內存協同工作的規範，也可以理解為__指令__與其操作的__數據__的行為。這樣，自然而然地引入了指令重排序、變量更改的可見性的探討。
JMM定義了一個偏序關係，稱之為happens-before。不滿足happens-before的兩個操作可以由JVM進行重排序。

6.1 什麼是偏序關係

假設 R 是集合 A 上的關係，如果R是自反的、反對稱的和傳遞的，則稱 R 是 A 上的一個偏序。偏序關係
那麼，自反的、反對稱的和傳遞的，又是什麼？下面粘貼了百度百科相關詞條：

• 自反關係：設 R是 A上的一個二元關係，若對於 A中的每一個元素 a， (a,a)都屬於 R，則稱 R為自反關係。
• 反對稱關係：集合 A 上的二元關係 R 是反對稱的，當且僅當對於X里的任意元素a, b，若a R-關係於 b 且 b R-關係於 a，則a=b。
• 傳遞關係：令R是A上的二元關係，對於A中任意的 ，若 ，且 ，則 ，則稱R具有傳遞性(或稱R是傳遞關係)。

上面的反對稱關係稍微不好理解，轉換成逆否命題就好理解了：若a!=b，那麼R中不能同存在aRb和bRa。

6.2 偏序關係和JMM

將R作為兩個操作間的關係，集合A是所有操作的集合，那麼就可以理解JMM為什麼實際上是一套偏序關係了。

6.3 happens-before規則

這部分的說明很多文章都是有差異，比如鎖原則，JLS（Java Language Specification，Java語言規範）特指的是監視器鎖，只不過顯式鎖和內置鎖有相同的內存語義而已。這裏直接摘錄原文並配上說明。原文見Chapter 17. Threads and Locks

If we have two actions x and y, we write hb(x, y) to indicate that x happens-before y.

If x and y are actions of the same thread and x comes before y in program order, then hb(x, y).

There is a happens-before edge from the end of a constructor of an object to the start of a finalizer (§12.6) for that object.

If an action x synchronizes-with a following action y, then we also have hb(x, y).

If hb(x, y) and hb(y, z), then hb(x, z).

The wait methods of class Object (§17.2.1) have lock and unlock actions associated with them; their happens-before relationships are defined by these associated actions.

It should be noted that the presence of a happens-before relationship between two actions does not necessarily imply that they have to take place in that order in an implementation. If the reordering produces results consistent with a legal execution, it is not illegal.

For example, the write of a default value to every field of an object constructed by a thread need not happen before the beginning of that thread, as long as no read ever observes that fact.

More specifically, if two actions share a happens-before relationship, they do not necessarily have to appear to have happened in that order to any code with which they do not share a happens-before relationship. Writes in one thread that are in a data race with reads in another thread may, for example, appear to occur out of order to those reads.

The happens-before relation defines when data races take place.

A set of synchronization edges, S, is sufficient if it is the minimal set such that the transitive closure of S with the program order determines all of the happens-before edges in the execution. This set is unique.

It follows from the above definitions that:

An unlock on a monitor happens-before every subsequent lock on that monitor.

A write to a volatile field (§8.3.1.4) happens-before every subsequent read of that field.

A call to start() on a thread happens-before any actions in the started thread.

All actions in a thread happen-before any other thread successfully returns from a join() on that thread.

The default initialization of any object happens-before any other actions (other than default-writes) of a program.

試着翻譯一下各項規則：
先定義hb(x, y)表示操作x和操作y的happens-before關係。

1. 同一個線程的操作x, y，代碼中順序為x, y，那麼hb(x, y)
2. 對象構造方法要早於終結方法完成
3. 如果x synchronizes-with y那麼hb(x,y)
4. 傳遞性，hb(x, y) 且hb(y,z)則hb(x,z)
5. 同一個監視器鎖解鎖需要hb所有加鎖（注：該規則擴展到顯式鎖）
6. volatile的讀hb所有寫（該規則擴展到原子操作）
7. 線程start() hb所有它的啟動后的任何動作
8. 線程中所有操作hb 對它的join()
9. 對象默認構造器hb對它的讀寫

synchronizes-with又是啥？查閱了一下，表示”這個關係表示一個行為在發生時，它首先把要操作的那些對象同主存同步完畢之後才繼續執行“。參考JMM（Java內存模型）中的核心概念。
JLS上對happens-before的解釋翻譯過來還是不太好理解，《Java併發編程實戰》的解釋和Happens-beofre 先行發生原則（JVM 規範）一樣，可以參考下。

最後可以發現，JMM只是一套規則，並沒有提到具體的實現，程序員知道Java有這一重保證即可。

7. 短篇話題整理總結

7.1 ThreadLocal的用法總結

應用場景：在多線程下替代類的靜態變量(static)，在多線程環境進行單個 類 的數據隔離。

為什麼推薦使用static修飾ThreadLocal？

這時才能保證”一個線程，一個ThreadLocal”，否則便成了“一個線程，（多個對象實例時）多個ThreadLocal”。
可能會有內存泄漏：ThreadLocalMap的key(Thread對象)是弱引用，但value不是，如果key被回收，value還在。解法是手動remove掉。
（本節參考了《Java併發編程實戰》）

7.2 CountDownLatch和CyclicBarrier區別

https://blog.csdn.net/tolcf/article/details/50925145
CountDownLatch的子任務調用countDown後會繼續執行直至該線程結束。
CyclicBarrier的子任務await時會暫停執行；可重複使用，即await的數目達到設置的值時，喚醒所有await的線程進行下一輪。

7.3 ReentrantLock用了CAS但為什麼不是樂觀鎖？

https://blog.csdn.net/qq_35688140/article/details/101223701
我的看法：因為仍有可能造成阻塞，而樂觀鎖更新失敗則會直接返回（CAS允許自旋）。
換一個角度，悲觀鎖是預先做最壞的設想——一定會有其他任務併發，那麼就先佔好坑再更新；樂觀鎖則是認為不一定有併發，更新時判斷再是否有問題。這樣看來ReentrantLock從使用方式上來說是悲觀鎖。

7.4 雙重檢查加鎖

public classDoubleCheckedLocking{ //1
      private static Instance instance; //2
      public staticI nstance getInstance(){ //3
            if(instance==null){ //4:第一次檢查
                  synchronized(DoubleCheckedLocking.class){ //5:加鎖
                        if(instance==null) //6:第二次檢查
                              instance=newInstance(); //7:問題的根源出在這裏
                  } //8
            }//9
            return instance;
      }
}

問題

一個線程看到另一個線程初始化該類的部分構造的對象，即以上代碼註釋第4處這裏讀到非null但未完全初始化

原因

註釋第7處，創建對象實例的三步指令1.分配內存空間2.初始化3.引用指向分配的地址，2和3可能重排序

解決

方案1，給instance加violatile
方案2，使用佔位類，在類初始化時初始化對象，如下

public class InstanceFactory {
      private static class InstanceHolder{
            public static Instance instance= newInstance();
      }
      public static Instance getInstance() {
            return InstanceHolder.instance;　　//這裏將導致InstanceHolder類被初始化
      }
}

7.5 FutureTask

FutureTask是Future的實現類，可以使用Future來接收線程池的submit()方法，也可以直接用FutureTask封裝任務，作為submit()的參數。具體的用法可以參考Java併發編程：Callable、Future和FutureTask 。
新版的FutureTask不再使用AQS。
FutureTask設置了當前工作線程，對於其任務維護了一個內部狀態轉換狀態機，通過CAS做狀態判斷和轉換。
當其他線程來get()時，如果任務未完成則放入等待隊列，自旋直到取到結果（for循環+LockSupport.park()），否則直接取結果。
具體實現原理可以參考《線程池系列一》-FutureTask原理講解與源碼剖析。

7.6 JDK1.6鎖優化之輕量級鎖和偏向鎖

實際上二者是有聯繫的，都是基於mark word實現。這個轉換關係可以用《深入理解Java虛擬機》第十三章的插圖表現

但是這個圖沒有體現輕量級鎖釋放后，仍可恢復為可偏向的。

7.7 問題排查三板斧

1. top查看內存佔用率，-H可以看線程（不會完整展示），-p [pid]看指定進程的線程
   注意：linux線程和進程id都是在pid這一列展示的。
2. pstack跟蹤進程棧，strace查看進程的系統操作。多次執行pstack來觀察進程是不是總是處於某種上下文中。
3. jps直接獲取java進程id，jstat看java進程情況。jstate可用不同的參數來查看不同緯度的信息：類加載情況、gc統計、堆內存統計、新生代/老年代內存統計等，具體可以參考【JVM】jstat命令詳解—JVM的統計監測工具
4. jstack打印java線程堆棧，和pstack展示方式很像，是java緯度的
5. jmap打印java內存情況，-dump可以生成dump文件
6. 分析dump文件，如MAT

8. LeetCode多線程習題

原題目和詳解參考Concurrency – 力扣

1114.按序打印

按照指定次序完成一系列動作，可以看做是buffer為1的1對1生產者消費者模型。

1115.交替打印FooBar

交替執行(不完全是生產者-消費者模型)某些動作。
可用的解法：

• synchronized
• Semaphore
• CountDownLatch
• CyclicBarrier
• Lock

1116.打印零與奇偶數:0102…

和1114類似

1188. 設計有限阻塞隊列

注意: 使用synchronize解法時，wait()應置於while中循環判斷.
如果只用if，喚醒后不再次判斷dequeue可能NPE
本題可以加深理解為什麼要用while

1195. 交替打印字符串

根據AC的解法推斷, 每個線程只調用對應方法一次，因此需要在方法內部循環
不推薦只用synchronized,四個線程按順序打印, 如果使用單一的鎖很容易飢餓導致超時

推薦解法：
AtomicInteger無鎖解法
CylicBarrier高效解法
Semaphore加鎖

1279. 紅綠燈路口

題目難懂，暗含條件：車來時紅綠燈不是綠的，則強制變綠通過。紅綠燈本身的時間沒有嚴格控制

延伸閱讀

什麼是分佈式鎖
一文了解分佈式鎖

9. 未展開的話題

併發研究之CPU緩存一致性協議(MESI)
線程池原理（四）：ScheduledThreadPoolExecutor
一半是天使一半是魔鬼的Unsafe類詳解 —— unsafe類都有什麼？用偏移量直接訪問、線程操作、內存管理和內存屏障、CAS

10. 其他參考

Java併發高頻面試題

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

【其他文章推薦】

※USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

※台北網頁設計公司這麼多該如何選擇?

※智慧手機時代的來臨，RWD網頁設計為架站首選

※評比南投搬家公司費用收費行情懶人包大公開

※回頭車貨運收費標準

一：背景

1. 講故事

好消息，.NET 5.0 終於在2020年6月10日發布了第五個預覽版，眼尖的同學一定看到了在這個版本中終於支持了 C# 9.0，此處有掌聲，太好了！！！

.Net5官方鏈接

可以看到目前的C#9還是預覽版，實現了一部分新語法供開發者提前嘗鮮，從github的roslyn倉庫上可以看到目前準備實現 17個新特性，現階段已經實現了8個，其中的 In Progress 表示正在開發中。

新特性預覽

2. 安裝必備

• 下載最新的net5 sdk吧： dotnet-sdk-5.0.100-preview.5.20279.10-win-x64.exe

• 下載最新的 visual studio 2019 preview 2

找好你自己的vs版本類型哦。。。

二：新特性研究

1. Target-typed new

這個取名一定要留給學易經的大師傅，沒見過世面的我不敢造次，取得不佳影響時運，所謂 運去金成鐵, 時來鐵似金 ，不過大概意思就是說直接new你定義的局部變量的類型，用issues中總結的話就是：

Summary: Allow Point p = new (x, y);
Shipped in preview in 16.7p1.

接下來就是全部代碼，看看使用前 和 使用后 的具體差別。

    class Program
    {
        static void Main(string[] args)
        {
            //老語法
            var person = new Person("mary", "123456");

            //新語法
            Person person2 = new("mary", "123456");

            Console.WriteLine($"person={person}person2={person2}");
        }
    }

    public class Person
    {
        private string username;
        private string password;

        public Person(string username, string password)
        {
            this.username = username;
            this.password = password;
        }

        public override string ToString()
        {
            return $"username={username},password={password} \n";
        }
    }

然後用ilspy去看看下面的il代碼，是不是省略了Person，讓自己心裏踏實一點。

總的來說這語法還行吧，能起到延長鍵盤使用壽命的功效。

2. Lambda discard parameters

從字面上看大概就是說可以在lambda上使用取消參數，聽起來怪怪的，那本意是什麼呢？有時候lambda上的匿名方法簽名的參數是不需要的，但在以前必須實打實的定義，這樣就會污染方法體，也就是可以在body中被訪問，如下圖：

但有時候因為客觀原因必須使用Func<int,int,int>這樣的委託，而且還不想讓方法簽名的參數污染方法體，我猜測在函數式編程中有這樣的場景吧，可能有點類似MVC中的EmptyResult效果。

好了，我想你大概知道啥意思了，接下來實操一把。。。

    Func<int, int, int> func = (_, _) =>
    {
        return 0;
    };

    var result = func(10, 20);

從圖中可以看到，我在方法體中是找不到所謂的 _ 變量的，這就神奇了，怎麼做到的呢？ 帶着這個好奇心看看它的IL代碼是個什麼樣子。

.method private hidebysig static 
	void Main (
		string[] args
	) cil managed 
{
	// Method begins at RVA 0x2048
	// Code size 45 (0x2d)
	.maxstack 3
	.entrypoint
	.locals init (
		[0] class [System.Runtime]System.Func`3<int32, int32, int32> func,
		[1] int32 result
	)

	IL_0000: nop
	IL_0001: ldsfld class [System.Runtime]System.Func`3<int32, int32, int32> ConsoleApp1.Program/'<>c'::'<>9__0_0'
	IL_0006: dup
	IL_0007: brtrue.s IL_0020

	IL_0009: pop
	IL_000a: ldsfld class ConsoleApp1.Program/'<>c' ConsoleApp1.Program/'<>c'::'<>9'
	IL_000f: ldftn instance int32 ConsoleApp1.Program/'<>c'::'<Main>b__0_0'(int32, int32)
	IL_0015: newobj instance void class [System.Runtime]System.Func`3<int32, int32, int32>::.ctor(object, native int)
	IL_001a: dup
	IL_001b: stsfld class [System.Runtime]System.Func`3<int32, int32, int32> ConsoleApp1.Program/'<>c'::'<>9__0_0'

	IL_0020: stloc.0
	IL_0021: ldloc.0
	IL_0022: ldc.i4.s 10
	IL_0024: ldc.i4.s 20
	IL_0026: callvirt instance !2 class [System.Runtime]System.Func`3<int32, int32, int32>::Invoke(!0, !1)
	IL_002b: stloc.1
	IL_002c: ret
} // end of method Program::Main

從上面的IL代碼來看 匿名方法 變成了<>c類的<Main>b__0_0方法,完整簽名： ConsoleApp1.Program/'<>c'::'<Main>b__0_0'(int32, int32)，然後再找一下 <Main>b__0_0 方法的定義。

.class nested private auto ansi sealed serializable beforefieldinit '<>c'
	extends [System.Runtime]System.Object
	.method assembly hidebysig 
		instance int32 '<Main>b__0_0' (
			int32 _,
			int32 _
		) cil managed 
	{
		// Method begins at RVA 0x2100
		// Code size 7 (0x7)
		.maxstack 1
		.locals init (
			[0] int32
		)

		IL_0000: nop
		IL_0001: ldc.i4.0
		IL_0002: stloc.0
		IL_0003: br.s IL_0005

		IL_0005: ldloc.0
		IL_0006: ret
	} // end of method '<>c'::'<Main>b__0_0'

這說明什麼呢？ 說明兩個參數是真實存在的，但編譯器搗了鬼，做了語法上的限制，不讓你訪問所謂的 _。

等等。。。有一個問題，IL中的方法簽名怎麼是這樣的： <Main>b__0_0 (int32 _,int32 _) , 大家應該知道方法簽名中不可以出現重複的參數名，比如下面這樣定義肯定是報錯的。

這說明什麼？ 說明這個語法糖不僅需要編譯器支持，更需要底層的JIT支持，那怎麼證明呢？我們用windbg去底層挖一挖。。。為了方便調試,修改如下：

        static void Main(string[] args)
        {
            Func<int, int, int> func = (_, _) =>
            {
                Console.WriteLine("進入方法體了！！！");
                Console.ReadLine();
                return 0;
            };

            var result = func(10, 20);
        }

0:000> !clrstack -p
OS Thread Id: 0x52e8 (0)
0000007035F7E5C0 00007ffaff362655 ConsoleApp1.Program+c.b__0_0(Int32, Int32) [C:\5\ConsoleApp1\ConsoleApp1\Program.cs @ 13]
    PARAMETERS:
        this (0x0000007035F7E600) = 0x000001968000cb48
        _ (0x0000007035F7E608) = 0x000000000000000a
        _ (0x0000007035F7E610) = 0x0000000000000014

從圖中可以看到，雖然都是 _ ,但在線程棧上是完完全全的兩個棧地址。 0x0000007035F7E608 和 0x0000007035F7E610。

三：總結

總的來說，C#是越來越向函數式編程靠攏，越來越像Scala，就像Jquery的口號一樣： Write less，do more。

好了，先就說這兩個吧，大家先安裝好工具，明天繼續解剖~~~

如您有更多問題與我互動，掃描下方進來吧~

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

【其他文章推薦】

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

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

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

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

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

※超省錢租車方案

前言

生產環境中運行的組件，只要有數據存儲，定時備份、災難恢復是必修課，mysql數據庫的備份方案已經非常成熟，Elasticsearch也同樣有成熟的數據備份、恢復方案，我們來了解一下。

概要

本篇介紹Elasticsearch生產集群數據的數據備份、恢復和升級的常規操作。

curl命令

curl是Linux操作的必備工具，Elasticsearch生產環境的搭建，不能保證都能使用kibana訪問到，而Elasticsearch Restful API都可以使用curl工具來完成訪問。

使用curl還有一個好處：有些操作需要一連串的請求才能完成，我們可以使用shell腳本將這些關聯的操作，封裝到腳本里，後續使用起來就非常方便。

如果有定時執行的命令，也是使用shell將一系列操作封裝好，運用Linux自帶的crontab進行觸發。

後續的一些操作命令，將會用curl來完成，並且只需要將完整的curl請求拷貝到kibana的dev tools上，kibana能夠自動轉化成我們之前常見的請求，非常方便。

在Linux下的請求命令：

[esuser@elasticsearch02 ~]$ curl -XGET 'http://elasticsearch02:9200/music/children/_search?pretty' -H 'Content-Type: application/json' -d '
{
  "query": {
    "match_all": {}
  }
}
'

完整的命令拷貝到dev tools里時，自動會變成：

GET /music/children/_search
{

  "query": {

    "match_all": {}

  }

}

這工具真是強大，不過反過來操作不行的，我已經試過了。

curl命令，有Body體的，記得加上-H 'Content-Type: application/json'，?pretty參數可以讓響應結果格式化輸出。

數據備份

我們知道Elasticsearch的索引拆分成多個shard進行存儲在磁盤裡，shard雖然分了primary shard和replica shard，可以保證集群的數據不丟失，數據訪問不間斷，但如果機房停電導致集群節點全部宕機這種重大事故時，我們就需要提前定期地對數據進行備份，以防萬一。

既然是磁盤文件存儲，那存儲介質的選擇就有很多了：本地磁盤，NAS，文件存儲服務器（如FastDFS、HDFS等），各種雲存儲（Amazon S3, 阿里雲OSS）等

同樣的，Elasticsearch也提供snapshot api命令來完成數據備份操作，可以把集群當前的狀態和數據全部存儲到一個其他目錄上，本地路徑或網絡路徑均可，並且支持增量備份。可以根據數據量來決定備份的執行頻率，增量備份的速度還是很快的。

創建備份倉庫

我們把倉庫地址暫定為本地磁盤的/home/esuser/esbackup目錄，

首先，我們需要在elasticsearch.yml配置文件中加上

path.repo: /home/esuser/esbackup

並重啟Elasticsearch。

啟動成功后，發送創建倉庫的請求：

[esuser@elasticsearch02 ~]$ curl -XPUT 'http://elasticsearch02:9200/_snapshot/esbackup?pretty' -H 'Content-Type: application/json' -d '
{
    "type": "fs", 
    "settings": {
        "location": "/home/esuser/esbackup",
        "max_snapshot_bytes_per_sec" : "50mb", 
        "max_restore_bytes_per_sec" : "50mb"
    }
}
'
{"acknowledged":true}
[esuser@elasticsearch02 ~]$ 

參數解釋：

• type: 倉庫的類型名稱，請求里都是fs，表示file system。
• location: 倉庫的地址，要與elasticsearch.yml配置文件相同，否則會報錯
• max_snapshot_bytes_per_sec: 指定數據從Elasticsearch到倉庫（數據備份）的寫入速度上限，默認是20mb/s
• max_restore_bytes_per_sec: 指定數據從倉庫到Elasticsearch（數據恢復）的寫入速度上限，默認也是20mb/s

用於限流的兩個參數，需要根據實際的網絡進行設置，如果備份目錄在同一局域網內，可以設置得大一些，便於加快備份和恢復的速度。

也有查詢命令可以看倉庫的信息：

[esuser@elasticsearch02 ~]$ curl -XGET 'http://elasticsearch02:9200/_snapshot/esbackup?pretty'

{"esbackup":{"type":"fs","settings":{"location":"/home/esuser/esbackup","max_restore_bytes_per_sec":"50mb","max_snapshot_bytes_per_sec":"50mb"}}}

[esuser@elasticsearch02 ~]$

使用hdfs創建倉庫

大數據這塊跟hadoop生態整合還是非常推薦的方案，數據備份這塊可以用hadoop下的hdfs分佈式文件存儲系統，關於hadoop集群的搭建方法，需要自行完成，本篇末尾有補充說明，可供參考。

對Elasticsearch來說，需要安裝repository-hdfs的插件，我們的Elasticsearch版本是6.3.1，對應的插件則使用repository-hdfs-6.3.1.zip，hadoop則使用2.8.1版本的。

插件下載安裝命令：

./elasticsearch-plugin install https://artifacts.elastic.co/downloads/elasticsearch-plugins/repository-hdfs/repository-hdfs-6.3.1.zip

如果生產環境的服務器無法連接外網，可以先在其他機器上下載好，上傳到生產服務器，解壓到本地，再執行安裝：

./elasticsearch-plugin install file:///opt/elasticsearch/repository-hdfs-6.3.1

安裝完成後記得重啟Elasticsearch節點。

查看節點狀態：

[esuser@elasticsearch02 ~]$ curl -XGET elasticsearch02:9200/_cat/nodes?v

ip             heap.percent ram.percent cpu load_1m load_5m load_15m node.role master name
192.168.17.137           38          95   2    0.03    0.03     0.05 mdi       *      node-1

創建hdfs倉庫

先查看節點的shard信息

[esuser@elasticsearch02 ~]$ curl -XGET 'http://elasticsearch02:9200/_count?pretty' -H 'Content-Type: application/json' -d '
 {
     "query": {
         "match_all": {}
     }
}'


{
  "count" : 5392,
  "_shards" : {
    "total" : 108,
    "successful" : 108,
    "skipped" : 0,
    "failed" : 0
  }
}

創建一個hdfs的倉庫，名稱為hdfsbackup

[esuser@elasticsearch02 ~]$ curl -XPUT  'http://elasticsearch02:9200/_snapshot/hdfsbackup?pretty' -H 'Content-Type: application/json' -d '
 {
   "type": "hdfs",
   "settings": {
     "uri": "hdfs://elasticsearch02:9000/",
     "path": "/home/esuser/hdfsbackup",
   "conf.dfs.client.read.shortcircuit": "false",
   "max_snapshot_bytes_per_sec" : "50mb", 
     "max_restore_bytes_per_sec" : "50mb"
   }
 }'

{
  "acknowledged" : true
}

驗證倉庫

倉庫創建好了之後，可以用verify命令驗證一下：

[esuser@elasticsearch02 ~]$ curl -XPOST 'http://elasticsearch02:9200/_snapshot/hdfsbackup/_verify?pretty'
{
  "nodes" : {
    "A1s1uus7TpuDSiT4xFLOoQ" : {
      "name" : "node-1"
    }
  }
}

索引備份

倉庫創建好並驗證完成后，可以執行snapshot api對索引進行備份了，

如果不指定索引名稱，表示備份當前所有open狀態的索引都備份，還有一個參數wait_for_completion，表示是否需要等待備份完成后才響應結果，默認是false，請求提交後會立即返回，然後備份操作在後台異步執行，如果設置為true，請求就變成同步方式，後台備份完成后，才會有響應。建議使用默認值即可，有時備份的整個過程會持續1-2小時。

示例1：備份所有的索引，備份名稱為snapshot_20200122

[esuser@elasticsearch02 ~]$ curl -XPUT 'http://elasticsearch02:9200/_snapshot/hdfsbackup/snapshot_20200122?pretty'
{
  "accepted" : true
}

示例2：備份索引music的數據，備份名稱為snapshot_20200122_02，並指定wait_for_completion為true

[esuser@elasticsearch02 ~]$ curl -XPUT 'http://elasticsearch02:9200/_snapshot/hdfsbackup/snapshot_20200122_02?wait_for_completion=true&pretty' -H 'Content-Type: application/json' -d '
{
  "indices": "music",
  "ignore_unavailable": true,
  "include_global_state": false,
  "partial": true
}'


{
  "snapshot" : {
    "snapshot" : "snapshot_20200122_02",
    "uuid" : "KRXnzc6XSWagCQO92EQx6A",
    "version_id" : 6030199,
    "version" : "6.3.1",
    "indices" : [
      "music"
    ],
    "include_global_state" : false,
    "state" : "SUCCESS",
    "start_time" : "2020-01-22T07:11:06.594Z",
    "start_time_in_millis" : 1579677066594,
    "end_time" : "2020-01-22T07:11:07.313Z",
    "end_time_in_millis" : 1579677067313,
    "duration_in_millis" : 719,
    "failures" : [ ],
    "shards" : {
      "total" : 5,
      "failed" : 0,
      "successful" : 5
    }
  }
}

這條命令中幾個參數介紹：

• indices：索引名稱，允許寫多個，用”,”分隔，支持通配符。
• ignore_unavailable：可選值true/false，如果為true，indices里不存在的index就可以忽略掉，備份操作正常執行，默認是false，如果某個index不存在，備份操作會提示失敗。
• include_global_state：可選值true/false，含義是要不要備份集群的全局state數據。
• partial：可選值true/false，是否支持備份部分shard的數據。默認值為false，如果索引的部分primary shard不可用，partial為false時備份過程會提示失敗。

使用snapshot api對數據的備份是增量進行的，執行snapshotting的時候，Elasticsearch會分析已經存在於倉庫中的snapshot對應的index file，在前一次snapshot基礎上，僅備份創建的或者發生過修改的index files。這就允許多個snapshot在倉庫中可以用一種緊湊的模式來存儲，非常節省存儲空間，並且snapshotting過程是不會阻塞所有的Elasticsearch讀寫操作的。

同樣的，snapshot作為數據快照，在它之後寫入index中的數據，是不會反應到這次snapshot中的，snapshot數據的內容包含index的副本，也可以選擇是否保存全局的cluster元數據，元數據裡面包含了全局的cluster設置和template。

每次只能執行一次snapshot操作，如果某個shard正在被snapshot備份，那麼這個shard此時就不能被移動到其他node上去，這會影響shard rebalance的操作。只有在snapshot結束之後，這個shard才能夠被移動到其他的node上去。

查看snapshot備份列表

1. 查看倉庫內所有的備份列表

curl -XGET 'http://elasticsearch02:9200/_snapshot/hdfsbackup/_all?pretty'

1. 查看單個備份數據

[esuser@elasticsearch02 ~]$ curl -XGET 'http://elasticsearch02:9200/_snapshot/hdfsbackup/snapshot_20200122_02?pretty'
{
  "snapshots" : [
    {
      "snapshot" : "snapshot_20200122_02",
      "uuid" : "KRXnzc6XSWagCQO92EQx6A",
      "version_id" : 6030199,
      "version" : "6.3.1",
      "indices" : [
        "music"
      ],
      "include_global_state" : false,
      "state" : "SUCCESS",
      "start_time" : "2020-01-22T07:11:06.594Z",
      "start_time_in_millis" : 1579677066594,
      "end_time" : "2020-01-22T07:11:07.313Z",
      "end_time_in_millis" : 1579677067313,
      "duration_in_millis" : 719,
      "failures" : [ ],
      "shards" : {
        "total" : 5,
        "failed" : 0,
        "successful" : 5
      }
    }
  ]
}

刪除snapshot備份

如果需要刪除某個snapshot備份快照，一定要使用delete命令，造成別自個跑到服務器目錄下做rm操作，因為snapshot是增量備份的，裏面有各種依賴關係，極可能損壞backup數據，記住不要上來就自己干文件，讓人家標準的命令來執行，命令如下：

[esuser@elasticsearch02 ~]$ curl -XDELETE 'http://elasticsearch02:9200/_snapshot/hdfsbackup/snapshot_20200122?pretty'
{
  "acknowledged" : true
}

查看備份進度

備份過程長短視數據量而定，wait_for_completion設置為true雖然可以同步得到結果，但時間太長的話也不現實，我們是希望備份操作後台自己搞，我們時不時的看看進度就行，其實還是調用的snapshot的get操作命令，加上_status參數即可，備份過程中會显示什麼時間開始的，有幾個shard在備份等等信息：

curl -XGET 'http://elasticsearch02:9200/_snapshot/hdfsbackup/snapshot_20200122_02/_status?pretty'

取消備份

正在備份的數據可以執行取消，使用的是delete命令：

curl -XDELETE 'http://elasticsearch02:9200/_snapshot/hdfsbackup/snapshot_20200122?pretty'

這個命令有兩個作用：

1. 如果備份正在進行中，那麼取消備份操作，並且刪除備份了一半的數據。
2. 如果備份已經完成，直接刪除備份數據。

數據恢復

生產環境的備份操作，是定期執行的，執行的頻率看實際的數據量，有1天執行1次的，有4小時一次的，簡單的操作是使用shell腳本封裝備份的命令，然後使用Linux的crontab定時執行。

既然數據有備份，那如果數據出現異常，或者需要使用到備份數據時，恢復操作就能派上用場了。

常規恢復

數據恢復使用restore命令，示例如下：

[esuser@elasticsearch02 ~]$ curl -XPOST 'http://elasticsearch02:9200/_snapshot/hdfsbackup/snapshot_20200122_02/_restore?pretty'
{
  "accepted" : true
}

注意一下被恢復的索引，必須全部是close狀態的，否則會報錯，關閉索引的命令：

[esuser@elasticsearch02 ~]$ curl -XPOST  'http://elasticsearch02:9200/music/_close?pretty'

恢復完成后，索引自動還原成open狀態。

同樣有些參數可以進行選擇：

[esuser@elasticsearch02 ~]$ curl -XPOST 'http://elasticsearch02:9200/_snapshot/hdfsbackup/snapshot_20200122_02/_restore
{
    "indices": "music", 
	"ignore_unavailable": true,
	"include_global_state": true
}

默認會把備份數據里的索引全部還原，我們可以使用indices參數指定需要恢復的索引名稱。同樣可以使用wait_for_completion參數，ignore_unavailable、partial和include_global_state與備份時效果相同，不贅述。

監控restore的進度

與備份類似，調用的recovery的get操作命令查看恢復的進度：

curl -XGET 'http://elasticsearch02:9200/music/_recovery?pretty'

music為索引名稱。

取消restore

與備份類似，delete正在恢復的索引可以取消恢復過程：

curl -XDELETE 'http://elasticsearch02:9200/music'

集群升級

我們現在使用的版本是6.3.1，目前官網最新版本已經是7.5.2了，如果沒有重大的變更或嚴重bug報告的情況下，一般是不需要做升級，畢竟升級有風險，發布要謹慎。

這裏就簡單說一下通用的步驟，謹慎操作:

1. 查看官網最新版本的文檔，從當前版本到目標版本的升級，有哪些變化，新加入的功能和修復的bug。
2. 在開發環境或測試環境先執行升級，相應的插件也做一次匹配升級，穩定運行幾個項目版本周期后，再考慮生產環境的升級事宜。
3. 升級前對數據進行全量的備份，萬一升級失敗，還有挽救的餘地。
4. 申請生產環境升級的時間窗口，逐個node進行升級驗證。

補充hadoop集群搭建

Elasticsearch的數據備份，通常建議的實踐方案是結合hadoop的hdfs文件存儲，這裏我們搭建一個hadoop的集群環境用作演示，hadoop相關的基礎知識請自行了解，已經掌握的童鞋可以跳過。

版本環境：
hadoop 2.8.1

虛擬機環境

hadoop集群至少需要3個節點。我們選用elasticsearch02、elasticsearch03、elasticsearch04三台機器用於搭建。

1. 下載解壓

官網下載hadoop-2.8.1.tar.gz，解壓至/opt/hadoop目錄

1. 設置環境變量

演示環境擁有root權限，就介紹一種最簡單的設置方法，修改/etc/profile文件，添加變量後記得source一下該文件。

[root@elasticsearch02 ~]# vi /etc/profile

# 文件末尾添加
export HADOOP_HOME=/opt/hadoop/hadoop-2.8.1
export PATH=${HADOOP_HOME}/bin:$PATH

[root@elasticsearch02 ~]# source /etc/profile

1. 創建hadoop數據目錄，啟動hadoop時我們使用esuser賬戶，就在/home/esuser下創建目錄，如 /home/esuser/hadoopdata

2. 修改hadoop的配置文件，在/opt/hadoop/hadoop-2.8.1/etc/hadoop目錄下，基本上是添加配置，涉及的配置文件：

• core-site.xml
• hdfs-site.xml
• yarn-site.xml
• mapred-site.xml
• slaves(注：我們選定elasticsearch02為master，其餘兩個為slave)

示例修改如下：

core-site.xml

<property>
  <name>fs.defaultFS</name>
  <value>hdfs://elasticsearch02:9000</value>
</property>

hdfs-site.xml

<property>
  <name>dfs.namenode.name.dir</name>
  <value>/home/esuser/hadoopdata/namenode</value>
</property>
<property>
  <name>dfs.datanode.data.dir</name>
  <value>/home/esuser/hadoopdata/datanode</value>
</property>

yarn-site.xml

<property>
  <name>yarn.resourcemanager.hostname</name>
  <value>elasticsearch02</value>
</property>

mapred-site.xml

<property>
  <name>mapreduce.framework.name</name>
  <value>yarn</value>
</property>

slaves

elasticsearch03
elasticsearch04

1. 拷貝設置后的文件到另外兩台機器上

scp -r /opt/hadoop/hadoop-2.8.1 esuser@elasticsearch03:/opt/hadoop/hadoop-2.8.1
scp -r /opt/hadoop/hadoop-2.8.1 esuser@elasticsearch04:/opt/hadoop/hadoop-2.8.1

拷貝的文件有點大，需要等一會兒，拷貝完成后，在elasticsearch03、elasticsearch04再設置一次HADOOP_HOME環境變量

1. 啟動集群

格式化namenode，在hadoop master節點(elasticsearch02)，HADOOP_HOME/sbin目錄下執行hdfs namenode -format

執行啟動命令：start-dfs.sh
這個啟動過程會建立到elasticsearch03、elasticsearch04的ssh連接，輸入esuser的密碼即可，也可以提前建立好免密ssh連接。

我們只需要用它的hdfs服務，其他的組件可以不啟動。

驗證啟動是否成功，三台機器分別輸入jps，看下面的進程，如無意外理論上應該是這樣：
elasticsearch02：NameNode、SecondaryNameNode
elasticsearch03：DataNode
elasticsearch04：DataNode

同時在瀏覽器上輸入hadoop master的控制台地址：http://192.168.17.137:50070/dfshealth.html#tab-overview，應該能看到這兩個界面：

datanodes看到2個結點，表示集群啟動成功，如果只能看到一個或一個都沒有，可以查看相應的日誌：/opt/hadoop/hadoop-2.8.1/logs

Error: JAVA_HOME is not set and could not be found 錯誤解決辦法

這個明明已經設置了JAVA_HOME，並且export命令也能看到，啟動時死活就是不行，不跟他杠了，直接在/opt/hadoop/hadoop-2.8.1/etc/hadoop/hadoop-env.sh文件加上

export JAVA_HOME="/opt/jdk1.8.0_211"

小結

本篇主要以hadoop分佈式文件存儲為背景，講解了Elasticsearch數據的備份與恢復，可以了解一下。集群版本升級這類操作，實踐起來比較複雜，受項目本身影響比較大，這裏就簡單提及要注意的地方，沒有作詳細的案例操作，真要有版本升級的操作，請各位慎重操作，多驗證，確保測試環境充分測試后再上生產，記得數據要備份。

專註Java高併發、分佈式架構，更多技術乾貨分享與心得，請關注公眾號：Java架構社區
可以掃左邊二維碼添加好友，邀請你加入Java架構社區微信群共同探討技術

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

【其他文章推薦】

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

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

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

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

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

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

前言

寫這篇博客主要目的是記錄一下自己的學習過程，只能是簡單入門級別的，因為水平有限就寫到哪算哪吧，寫的不對之處歡迎指正。
代碼放在：https://github.com/xiajingren/NetCoreMicroserviceDemo

什麼是微服務？

關於微服務的概念解釋網上有很多…
個人理解，微服務是一種系統架構模式，它和語言無關，和框架無關，和工具無關，和服務器環境無關…
微服務思想是將傳統的單體系統按照業務拆分成多個職責單一、且可獨立運行的接口服務。至於服務如何拆分，沒有明確的定義。
幾乎任何後端語言都能做微服務開發。
微服務也並不是完美無缺的，微服務架構會帶來更多的問題，增加系統的複雜度，引入更多的技術棧…

創建項目

一個客戶端，一個產品服務，一個訂單服務。3個項目都是asp.net core web應用程序。創建項目的時候記得啟用一下Docker支持，或者後面添加也行。

為產品、訂單服務添加一些基礎代碼，就簡單的返回一下 服務名稱，當前時間，服務的ip、端口。

在Docker中運行服務

為了方便，我使用Docker來運行服務，不用Docker也行，關於docker的安裝及基本使用就不介紹了。

• build鏡像：

在項目根目錄打開PowerShell窗口執行：docker build -t productapi -f ./Product.API/Dockerfile .

Successfully代表build成功了。

• 運行容器:

執行：docker run -d -p 9050:80 --name productservice productapi

執行：docker ps查看運行的容器：

沒問題，使用瀏覽器訪問一下接口：

也沒問題，其中的ip端口是Docker容器內部的ip端口，所以端口是80，這個無所謂。

• 產品服務部署好了，下面部署一下訂單服務，也是同樣的流程，就把指令簡單貼一下吧：

build鏡像：docker build -t orderapi -f ./Order.API/Dockerfile .
運行容器：docker run -d -p 9060:80 --name orderservice orderapi
瀏覽器訪問一下：

OK，訂單服務也部署完成了。

客戶端調用

客戶端我這裏只做了一個web客戶端，實際可能是各種業務系統、什麼PC端、手機端、小程序。。。這個明白就好，為了簡單就不搞那麼多了。

• 因為客戶端需要http請求服務端接口，所以需要一個http請求客戶端，我個人比較習慣RestSharp，安利一波：https://github.com/restsharp/RestSharp

• 添加基礎代碼：

IServiceHelper.cs：

    public interface IServiceHelper
    {
        /// <summary>
        /// 獲取產品數據
        /// </summary>
        /// <returns></returns>
        Task<string> GetProduct();

        /// <summary>
        /// 獲取訂單數據
        /// </summary>
        /// <returns></returns>
        Task<string> GetOrder();
    }

ServiceHelper.cs：

    public class ServiceHelper : IServiceHelper
    {
        public async Task<string> GetOrder()
        {
            string serviceUrl = "http://localhost:9060";//訂單服務的地址，可以放在配置文件或者數據庫等等...

            var Client = new RestClient(serviceUrl);
            var request = new RestRequest("/orders", Method.GET);

            var response = await Client.ExecuteAsync(request);
            return response.Content;
        }

        public async Task<string> GetProduct()
        {
            string serviceUrl = "http://localhost:9050";//產品服務的地址，可以放在配置文件或者數據庫等等...

            var Client = new RestClient(serviceUrl);
            var request = new RestRequest("/products", Method.GET);

            var response = await Client.ExecuteAsync(request);
            return response.Content;
        }
    }

Startup.cs：

    public class Startup
    {
        public Startup(IConfiguration configuration)
        {
            Configuration = configuration;
        }

        public IConfiguration Configuration { get; }

        // This method gets called by the runtime. Use this method to add services to the container.
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddControllersWithViews();
            
            //注入IServiceHelper
            services.AddSingleton<IServiceHelper, ServiceHelper>();
        }

        // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
        public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
        {
            if (env.IsDevelopment())
            {
                app.UseDeveloperExceptionPage();
            }
            else
            {
                app.UseExceptionHandler("/Home/Error");
            }
            app.UseStaticFiles();

            app.UseRouting();

            app.UseAuthorization();

            app.UseEndpoints(endpoints =>
            {
                endpoints.MapControllerRoute(
                    name: "default",
                    pattern: "{controller=Home}/{action=Index}/{id?}");
            });
        }
    }

HomeController.cs：

    public class HomeController : Controller
    {
        private readonly ILogger<HomeController> _logger;
        private readonly IServiceHelper _serviceHelper;

        public HomeController(ILogger<HomeController> logger, IServiceHelper serviceHelper)
        {
            _logger = logger;
            _serviceHelper = serviceHelper;
        }

        public async Task<IActionResult> Index()
        {
            ViewBag.OrderData = await _serviceHelper.GetOrder();
            ViewBag.ProductData = await _serviceHelper.GetProduct();

            return View();
        }

        public IActionResult Privacy()
        {
            return View();
        }

        [ResponseCache(Duration = 0, Location = ResponseCacheLocation.None, NoStore = true)]
        public IActionResult Error()
        {
            return View(new ErrorViewModel { RequestId = Activity.Current?.Id ?? HttpContext.TraceIdentifier });
        }
    }

Index.cshtml：

@{
    ViewData["Title"] = "Home Page";
}

<div class="text-center">
    <h1 class="display-4">Welcome</h1>
    <p>
        @ViewBag.OrderData
    </p>
    <p>
        @ViewBag.ProductData
    </p>
</div>

代碼比較簡單，這裏就不用docker了，直接控制台啟動，使用瀏覽器訪問：

• 一切正常。進行到這裏，各個服務也獨立運行了，客戶端也能正常調用了，貌似算是完成一個簡易的微服務了。但是，微服務架構最重要的原則就是——“高可用”。以上的做法明顯不能滿足高可用性，因為任何一個服務掛掉，所有依賴這個服務的業務系統都會受影響。

停止一下訂單服務：docker stop orderservice

訂單服務停止，導致客戶端業務系統無法獲取訂單數據。
要解決這個問題，很容易想到：集群。

簡單的服務集群

既然單個服務實例有掛掉的風險，那麼部署多個服務實例就好了嘛，只要大家不同時全掛就行。

• 使用docker運行多個服務實例：

docker run -d -p 9061:80 --name orderservice1 orderapi
docker run -d -p 9062:80 --name orderservice2 orderapi
docker run -d -p 9051:80 --name productservice1 productapi
docker run -d -p 9052:80 --name productservice2 productapi

現在訂單服務和產品服務都增加到3個服務實例。

• 那麼稍微改造一下客戶端代碼吧：
  ServiceHelper.cs：

public class ServiceHelper : IServiceHelper
    {
        public async Task<string> GetOrder()
        {
            string[] serviceUrls = { "http://localhost:9060", "http://localhost:9061", "http://localhost:9062" };//訂單服務的地址，可以放在配置文件或者數據庫等等...

            //每次隨機訪問一個服務實例
            var Client = new RestClient(serviceUrls[new Random().Next(0, 3)]);
            var request = new RestRequest("/orders", Method.GET);

            var response = await Client.ExecuteAsync(request);
            return response.Content;
        }

        public async Task<string> GetProduct()
        {
            string[] serviceUrls = { "http://localhost:9050", "http://localhost:9051", "http://localhost:9052" };//產品服務的地址，可以放在配置文件或者數據庫等等...

            //每次隨機訪問一個服務實例
            var Client = new RestClient(serviceUrls[new Random().Next(0, 3)]);
            var request = new RestRequest("/products", Method.GET);

            var response = await Client.ExecuteAsync(request);
            return response.Content;
        }
    }

當然拿到這些服務地址可以自己做複雜的負載均衡策略，比如輪詢，隨機，權重等等 都行，甚至在中間弄個nginx也可以。這些不是重點，所以就簡單做一個隨機吧，每次請求來了隨便訪問一個服務實例。

• 瀏覽器測試一下：

  可以看到請求被隨機分配了。但是這種做法依然不安全，如果隨機訪問到的實例剛好掛掉，那麼業務系統依然會出問題。
  簡單處理思路是：
  1.如果某個地址請求失敗了，那麼換一個地址接着執行。
  2.如果某個地址的請求連續多次失敗了，那麼就移除這個地址，下次就不會訪問到它了。
  。。。。。。
  業務系統實現以上邏輯，基本上風險就很低了，也算是大大增加了系統可用性了。

• 然後思考另一個問題：

實際應用中，上層的業務系統可能非常多，為了保證可用性，每個業務系統都去考慮服務實例掛沒掛掉嗎？
而且實際應用中服務實例的數量或者地址大多是不固定的，例如雙十一來了，流量大了，增加了一堆服務實例，這時候每個業務系統再去配置文件里配置一下這些地址嗎？雙十一過了又去把配置刪掉嗎？顯然是不現實的，服務必須要做到可靈活伸縮。

• 這時候就引入一個名詞：服務註冊與發現

未完待續…

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

【其他文章推薦】

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

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

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

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

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

新車基於UKL前驅平台打造，1系三廂將會是前驅車型，新車的動力系統或為1。5T 136馬力+6速自動變速箱，2。0T 192馬力/231馬力+8擋手自一體變速箱，其中官方聲稱1。5T車型的百公里油耗最低可以達到5。5L。新車將會有無鑰匙啟動、抬頭显示、倒車影像、全景天窗、LED大燈、pM2。

寶馬1系三廂自從誕生的那一刻就備受關注，吸睛無數，小編也和大家一樣，時刻關注着寶馬1系三廂的所有信息。畢竟，哪個男人心裏沒有這一個藍天白雲夢，開寶馬也是很多人的心愿。

但是寶馬作為豪華品牌，價格不是每個消費者都能承受的起的。寶馬進口1系的價格不是很貴，但是國人就是對兩廂車有一定的排斥心理，比較鍾愛三廂車，所以即使進口1系兩廂的價格足夠便宜，但是買單的人照樣不多。

當然，寶馬3系是三廂車，但是3系的價格在30萬左右，還是有點小貴。所以在20萬級別這個領域，出現了一個市場空缺。如果能有一台20萬的三廂寶馬，這應該是極好的選擇。

看看奧迪A3就知道了，作為豪華品牌的A3將價格做到了20萬元左右，在這個沒有直接競爭對手的領域，A3的銷量好的一塌糊塗，A3在11月份交出了9883輛的銷量，將近萬輛的銷量足以看出來這個細分市場有很大的潛力。

所以寶馬義不容辭的推出了國產版的1系三廂。

寶馬1系三廂版基於Compact Sedan概念車打造出來的，新車的設計借鑒了概念車的設計元素，同時也具有着寶馬家族化的設計風格，前大燈的造型與3系比較相似，側麵線條比較平直，從外觀看像是縮小的3系，看起來短小精悍，富有運動感。

內飾看起來還是那麼熟悉的感覺，畢竟也是採用了寶馬家族化的設計特徵，中控台造型很有層次感，並配備了8.8英寸液晶屏，同時還有大面積的鍍鉻裝飾，可以增強內飾的精緻感。

新車基於UKL前驅平台打造，1系三廂將會是前驅車型，新車的動力系統或為1.5T 136馬力+6速自動變速箱，2.0T 192馬力/231馬力+8擋手自一體變速箱，其中官方聲稱1.5T車型的百公里油耗最低可以達到5.5L。

新車將會有無鑰匙啟動、抬頭显示、倒車影像、全景天窗、LED大燈、pM2.5濾清器等，這些都是國人比較看重的配置。新車的預售價為20.5萬起，這就意味着也許不到20萬的價格就可以買到帥氣的三廂寶馬，估計很多消費者都會心動吧！如果實際售價能比預售價還低一點的話，那麼A3就會瞬間感到壓力了。

競爭對手

奧迪A3

指導價：18.49-28.10萬

A3在國內的成就有目共睹，但是它真正的競爭對手預計會在2月份正式上市銷售，憑藉著“藍天白雲”在國內的號召力，A3將會面臨很大壓力，那麼它能否守住自己的陣地，就讓我們拭目以待吧！本站聲明:網站內容來源於http://www.auto6s.com/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

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

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

※超省錢租車方案

輪轂方面使用的是美規思域Si上的18英寸旋風式輪轂，樣式非常漂亮。內飾：基本沒有變化內飾方面也僅僅進行了一些小改動，基本保持着舊款的設計。但是增加了配色，導航也終於被加入到車機裏面。依舊使用着数字显示式儀錶盤，這是一種用過都說好的設計，数字显示的時速一目瞭然，而且較高的位置真有點抬頭显示的感覺。

前言

本田思域可謂是現今緊湊型轎車的當紅辣子雞，即使需要加價也難以阻擋人們對它的喜愛之情，其中最吸引他們的就是思域上那款動力油耗表現出色的1.5T發動機以及極高的配置。而最近本田為我們帶來2017款本田傑德，這輛大兩廂汽車更換了十代思域上的1.5T發動機，功率雖然有所降低，但是配置更為豐富，那麼它具體又有着怎樣的升級呢？又是否值得購買呢？

傑德定位非常獨特，你可以將它當做一輛旅行車，也可以將其當做是MpV。而本田定位它是一輛緊湊型汽車，在筆者的眼中它是一輛大兩廂車，首先來自老款思域的平台讓它有着轎車般的駕駛感受，其次6座的布局實在和7座MpV有差異。這次2017款傑德的定價為12.99-17.99萬，購買門檻比起之前更低了。

外觀：更換了LED光源以及新增配色，更為精神

2017款傑德增加了新翠綠、波爾多紅等車身顏色，更為個性以及好看。在前臉方面進氣格柵增加了少許鍍鉻裝飾，看着更為時尚，而這種家族史設計使得傑德前臉更像是雅閣。

而集成了LED日間行車燈的LED大燈非常醒目，亮度相當高，不過這是1.5T中高配車型的專利，1.8L以及1.5T最低配都與這LED大燈無緣。

尾燈也同樣使用了LED燈源，辨識率更高，關鍵的一點是視覺效果更佳。除此以外，傑德使用了雙邊共雙出的排氣管設計，看着有一點性能味。

 

離地間隙提升了10mm，舊款托底的毛病減弱不少，但是有一點需要注意，傑德是一輛兩廂車，對於兩廂車而言這個離地間隙已經是很大的了。輪轂方面使用的是美規思域Si上的18英寸旋風式輪轂，樣式非常漂亮。

內飾：基本沒有變化

內飾方面也僅僅進行了一些小改動，基本保持着舊款的設計。但是增加了配色，導航也終於被加入到車機裏面。

依舊使用着数字显示式儀錶盤，這是一種用過都說好的設計，数字显示的時速一目瞭然，而且較高的位置真有點抬頭显示的感覺。

配置：提升不少，但尚未完美

這次配置上的提升是比較明顯的，讓傑德競爭力一下子提升不少，增加了LED大燈、日間行車燈、LED尾燈、18英寸輪轂、併線輔助系統、車道偏離系統、主動剎車、自適應巡航系統等，而且1.5T車型全系標配了全景天窗。

傑德似乎是想要向思域看齊，但傑德卻沒有思域上的自動駐車以及电子手剎，稍顯遺憾，而且作為最低配車型並沒有配備ESp車身穩定系統。總的而言，傑德在配置上以及性價比上提升不少。

動力：最大的亮點

一直以來，傑德都是被當做是老款思域的重造產物，使用了同樣的平台，同樣的動力總成。底盤表現非常出色，但1.8L自然吸氣發動機卻難以挑起重擔，油耗以及耐用度優秀，卻給人白開水的感覺。這次加入的1.5T發動機，就像是辣椒一樣一下子把這款大兩廂車變得勁爆刺激味蕾。最大功率115千瓦，最大扭矩203牛米，比起思域上略有下降，依然是優秀水平，可以理解為思域上的1.5T低功率版本。

和這款1.5T渦輪增壓發動機相配合的依然是CVT變速箱，和思域一樣，根據思域上平均8.5L的實測油耗，我們有理由相信使用同款發動機的傑德油耗也能同樣優秀。

這一次2017款本田傑德的上市把之前的一些小問題幾乎都解決了，更低的價格、換上了更強悍的動力總成、加入了導航、更豐富的舒適性配置以及安全配置，這樣看來傑德是越趨完美。加上同價位並沒有相似的競爭對手，傑德可能會成為東本的第二個思域。本站聲明:網站內容來源於http://www.auto6s.com/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

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

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

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