Java併發相關知識點梳理和研究

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網頁設計為架站首選

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

※回頭車貨運收費標準

C#9.0 終於來了,您還學的動嗎? 帶上VS一起解讀吧!(應該是全網第一篇)

一:背景

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

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

三:總結

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

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

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

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

【其他文章推薦】

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

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

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

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

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

※超省錢租車方案

Elasticsearch系列—生產數據備份恢復方案

前言

生產環境中運行的組件,只要有數據存儲,定時備份、災難恢復是必修課,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地圖已可更新顯示潭子電動車充電站設置地點!!

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

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

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

Accord.NET重啟4.0 開發

Accord.NET Framework是在AForge.NET基礎上封裝和進一步開發來的。功能也很強大,因為AForge.NET更注重與一些底層和廣度,而Accord.NET Framework更注重與機器學習這個專業,在其基礎上提供了更多統計分析和處理函數,包括圖像處理和計算機視覺算法,所以側重點不同,但都非常有用。 官方網站:http://accord-framework.net/

在項目中斷2年時間之後,作者cesarsouza 在2020年5月1日更新了項目狀態, 他在歐洲完成博士,雖然他的工作中主要使用Python完成他的工作,但是他喜歡C#/.NET,一直在考慮Accprd.NET的發展問題,5月15日重新設定了4.0 版本的路線圖https://github.com/accord-net/framework/issues/2123,  其中他寫道:“我看到這個項目仍然被認為對許多人有用,我不認為讓項目消亡符合任何人的利益。我最初認為這個項目將由ML.NET取代,但事實並非如此。我們可以轉換框架,轉而與它合作。”

我們在ML.NET的最初宣布文章中有Accord.NET的影子:

CNTK 已經死了,目前只有 Tensoflow.NET在蓬勃發展,發展的情況很不錯,隨着Accord.NET的加入,這個生態又重新激活,期待大家一起加入,推動.NET機器學習生態的發展。

(一)框架的三大功能模塊

Accord.NET框架主要有三個大的功能性模塊。

  • 分別為科學技術,
  • 信號與圖像處理,
  • 支持組件。

下面將對3個模型的命名空間和功能進行簡單介紹。可以讓大家更快的接觸和了解其功能是否是自己想要的,下面是主要的命名空間介紹。

(二) 科學計算

Accord.Math:包括矩陣擴展程序,以及一組矩陣數值計算和分解的方法,也包括一些約束和非約束問題的數值優化算法,還有一些特殊函數以及其他一些輔助工具。

Accord.Statistics:包含概率分佈、假設檢驗、線性和邏輯回歸等統計模型和方法,隱馬爾科夫模型,(隱藏)條件隨機域、主成分分析、偏最小二乘判別分析、內核方法和許多其他相關的技術。

Accord.MachineLearning: 為機器學習應用程序提供包括支持向量機,決策樹,樸素貝恭弘=叶 恭弘斯模型,k-means聚類算法,高斯混合模型和通用算法如Ransac,交叉驗證和網格搜索等算法。

Accord.Neuro:包括大量的神經網絡學習算法,如Levenberg-Marquardt,Parallel Resilient Backpropagation,Nguyen-Widrow初始化算法,深層的信念網絡和許多其他神經網絡相關的算法。具體看參考幫助文檔。

(三)信號與圖像處理

Accord.Imaging:包含特徵點探測器(如Harris, SURF, FAST and  FREAK),圖像過濾器、圖像匹配和圖像拼接方法,還有一些特徵提取器。

Accord.Audio:包含一些機器學習和統計應用程序說需要的處理、轉換過濾器以及處理音頻信號的方法。

Accord.Vision:實時人臉檢測和跟蹤,以及對人流圖像中的一般的檢測、跟蹤和轉換方法,還有動態模板匹配追蹤器。

(四) 支持組件

主要是為上述一些組件提供數據显示,繪圖的控件,分為以下幾個命名空間:

Accord.Controls:包括科學計算應用程序常見的柱狀圖、散點圖和表格數據瀏覽。

Accord.Controls.Imaging:包括用來显示和處理的圖像的WinForm控件,包含一個方便快速显示圖像的對話框。

Accord.Controls.Audio:显示波形和音頻相關性信息的WinForm控件。

Accord.Controls.Vision:包括跟蹤頭部,臉部和手部運動以及其他計算機視覺相關的任務WinForm控件。

(五) 支持的算法介紹

下面將Accord.NET框架包括的主要功能算法按照類別進行介紹。來源主要是官網介紹,進行了簡單的翻譯和整理。

1、分類(Classification)

SVM(支持向量機,類SupportVectorMachine、類KernelSupportVectorMachine、類SequentialMinimalOptimization—序列最小優化算法)、

K-NN鄰近算法(類KNearestNeighbors);

Logistic Regression(邏輯回歸)、

Decision Trees(決策樹,類DecisionTree、ID3Learning、C45Learning)、

Neural Networks(神經網絡)、

Deep Learning(深度學習)

(Deep Neural Networks深層神經網絡)、

Levenberg-Marquardt with Bayesian Regularization、

Restricted Boltzmann Machines(限制玻耳茲曼機)、

Sequence classification (序列分類),

Hidden Markov Classifiers and Hidden Conditional Random Fields(隱馬爾科夫分類器和隱藏條件隨機域)。

2、回歸(Regression)

Multiple linear regression(多元線性回歸-單因變量多自變量)、

SimpleLinearRegression(線性回歸,類SimpleLinearRegression)、

Multivariate linear regression(多元線性回歸-多因變量多自變量)、polynomial regression (多項式回歸)、logarithmic regression(對數回歸)、Logistic regression(邏輯回歸)、multinomial logistic regression(多項式邏輯回歸)(softmax) and generalized linear models(廣義線性模型)、L2-regularized L2-loss logistic regression , L2-regularized logistic regression , L1-regularized logistic regression , L2-regularized logistic regression in the dual form and regression support vector machines。

3、聚類(Clustering)

K-Means、K-Modes、Mean-Shift(均值漂移)、Gaussian Mixture Models(高斯混合模型)、Binary Split(二元分裂)、Deep Belief Networks(深層的信念網絡)、 Restricted Boltzmann Machines(限制玻耳茲曼機)。聚類算法可以應用於任意數據,包括圖像、數據表、視頻和音頻。

4、概率分佈(Distributions)

包括40多個分佈的參數和非參數估計。包括一些常見的分佈如正態分佈、柯西分佈、超幾何分佈、泊松分佈、伯努利;也包括一些特殊的分佈如Kolmogorov-Smirnov , Nakagami、Weibull、and Von-Mises distributions。也包括多元分佈如多元正態分佈、Multinomial 、Independent 、Joint and Mixture distributions。

5、假設檢驗(Hypothesis Tests)

超過35統計假設測試,包括單向和雙向方差分析測試、非參數測試如Kolmogorov-Smirnov測試和媒體中的信號測試。contingency table tests such as the Kappa test,with variations for multiple tables , as well as the Bhapkar and Bowker tests; and the more traditional Chi-Square , Z , F , T and Wald tests .

6、核方法(Kernel Methods)

內核支持向量機,多類和多標籤向量機、序列最小優化、最小二乘學習、概率學習。Including special methods for linear machines such as LIBLINEAR’s methods for Linear Coordinate Descent , Linear Newton Method , Probabilistic Coordinate Descent , Probabilistic Coordinate Descent in the Dual , Probabilistic Newton Method for L1 and L2 machines in both the dual and primal formulations .

7、圖像(Imaging)

興趣和特徵點探測器如Harris,FREAK,SURF,FAST。灰度共生矩陣,Border following,Bag-of-Visual-Words (BoW),RANSAC-based homography estimation , integral images , haralick textural feature extraction , and dense descriptors such as histogram of oriented gradients (HOG) and Local Binary Pattern (LBP).Several image filters for image processing applications such as difference of Gaussians , Gabor , Niblack and Sauvola thresholding。還有幾個圖像處理中經常用到的圖像過濾器。

8、音頻信號(Audio and Signal)

音頻信號的加載、解析、保存、過濾和轉換,如在空間域和頻域應用音頻過濾器。WAV文件、音頻捕捉、時域濾波器,高通,低通,波整流過濾器。Frequency-domain operators such as differential rectification filter and comb filter with Dirac’s delta functions . Signal generators for Cosine , Impulse , Square signals.

9、視覺(Vision)

實時人臉檢測和跟蹤,以及圖像流中檢測、跟蹤、轉換的一般的檢測方法。Contains cascade definitions , Camshift and Dynamic Template Matching trackers . Includes pre-created classifiers for human faces and some facial features such as noses。

10、降維技術

SVD奇異值分解(OctaveEnvironment.svd方法);

PCA主成分分析(類PrincipalComponent);

ICA獨立成份分析(類IndependentComponetAnalysis)

11、算法精度測算

混淆矩陣(類ConfusionMatrix);

ROC曲線評估(類ReceiverOperatingCharacteristic);

Bootstrap算法(自助算法;類(Bootstrap));

CrossValidation算法(交叉檢驗;類(CrossValidation));

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

【其他文章推薦】

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

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

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

南投搬家公司費用需注意的眉眉角角,別等搬了再說!

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

.Net Core微服務入門全紀錄(一)——項目搭建

前言

寫這篇博客主要目的是記錄一下自己的學習過程,只能是簡單入門級別的,因為水平有限就寫到哪算哪吧,寫的不對之處歡迎指正。
代碼放在: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.如果某個地址的請求連續多次失敗了,那麼就移除這個地址,下次就不會訪問到它了。
    。。。。。。
    業務系統實現以上邏輯,基本上風險就很低了,也算是大大增加了系統可用性了。

  • 然後思考另一個問題:

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

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

未完待續…

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

【其他文章推薦】

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

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

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

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

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

redis 數據刪除策略和逐出算法

數據存儲和有效期

redis 工作流程中,過期的數據並不需要馬上就要執行刪除操作。因為這些刪不刪除只是一種狀態表示,可以異步的去處理,在不忙的時候去把這些不緊急的刪除操作做了,從而保證 redis 的高效

數據的存儲

在redis中數據的存儲不僅僅需要保存數據本身還要保存數據的生命周期,也就是過期時間。在redis 中 數據的存儲結構如下圖:

獲取有效期

Redis是一種內存級數據庫,所有數據均存放在內存中,內存中的數據可以通過TTL指令獲取其狀態

刪除策略

在內存佔用與CPU佔用之間尋找一種平衡,顧此失彼都會造成整體redis性能的下降,甚至引發服務器宕機或內存泄漏。

定時刪除

創建一個定時器,當key設置過期時間,且過期時間到達時,由定時器任務立即執行對鍵的刪除操作

優點

節約內存,到時就刪除,快速釋放掉不必要的內存佔用

缺點

CPU壓力很大,無論CPU此時負載多高,均佔用CPU,會影響redis服務器響應時間和指令吞吐量

總結

用處理器性能換取存儲空間

惰性刪除

數據到達過期時間,不做處理。等下次訪問該數據,如果未過期,返回數據。發現已經過期,刪除,返回不存在。這樣每次讀寫數據都需要檢測數據是否已經到達過期時間。也就是惰性刪除總是在數據的讀寫時發生的。

expireIfNeeded函數

對所有的讀寫命令進行檢查,檢查操作的對象是否過期。過期就刪除返回過期,不過期就什麼也不做~。

執行數據寫入過程中,首先通過expireIfNeeded函數對寫入的key進行過期判斷。

/*
 * 為執行寫入操作而取出鍵 key 在數據庫 db 中的值。
 *
 * 和 lookupKeyRead 不同,這個函數不會更新服務器的命中/不命中信息。
 *
 * 找到時返回值對象,沒找到返回 NULL 。
 */
robj *lookupKeyWrite(redisDb *db, robj *key) {

    // 刪除過期鍵
    expireIfNeeded(db,key);

    // 查找並返回 key 的值對象
    return lookupKey(db,key);
}

執行數據讀取過程中,首先通過expireIfNeeded函數對寫入的key進行過期判斷。

/*
 * 為執行讀取操作而取出鍵 key 在數據庫 db 中的值。
 *
 * 並根據是否成功找到值,更新服務器的命中/不命中信息。
 *
 * 找到時返回值對象,沒找到返回 NULL 。
 */
robj *lookupKeyRead(redisDb *db, robj *key) {
    robj *val;

    // 檢查 key 釋放已經過期
    expireIfNeeded(db,key);

    // 從數據庫中取出鍵的值
    val = lookupKey(db,key);

    // 更新命中/不命中信息
    if (val == NULL)
        server.stat_keyspace_misses++;
    else
        server.stat_keyspace_hits++;

    // 返回值
    return val;
}

執行過期動作expireIfNeeded其實內部做了三件事情,分別是:

  • 查看key判斷是否過期
  • 向slave節點傳播執行過期key的動作併發送事件通知
  • 刪除過期key
/*
 * 檢查 key 是否已經過期,如果是的話,將它從數據庫中刪除。
 *
 * 返回 0 表示鍵沒有過期時間,或者鍵未過期。
 *
 * 返回 1 表示鍵已經因為過期而被刪除了。
 */
int expireIfNeeded(redisDb *db, robj *key) {

    // 取出鍵的過期時間
    mstime_t when = getExpire(db,key);
    mstime_t now;

    // 沒有過期時間
    if (when < 0) return 0; /* No expire for this key */

    /* Don't expire anything while loading. It will be done later. */
    // 如果服務器正在進行載入,那麼不進行任何過期檢查
    if (server.loading) return 0;

    // 當服務器運行在 replication 模式時
    // 附屬節點並不主動刪除 key
    // 它只返回一個邏輯上正確的返回值
    // 真正的刪除操作要等待主節點發來刪除命令時才執行
    // 從而保證數據的同步
    if (server.masterhost != NULL) return now > when;

    // 運行到這裏,表示鍵帶有過期時間,並且服務器為主節點

    /* Return when this key has not expired */
    // 如果未過期,返回 0
    if (now <= when) return 0;

    /* Delete the key */
    server.stat_expiredkeys++;

    // 向 AOF 文件和附屬節點傳播過期信息
    propagateExpire(db,key);

    // 發送事件通知
    notifyKeyspaceEvent(REDIS_NOTIFY_EXPIRED,
        "expired",key,db->id);

    // 將過期鍵從數據庫中刪除
    return dbDelete(db,key);
}

判斷key是否過期的數據結構是db->expires,也就是通過expires的數據結構判斷數據是否過期。
內部獲取過期時間並返回。

/*
 * 返回字典中包含鍵 key 的節點
 *
 * 找到返回節點,找不到返回 NULL
 *
 * T = O(1)
 */
dictEntry *dictFind(dict *d, const void *key)
{
    dictEntry *he;
    unsigned int h, idx, table;

    // 字典(的哈希表)為空
    if (d->ht[0].size == 0) return NULL; /* We don't have a table at all */

    // 如果條件允許的話,進行單步 rehash
    if (dictIsRehashing(d)) _dictRehashStep(d);

    // 計算鍵的哈希值
    h = dictHashKey(d, key);
    // 在字典的哈希表中查找這個鍵
    // T = O(1)
    for (table = 0; table <= 1; table++) {

        // 計算索引值
        idx = h & d->ht[table].sizemask;

        // 遍歷給定索引上的鏈表的所有節點,查找 key
        he = d->ht[table].table[idx];
        // T = O(1)
        while(he) {

            if (dictCompareKeys(d, key, he->key))
                return he;

            he = he->next;
        }

        // 如果程序遍歷完 0 號哈希表,仍然沒找到指定的鍵的節點
        // 那麼程序會檢查字典是否在進行 rehash ,
        // 然後才決定是直接返回 NULL ,還是繼續查找 1 號哈希表
        if (!dictIsRehashing(d)) return NULL;
    }

    // 進行到這裏時,說明兩個哈希表都沒找到
    return NULL;
}

優點

節約CPU性能,發現必須刪除的時候才刪除。

缺點

內存壓力很大,出現長期佔用內存的數據。

總結

用存儲空間換取處理器性能

定期刪除

周期性輪詢redis庫中時效性數據,採用隨機抽取的策略,利用過期數據佔比的方式刪除頻度。

優點

CPU性能佔用設置有峰值,檢測頻度可自定義設置

內存壓力不是很大,長期佔用內存的冷數據會被持續清理

缺點

需要周期性抽查存儲空間

定期刪除詳解

redis的定期刪除是通過定時任務實現的,也就是定時任務會循環調用serverCron方法。然後定時檢查過期數據的方法是databasesCron。定期刪除的一大特點就是考慮了定時刪除過期數據會佔用cpu時間,所以每次執行databasesCron的時候會限制cpu的佔用不超過25%。真正執行刪除的是 activeExpireCycle方法。

時間事件

對於持續運行的服務器來說, 服務器需要定期對自身的資源和狀態進行必要的檢查和整理, 從而讓服務器維持在一個健康穩定的狀態, 這類操作被統稱為常規操作(cron job

在 Redis 中, 常規操作由 redis.c/serverCron() 實現, 它主要執行以下操作

1 更新服務器的各類統計信息,比如時間、內存佔用、數據庫佔用情況等。

2 清理數據庫中的過期鍵值對。

3 對不合理的數據庫進行大小調整。

4 關閉和清理連接失效的客戶端。

5 嘗試進行 AOF 或 RDB 持久化操作。

6 如果服務器是主節點的話,對附屬節點進行定期同步。

7 如果處於集群模式的話,對集群進行定期同步和連接測試。

因為 serverCron() 需要在 Redis 服務器運行期間一直定期運行, 所以它是一個循環時間事件: serverCron() 會一直定期執行,直到服務器關閉為止。

在 Redis 2.6 版本中, 程序規定 serverCron() 每秒運行 10 次, 平均每 100 毫秒運行一次。 從 Redis 2.8 開始, 用戶可以通過修改 hz選項來調整 serverCron() 的每秒執行次數, 具體信息請參考 redis.conf 文件中關於 hz 選項的說明

查看hz

way1 : config get hz  # "hz" "10"
way2 : info server  # server.hz 10

serverCron()

serverCron()會定期的執行,在serverCron()執行中會調用databasesCron() 方法(serverCron()還做了其他很多事情,但是現在不討論,只談刪除策略)

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    // 略去多無關代碼

    /* We need to do a few operations on clients asynchronously. */
    // 檢查客戶端,關閉超時客戶端,並釋放客戶端多餘的緩衝區
    clientsCron();

    /* Handle background operations on Redis databases. */
    // 對數據庫執行各種操作
    databasesCron();   /* !我們關注的方法! */

databasesCron()

databasesCron() 中 調用了 activeExpireCycle()方法,來對過期的數據進行處理。(在這裏還會做一些其他操作~ 調整數據庫大小,主動和漸進式rehash)

// 對數據庫執行刪除過期鍵,調整大小,以及主動和漸進式 rehash
void databasesCron(void) {

    // 判斷是否是主服務器 如果是 執行主動過期鍵清除
    if (server.active_expire_enabled && server.masterhost == NULL)
        // 清除模式為 CYCLE_SLOW ,這個模式會盡量多清除過期鍵
        activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW);

    // 在沒有 BGSAVE 或者 BGREWRITEAOF 執行時,對哈希表進行 rehash
    if (server.rdb_child_pid == -1 && server.aof_child_pid == -1) {
        static unsigned int resize_db = 0;
        static unsigned int rehash_db = 0;
        unsigned int dbs_per_call = REDIS_DBCRON_DBS_PER_CALL;
        unsigned int j;

        /* Don't test more DBs than we have. */
        // 設定要測試的數據庫數量
        if (dbs_per_call > server.dbnum) dbs_per_call = server.dbnum;

        /* Resize */
        // 調整字典的大小
        for (j = 0; j < dbs_per_call; j++) {
            tryResizeHashTables(resize_db % server.dbnum);
            resize_db++;
        }

        /* Rehash */
        // 對字典進行漸進式 rehash
        if (server.activerehashing) {
            for (j = 0; j < dbs_per_call; j++) {
                int work_done = incrementallyRehash(rehash_db % server.dbnum);
                rehash_db++;
                if (work_done) {
                    /* If the function did some work, stop here, we'll do
                     * more at the next cron loop. */
                    break;
                }
            }
        }
    }
}

activeExpireCycle()

大致流程如下

1 遍歷指定個數的db(默認的 16 )進行刪除操作

2 針對每個db隨機獲取過期數據每次遍歷不超過指定數量(如20),發現過期數據並進行刪除。

3 如果有多於25%的keys過期,重複步驟 2

除了主動淘汰的頻率外,Redis對每次淘汰任務執行的最大時長也有一個限定,這樣保證了每次主動淘汰不會過多阻塞應用請求,以下是這個限定計算公式:

#define ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC 25 /* CPU max % for keys collection */ ``... ``timelimit = 1000000*ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC/server.hz/100;

也就是每次執行時間的25%用於過期數據刪除。

void activeExpireCycle(int type) {
    // 靜態變量,用來累積函數連續執行時的數據
    static unsigned int current_db = 0; /* Last DB tested. */
    static int timelimit_exit = 0;      /* Time limit hit in previous call? */
    static long long last_fast_cycle = 0; /* When last fast cycle ran. */

    unsigned int j, iteration = 0;
    // 默認每次處理的數據庫數量
    unsigned int dbs_per_call = REDIS_DBCRON_DBS_PER_CALL;
    // 函數開始的時間
    long long start = ustime(), timelimit;

    // 快速模式
    if (type == ACTIVE_EXPIRE_CYCLE_FAST) {
        // 如果上次函數沒有觸發 timelimit_exit ,那麼不執行處理
        if (!timelimit_exit) return;
        // 如果距離上次執行未夠一定時間,那麼不執行處理
        if (start < last_fast_cycle + ACTIVE_EXPIRE_CYCLE_FAST_DURATION*2) return;
        // 運行到這裏,說明執行快速處理,記錄當前時間
        last_fast_cycle = start;
    }

    /* 
     * 一般情況下,函數只處理 REDIS_DBCRON_DBS_PER_CALL 個數據庫,
     * 除非:
     *
     * 1) 當前數據庫的數量小於 REDIS_DBCRON_DBS_PER_CALL
     * 2) 如果上次處理遇到了時間上限,那麼這次需要對所有數據庫進行掃描,
     *     這可以避免過多的過期鍵佔用空間
     */
    if (dbs_per_call > server.dbnum || timelimit_exit)
        dbs_per_call = server.dbnum;

    // 函數處理的微秒時間上限
    // ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC 默認為 25 ,也即是 25 % 的 CPU 時間
    timelimit = 1000000*ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC/server.hz/100;
    timelimit_exit = 0;
    if (timelimit <= 0) timelimit = 1;

    // 如果是運行在快速模式之下
    // 那麼最多只能運行 FAST_DURATION 微秒 
    // 默認值為 1000 (微秒)
    if (type == ACTIVE_EXPIRE_CYCLE_FAST)
        timelimit = ACTIVE_EXPIRE_CYCLE_FAST_DURATION; /* in microseconds. */

    // 遍曆數據庫
    for (j = 0; j < dbs_per_call; j++) {
        int expired;
        // 指向要處理的數據庫
        redisDb *db = server.db+(current_db % server.dbnum);

        // 為 DB 計數器加一,如果進入 do 循環之後因為超時而跳出
        // 那麼下次會直接從下個 DB 開始處理
        current_db++;

        do {
            unsigned long num, slots;
            long long now, ttl_sum;
            int ttl_samples;

            /* If there is nothing to expire try next DB ASAP. */
            // 獲取數據庫中帶過期時間的鍵的數量
            // 如果該數量為 0 ,直接跳過這個數據庫
            if ((num = dictSize(db->expires)) == 0) {
                db->avg_ttl = 0;
                break;
            }
            // 獲取數據庫中鍵值對的數量
            slots = dictSlots(db->expires);
            // 當前時間
            now = mstime();

            // 這個數據庫的使用率低於 1% ,掃描起來太費力了(大部分都會 MISS)
            // 跳過,等待字典收縮程序運行
            if (num && slots > DICT_HT_INITIAL_SIZE &&
                (num*100/slots < 1)) break;

            /* 
             * 樣本計數器
             */
            // 已處理過期鍵計數器
            expired = 0;
            // 鍵的總 TTL 計數器
            ttl_sum = 0;
            // 總共處理的鍵計數器
            ttl_samples = 0;

            // 每次最多只能檢查 LOOKUPS_PER_LOOP 個鍵
            if (num > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP)
                num = ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP;

            // 開始遍曆數據庫
            while (num--) {
                dictEntry *de;
                long long ttl;

                // 從 expires 中隨機取出一個帶過期時間的鍵
                if ((de = dictGetRandomKey(db->expires)) == NULL) break;
                // 計算 TTL
                ttl = dictGetSignedIntegerVal(de)-now;
                // 如果鍵已經過期,那麼刪除它,並將 expired 計數器增一
                if (activeExpireCycleTryExpire(db,de,now)) expired++;
                if (ttl < 0) ttl = 0;
                // 累積鍵的 TTL
                ttl_sum += ttl;
                // 累積處理鍵的個數
                ttl_samples++;
            }

            /* Update the average TTL stats for this database. */
            // 為這個數據庫更新平均 TTL 統計數據
            if (ttl_samples) {
                // 計算當前平均值
                long long avg_ttl = ttl_sum/ttl_samples;
                
                // 如果這是第一次設置數據庫平均 TTL ,那麼進行初始化
                if (db->avg_ttl == 0) db->avg_ttl = avg_ttl;
                /* Smooth the value averaging with the previous one. */
                // 取數據庫的上次平均 TTL 和今次平均 TTL 的平均值
                db->avg_ttl = (db->avg_ttl+avg_ttl)/2;
            }

            // 我們不能用太長時間處理過期鍵,
            // 所以這個函數執行一定時間之後就要返回

            // 更新遍歷次數
            iteration++;

            // 每遍歷 16 次執行一次
            if ((iteration & 0xf) == 0 && /* check once every 16 iterations. */
                (ustime()-start) > timelimit)
            {
                // 如果遍歷次數正好是 16 的倍數
                // 並且遍歷的時間超過了 timelimit
                // 那麼斷開 timelimit_exit
                timelimit_exit = 1;
            }

            // 已經超時了,返回
            if (timelimit_exit) return;

            // 如果已刪除的過期鍵占當前總數據庫帶過期時間的鍵數量的 25 %
            // 那麼不再遍歷
        } while (expired > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP/4);
    }
}

hz調大將會提高Redis主動淘汰的頻率,如果你的Redis存儲中包含很多冷數據佔用內存過大的話,可以考慮將這個值調大,但Redis作者建議這個值不要超過100。我們實際線上將這個值調大到100,觀察到CPU會增加2%左右,但對冷數據的內存釋放速度確實有明顯的提高(通過觀察keyspace個數和used_memory大小)。

可以看出timelimit和server.hz是一個倒數的關係,也就是說hz配置越大,timelimit就越小。換句話說是每秒鐘期望的主動淘汰頻率越高,則每次淘汰最長佔用時間就越短。這裏每秒鐘的最長淘汰佔用時間是固定的250ms(1000000*ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC/100),而淘汰頻率和每次淘汰的最長時間是通過hz參數控制的。

因此當redis中的過期key比率沒有超過25%之前,提高hz可以明顯提高掃描key的最小個數。假設hz為10,則一秒內最少掃描200個key(一秒調用10次*每次最少隨機取出20個key),如果hz改為100,則一秒內最少掃描2000個key;另一方面,如果過期key比率超過25%,則掃描key的個數無上限,但是cpu時間每秒鐘最多佔用250ms。

當REDIS運行在主從模式時,只有主結點才會執行上述這兩種過期刪除策略,然後把刪除操作”del key”同步到從結點。

if (server.active_expire_enabled && server.masterhost == NULL)  // 判斷是否是主節點 從節點不需要執行activeExpireCycle()函數。
        // 清除模式為 CYCLE_SLOW ,這個模式會盡量多清除過期鍵
        activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW);

隨機個數

redis.config.ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP 決定每次循環從數據庫 expire中隨機挑選值的個數

逐出算法

如果不限制 reids 對內存使用的限制,它將會使用全部的內存。可以通過 config.memory 來指定redis 對內存的使用量 。

下面是redis 配置文件中的說明

 543 # Set a memory usage limit to the specified amount of bytes.
 544 # When the memory limit is reached Redis will try to remove keys
 545 # according to the eviction policy selected (see maxmemory-policy).
 546 #
 547 # If Redis can't remove keys according to the policy, or if the policy is
 548 # set to 'noeviction', Redis will start to reply with errors to commands
 549 # that would use more memory, like SET, LPUSH, and so on, and will continue
 550 # to reply to read-only commands like GET.
 551 #
 552 # This option is usually useful when using Redis as an LRU or LFU cache, or to
 553 # set a hard memory limit for an instance (using the 'noeviction' policy).
 554 #
 555 # WARNING: If you have replicas attached to an instance with maxmemory on,
 556 # the size of the output buffers needed to feed the replicas are subtracted
 557 # from the used memory count, so that network problems / resyncs will
 558 # not trigger a loop where keys are evicted, and in turn the output
 559 # buffer of replicas is full with DELs of keys evicted triggering the deletion
 560 # of more keys, and so forth until the database is completely emptied.
 561 #
 562 # In short... if you have replicas attached it is suggested that you set a lower
 563 # limit for maxmemory so that there is some free RAM on the system for replica
 564 # output buffers (but this is not needed if the policy is 'noeviction').
 
將內存使用限制設置為指定的字節。當已達到內存限制Redis將根據所選的逐出策略(請參閱maxmemory策略)嘗試刪除數據。

如果Redis無法根據逐出策略移除密鑰,或者策略設置為“noeviction”,Redis將開始對使用更多內存的命令(如set、LPUSH等)進行錯誤回復,並將繼續回復只讀命令,如GET。

當將Redis用作LRU或LFU緩存或設置實例的硬內存限制(使用“noeviction”策略)時,此選項通常很有用。

警告:如果將副本附加到啟用maxmemory的實例,則將從已用內存計數中減去饋送副本所需的輸出緩衝區的大小,這樣,網絡問題/重新同步將不會觸發收回密鑰的循環,而副本的輸出緩衝區將充滿收回的密鑰增量,從而觸發刪除更多鍵,依此類推,直到數據庫完全清空。

簡而言之。。。如果附加了副本,建議您設置maxmemory的下限,以便系統上有一些空閑RAM用於副本輸出緩衝區(但如果策略為“noeviction”,則不需要此限制)。

驅逐策略的配置

Maxmemery-policy volatile-lru

當前已用內存超過 maxmemory 限定時,觸發主動清理策略

易失數據清理

volatile-lru:只對設置了過期時間的key進行LRU(默認值)

volatile-random:隨機刪除即將過期key

volatile-ttl : 刪除即將過期的

volatile-lfu:挑選最近使用次數最少的數據淘汰

全部數據清理

allkeys-lru : 刪除lru算法的key

allkeys-lfu:挑選最近使用次數最少的數據淘汰

allkeys-random:隨機刪除

禁止驅逐

(Redis 4.0 默認策略)

noeviction : 永不過期,返回錯誤當mem_used內存已經超過maxmemory的設定,對於所有的讀寫請求都會觸發redis.c/freeMemoryIfNeeded(void)函數以清理超出的內存。注意這個清理過程是阻塞的,直到清理出足夠的內存空間。所以如果在達到maxmemory並且調用方還在不斷寫入的情況下,可能會反覆觸發主動清理策略,導致請求會有一定的延遲。

清理時會根據用戶配置的maxmemory-policy來做適當的清理(一般是LRU或TTL),這裏的LRU或TTL策略並不是針對redis的所有key,而是以配置文件中的maxmemory-samples個key作為樣本池進行抽樣清理。

maxmemory-samples在redis-3.0.0中的默認配置為5,如果增加,會提高LRU或TTL的精準度,redis作者測試的結果是當這個配置為10時已經非常接近全量LRU的精準度了,並且增加maxmemory-samples會導致在主動清理時消耗更多的CPU時間,建議:

1 盡量不要觸發maxmemory,最好在mem_used內存佔用達到maxmemory的一定比例后,需要考慮調大hz以加快淘汰,或者進行集群擴容。

2 如果能夠控制住內存,則可以不用修改maxmemory-samples配置;如果Redis本身就作為LRU cache服務(這種服務一般長時間處於maxmemory狀態,由Redis自動做LRU淘汰),可以適當調大maxmemory-samples。

這裏提一句,實際上redis根本就不會準確的將整個數據庫中最久未被使用的鍵刪除,而是每次從數據庫中隨機取5個鍵並刪除這5個鍵里最久未被使用的鍵。上面提到的所有的隨機的操作實際上都是這樣的,這個5可以用過redis的配置文件中的maxmemeory-samples參數配置。

數據逐出策略配置依據

使用INFO命令輸出監控信息,查詢緩存int和miss的次數,根據業務需求調優Redis配置。

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

【其他文章推薦】

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

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

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

※超省錢租車方案

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

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

基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列(四)

系列文章

  1. 基於 abp vNext 和 .NET Core 開發博客項目 – 使用 abp cli 搭建項目
  2. 基於 abp vNext 和 .NET Core 開發博客項目 – 給項目瘦身,讓它跑起來
  3. 基於 abp vNext 和 .NET Core 開發博客項目 – 完善與美化,Swagger登場
  4. 基於 abp vNext 和 .NET Core 開發博客項目 – 數據訪問和代碼優先
  5. 基於 abp vNext 和 .NET Core 開發博客項目 – 自定義倉儲之增刪改查
  6. 基於 abp vNext 和 .NET Core 開發博客項目 – 統一規範API,包裝返回模型
  7. 基於 abp vNext 和 .NET Core 開發博客項目 – 再說Swagger,分組、描述、小綠鎖
  8. 基於 abp vNext 和 .NET Core 開發博客項目 – 接入GitHub,用JWT保護你的API
  9. 基於 abp vNext 和 .NET Core 開發博客項目 – 異常處理和日誌記錄
  10. 基於 abp vNext 和 .NET Core 開發博客項目 – 使用Redis緩存數據
  11. 基於 abp vNext 和 .NET Core 開發博客項目 – 集成Hangfire實現定時任務處理
  12. 基於 abp vNext 和 .NET Core 開發博客項目 – 用AutoMapper搞定對象映射
  13. 基於 abp vNext 和 .NET Core 開發博客項目 – 定時任務最佳實戰(一)
  14. 基於 abp vNext 和 .NET Core 開發博客項目 – 定時任務最佳實戰(二)
  15. 基於 abp vNext 和 .NET Core 開發博客項目 – 定時任務最佳實戰(三)
  16. 基於 abp vNext 和 .NET Core 開發博客項目 – 博客接口實戰篇(一)
  17. 基於 abp vNext 和 .NET Core 開發博客項目 – 博客接口實戰篇(二)
  18. 基於 abp vNext 和 .NET Core 開發博客項目 – 博客接口實戰篇(三)
  19. 基於 abp vNext 和 .NET Core 開發博客項目 – 博客接口實戰篇(四)
  20. 基於 abp vNext 和 .NET Core 開發博客項目 – 博客接口實戰篇(五)
  21. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列(一)
  22. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列(二)
  23. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列(三)
  24. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列(四)
  25. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列(五)
  26. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列(六)
  27. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列(七)
  28. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列(八)
  29. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列(九)
  30. 基於 abp vNext 和 .NET Core 開發博客項目 – 終結篇之發布項目

上一篇完成了博客的分頁查詢文章列表頁面的數據綁定和分頁功能,本篇將繼續完成剩下的幾個頁面。

在開始主題之前重新解決上一篇的最後一個問題,當點擊了頭部組件的/posts鏈接時直接強制刷新了頁面,經過查看文檔和實踐有了更好的解決方案。

先將頭部組件Header.razor中的NavLink恢復成<NavLink class="menu-item" href="posts">Posts</NavLink>,不需要點擊事件了。

然後在Posts.razor中添加生命周期函數OnParametersSetAsync(),在初始化完成后執行。

/// <summary>
/// 初始化完成后執行
/// </summary>
/// <returns></returns>
protected override async Task OnParametersSetAsync()
{
    if (!page.HasValue)
    {
        page = 1;
        await RenderPage(page);
    }
}

判斷當前page參數是否有值,有值的話說明請求肯定是來自於翻頁,當page沒有值的時候就說明是頭部的菜單點進來的。那麼此時給page賦值為1,調用API加載數據即可。

分類列表

Categories.razor是分類列表頁面,上篇文章已經實現了從API獲取數據的方法,所以這裏就很簡單了,指定接受類型,然後在生命周期初始化OnInitializedAsync()中去獲取數據。

@code{
    /// <summary>
    /// categories
    /// </summary>
    private ServiceResult<IEnumerable<QueryCategoryDto>> categories;

    /// <summary>
    /// 初始化
    /// </summary>
    protected override async Task OnInitializedAsync()
    {
        // 獲取數據
        categories = await Http.GetFromJsonAsync<ServiceResult<IEnumerable<QueryCategoryDto>>>($"/blog/categories");
    }
}

當獲取到數據的時候進行綁定,沒有數據的時候還是显示加載中的組件<Loading />讓他轉圈圈。

@if (categories == null)
{
    <Loading />
}
else
{
    <div class="container">
        <div class="post-wrap categories">
            <h2 class="post-title">-&nbsp;Categories&nbsp;-</h2>
            <div class="categories-card">
                @if (categories.Success && categories.Result.Any())
                {
                    @foreach (var item in categories.Result)
                    {
                        <div class="card-item">
                            <div class="categories">
                                <a href="/category/@item.DisplayName/">
                                    <h3>
                                        <i class="iconfont iconcode" style="padding-right:3px"></i>
                                        @item.CategoryName
                                    </h3>
                                    <small>(@item.Count)</small>
                                </a>
                            </div>
                        </div>
                    }
                }
                else
                {
                    <ErrorTip />
                }
            </div>
        </div>
    </div>
}

直接循環返回的數據列表categories.Result,綁定數據就好,當獲取失敗或者沒有返回數據的時候显示錯誤提示組件<ErrorTip />

標籤列表

Categories.razor是標籤列表頁面,和分類列表HTML結構差不多一樣的,除了返回類型和接口地址不一樣,將上面代碼複製過來改改即可。

@code{
    /// <summary>
    /// tags
    /// </summary>
    private ServiceResult<IEnumerable<QueryTagDto>> tags;

    /// <summary>
    /// 初始化
    /// </summary>
    protected override async Task OnInitializedAsync()
    {
        // 獲取數據
        tags = await Http.GetFromJsonAsync<ServiceResult<IEnumerable<QueryTagDto>>>($"/blog/tags");
    }
}
@if (tags == null)
{
    <Loading />
}
else
{
    <div class="container">
        <div class="post-wrap tags">
            <h2 class="post-title">-&nbsp;Tags&nbsp;-</h2>
            <div class="tag-cloud-tags">
                @if (tags.Success && tags.Result.Any())
                {
                    @foreach (var item in tags.Result)
                    {
                        <a href="/tag/@item.DisplayName/">@item.TagName<small>(@item.Count)</small></a>
                    }
                }
                else
                {
                    <ErrorTip />
                }
            </div>
        </div>
    </div>
}

友鏈列表

FriendLinks.razor是友情鏈接列表頁面,實現方式和上面兩個套路一模一樣。

@code {
    /// <summary>
    /// friendlinks
    /// </summary>
    private ServiceResult<IEnumerable<FriendLinkDto>> friendlinks;

    /// <summary>
    /// 初始化
    /// </summary>
    protected override async Task OnInitializedAsync()
    {
        // 獲取數據
        friendlinks = await Http.GetFromJsonAsync<ServiceResult<IEnumerable<FriendLinkDto>>>($"/blog/friendlinks");
    }
}
@if (friendlinks == null)
{
    <Loading />
}
else
{
    <div class="container">
        <div class="post-wrap categories">
            <h2 class="post-title">-&nbsp;FriendLinks&nbsp;-</h2>
            <div class="categories-card">
                @if (friendlinks.Success && friendlinks.Result.Any())
                {
                    @foreach (var item in friendlinks.Result)
                    {
                        <div class="card-item">
                            <div class="categories">
                                <a target="_blank" href="@item.LinkUrl">
                                    <h3>@item.Title</h3>
                                </a>
                            </div>
                        </div>
                    }
                }
                else
                {
                    <ErrorTip />
                }
            </div>
        </div>
    </div>
}

文章列表(分類)

Posts.Category.razor是根據分類查詢文章列表頁面,他接受一個參數name,我們要根據name去API查詢數據然後綁定頁面即可。

這裏的參數name實際上就是從標籤列表傳遞過來的DisplayName的值,它是一個比較友好的名稱,我們還要通過這個值去查詢真正的分類名稱進行展示,所以這裏需要調用兩個API,這點在設計API的時候沒有考慮好,我們其實可以將這兩個API合併變成一個,後續再進行優化吧,這裏就請求兩次。

添加兩個接收參數:分類名稱和返回的文章列表數據。

/// <summary>
/// 分類名稱
/// </summary>
private string categoryName;

/// <summary>
/// 文章列表數據
/// </summary>
private ServiceResult<IEnumerable<QueryPostDto>> posts;

然後在OnInitializedAsync()初始化方法中調用API獲取數據,賦值給變量。

/// <summary>
/// 初始化
/// </summary>
protected override async Task OnInitializedAsync()
{
    // TODO:獲取數據,可以在API中合併這兩個請求。
    var category = await Http.GetFromJsonAsync<ServiceResult<string>>($"/blog/category?name={name}");
    posts = await Http.GetFromJsonAsync<ServiceResult<IEnumerable<QueryPostDto>>>($"/blog/posts/category?name={name}");

    if (category.Success)
    {
        categoryName = category.Result;
    }
}

有了數據,直接在頁面上進行循環綁定。

@if (posts == null)
{
    <Loading />
}
else
{
    <div class="container">
        <div class="post-wrap tags">
            @if (categoryName != null)
            {
                <h2 class="post-title">-&nbsp;Category&nbsp;·&nbsp;@categoryName&nbsp;-</h2>
            }
        </div>
        <div class="post-wrap archive">
            @if (posts.Success && posts.Result.Any())
            {
                @foreach (var item in posts.Result)
                {
                    <h3>@item.Year</h3>
                    @foreach (var post in item.Posts)
                    {
                        <article class="archive-item">
                            <NavLink href="@("/post"+post.Url)">@post.Title</NavLink>
                            <span class="archive-item-date">@post.CreationTime</span>
                        </article>
                    }
                }
            }
            else
            {
                <ErrorTip />
            }
        </div>
    </div>
}

文章列表(標籤)

Posts.Tag.razor是根據標籤查詢文章列表,這個和分類查詢文章列表實現方式一樣,直接上代碼。

@code {
    /// <summary>
    /// 標籤名稱參數
    /// </summary>
    [Parameter]
    public string name { get; set; }

    /// <summary>
    /// 標籤名稱
    /// </summary>
    private string tagName;

    /// <summary>
    /// 文章列表數據
    /// </summary>
    private ServiceResult<IEnumerable<QueryPostDto>> posts;

    /// <summary>
    /// 初始化
    /// </summary>
    protected override async Task OnInitializedAsync()
    {
        // TODO:獲取數據,可以在API中合併這兩個請求。
        var tag = await Http.GetFromJsonAsync<ServiceResult<string>>($"/blog/tag?name={name}");
        posts = await Http.GetFromJsonAsync<ServiceResult<IEnumerable<QueryPostDto>>>($"/blog/posts/tag?name={name}");

        if (tag.Success)
        {
            tagName = tag.Result;
        }
    }
}
@if (posts == null)
{
    <Loading />
}
else
{
    <div class="container">
        <div class="post-wrap tags">
            @if (tagName != null)
            {
                <h2 class="post-title">-&nbsp;Tag&nbsp;·&nbsp;@tagName&nbsp;-</h2>
            }
        </div>
        <div class="post-wrap archive">
            @if (posts.Success && posts.Result.Any())
            {
                @foreach (var item in posts.Result)
                {
                    <h3>@item.Year</h3>
                    @foreach (var post in item.Posts)
                    {
                        <article class="archive-item">
                            <NavLink href="@("/post"+post.Url)">@post.Title</NavLink>
                            <span class="archive-item-date">@post.CreationTime</span>
                        </article>
                    }
                }
            }
            else
            {
                <ErrorTip />
            }
        </div>
    </div>
}

以上完成了以上幾個頁面的數據綁定,頁面之間的跳轉已經關聯起來了,然後還剩下文章詳情頁,大家可以先自己動手完成它,今天就到這裏,未完待續…

開源地址:https://github.com/Meowv/Blog/tree/blog_tutorial

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

【其他文章推薦】

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

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

※超省錢租車方案

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

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

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

小師妹學JavaIO之:MappedByteBuffer多大的文件我都裝得下

目錄

  • 簡介
  • 虛擬地址空間
  • 詳解MappedByteBuffer
    • MapMode
  • MappedByteBuffer的最大值
  • MappedByteBuffer的使用
  • MappedByteBuffer要注意的事項
  • 總結

簡介

大大大,我要大!小師妹要讀取的文件越來越大,該怎麼幫幫她,讓程序在性能和速度上面得到平衡呢?快來跟F師兄一起看看吧。

虛擬地址空間

小師妹:F師兄,你有沒有發現,最近硬盤的價格真的是好便宜好便宜,1T的硬盤大概要500塊,平均1M五毛錢。現在下個電影都1G起步,這是不是意味着我們買入了大數據時代?

沒錯,小師妹,硬件技術的進步也帶來了軟件技術的進步,兩者相輔相成,缺一不可。

小師妹:F師兄,如果要是去讀取G級的文件,有沒有什麼快捷簡單的方法?

更多精彩內容且看:

  • 區塊鏈從入門到放棄系列教程-涵蓋密碼學,超級賬本,以太坊,Libra,比特幣等持續更新
  • Spring Boot 2.X系列教程:七天從無到有掌握Spring Boot-持續更新
  • Spring 5.X系列教程:滿足你對Spring5的一切想象-持續更新
  • java程序員從小工到專家成神之路(2020版)-持續更新中,附詳細文章教程

還記得上次我們講的虛擬地址空間嗎?

再把上次講的圖搬過來:

通常來說我們的應用程序調用系統的接口從磁盤空間獲取Buffer數據,我們把自己的應用程序稱之為用戶空間,把系統的底層稱之為系統空間。

傳統的IO操作,是操作系統講磁盤中的文件讀入到系統空間裏面,然後再拷貝到用戶空間中,供用戶使用。

這中間多了一個Buffer拷貝的過程,如果這個量夠大的話,其實還是挺浪費時間的。

於是有人在想了,拷貝太麻煩太耗時了,我們單獨劃出一塊內存區域,讓系統空間和用戶空間同時映射到同一塊地址不就省略了拷貝的步驟嗎?

這個被劃出來的單獨的內存區域叫做虛擬地址空間,而不同空間到虛擬地址的映射就叫做Buffer Map。 Java中是有一個專門的MappedByteBuffer來代表這種操作。

小師妹:F師兄,那這個虛擬地址空間和內存有什麼區別呢?有了內存還要啥虛擬地址空間?

虛擬地址空間有兩個好處。

第一個好處就是虛擬地址空間對於應用程序本身而言是獨立的,從而保證了程序的互相隔離和程序中地址的確定性。比如說一個程序如果運行在虛擬地址空間中,那麼它的空間地址是固定的,不管他運行多少次。如果直接使用內存地址,那麼可能這次運行的時候內存地址可用,下次運行的時候內存地址不可用,就會導致潛在的程序出錯。

第二個好處就是虛擬空間地址可以比真實的內存地址大,這個大其實是對內存的使用做了優化,比如說會把很少使用的內存寫如磁盤,從而釋放出更多的內存來做更有意義的事情,而之前存儲到磁盤的數據,當真正需要的時候,再從磁盤中加載到內存中。

這樣物理內存實際上可以看做虛擬空間地址的緩存。

詳解MappedByteBuffer

小師妹:MappedByteBuffer聽起來好神奇,怎麼使用它呢?

我們先來看看MappedByteBuffer的定義:

public abstract class MappedByteBuffer
    extends ByteBuffer

它實際上是一個抽象類,具體的實現有兩個:

class DirectByteBuffer extends MappedByteBuffer implements DirectBuffer
class DirectByteBufferR extends DirectByteBuffer
implements DirectBuffer

分別是DirectByteBuffer和DirectByteBufferR。

小師妹:F師兄,這兩個ByteBuffer有什麼區別呢?這個R是什麼意思?

R代表的是ReadOnly的意思,可能是因為本身是個類的名字就夠長了,所以搞了個縮寫。但是也不寫個註解,讓人看起來十分費解….

我們可以從RandomAccessFile的FilChannel中調用map方法獲得它的實例。

我們看下map方法的定義:

 public abstract MappedByteBuffer map(MapMode mode, long position, long size)
        throws IOException;

MapMode代表的是映射的模式,position表示是map開始的地址,size表示是ByteBuffer的大小。

MapMode

小師妹:F師兄,文件有隻讀,讀寫兩種模式,是不是MapMode也包含這兩類?

對的,其實NIO中的MapMode除了這兩個之外,還有一些其他很有趣的用法。

  • FileChannel.MapMode.READ_ONLY 表示只讀模式
  • FileChannel.MapMode.READ_WRITE 表示讀寫模式
  • FileChannel.MapMode.PRIVATE 表示copy-on-write模式,這個模式和READ_ONLY有點相似,它的操作是先對原數據進行拷貝,然後可以在拷貝之後的Buffer中進行讀寫。但是這個寫入並不會影響原數據。可以看做是數據的本地拷貝,所以叫做Private。

基本的MapMode就這三種了,其實除了基礎的MapMode,還有兩種擴展的MapMode:

  • ExtendedMapMode.READ_ONLY_SYNC 同步的讀
  • ExtendedMapMode.READ_WRITE_SYNC 同步的讀寫

MappedByteBuffer的最大值

小師妹:F師兄,既然可以映射到虛擬內存空間,那麼這個MappedByteBuffer是不是可以無限大?

當然不是了,首先虛擬地址空間的大小是有限制的,如果是32位的CPU,那麼一個指針佔用的地址就是4個字節,那麼能夠表示的最大值是0xFFFFFFFF,也就是4G。

另外我們看下map方法中size的類型是long,在java中long能夠表示的最大值是0x7fffffff,也就是2147483647字節,換算一下大概是2G。也就是說MappedByteBuffer的最大值是2G,一次最多只能map 2G的數據。

MappedByteBuffer的使用

小師妹,F師兄我們來舉兩個使用MappedByteBuffer讀寫的例子吧。

善!

先看一下怎麼使用MappedByteBuffer來讀數據:

public void readWithMap() throws IOException {
        try (RandomAccessFile file = new RandomAccessFile(new File("src/main/resources/big.www.flydean.com"), "r"))
        {
            //get Channel
            FileChannel fileChannel = file.getChannel();
            //get mappedByteBuffer from fileChannel
            MappedByteBuffer buffer = fileChannel.map(FileChannel.MapMode.READ_ONLY, 0, fileChannel.size());
            // check buffer
            log.info("is Loaded in physical memory: {}",buffer.isLoaded());  //只是一個提醒而不是guarantee
            log.info("capacity {}",buffer.capacity());
            //read the buffer
            for (int i = 0; i < buffer.limit(); i++)
            {
                log.info("get {}", buffer.get());
            }
        }
    }

然後再看一個使用MappedByteBuffer來寫數據的例子:

public void writeWithMap() throws IOException {
        try (RandomAccessFile file = new RandomAccessFile(new File("src/main/resources/big.www.flydean.com"), "rw"))
        {
            //get Channel
            FileChannel fileChannel = file.getChannel();
            //get mappedByteBuffer from fileChannel
            MappedByteBuffer buffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, 4096 * 8 );
            // check buffer
            log.info("is Loaded in physical memory: {}",buffer.isLoaded());  //只是一個提醒而不是guarantee
            log.info("capacity {}",buffer.capacity());
            //write the content
            buffer.put("www.flydean.com".getBytes());
        }
    }

MappedByteBuffer要注意的事項

小師妹:F師兄,MappedByteBuffer因為使用了內存映射,所以讀寫的速度都會有所提升。那麼我們在使用中應該注意哪些問題呢?

MappedByteBuffer是沒有close方法的,即使它的FileChannel被close了,MappedByteBuffer仍然處於打開狀態,只有JVM進行垃圾回收的時候才會被關閉。而這個時間是不確定的。

總結

本文再次介紹了虛擬地址空間和MappedByteBuffer的使用。

本文的例子https://github.com/ddean2009/learn-java-io-nio

本文作者:flydean程序那些事

本文鏈接:http://www.flydean.com/io-nio-mappedbytebuffer/

本文來源:flydean的博客

歡迎關注我的公眾號:程序那些事,更多精彩等着您!

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

【其他文章推薦】

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

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

※回頭車貨運收費標準

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

※超省錢租車方案

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

【深度思考】JDK8中日期類型該如何使用?

在JDK8之前,處理日期時間,我們主要使用3個類,DateSimpleDateFormatCalendar

這3個類在使用時都或多或少的存在一些問題,比如SimpleDateFormat不是線程安全的,

比如DateCalendar獲取到的月份是0到11,而不是現實生活中的1到12,關於這一點,《阿里巴巴Java開發手冊》中也有提及,因為很容易犯錯:

不過,JDK8推出了全新的日期時間處理類解決了這些問題,比如InstantLocalDateLocalTimeLocalDateTimeDateTimeFormatter,在《阿里巴巴Java開發手冊》中也推薦使用Instant

LocalDateTimeDateTimeFormatter

但我發現好多項目中其實並沒有使用這些類,使用的還是之前的DateSimpleDateFormatCalendar,所以本篇博客就講解下JDK8新推出的日期時間類,主要是下面幾個:

  1. Instant
  2. LocalDate
  3. LocalTime
  4. LocalDateTime
  5. DateTimeFormatter

1. Instant

1.1 獲取當前時間

既然Instant可以代替Date類,那它肯定可以獲取當前時間:

Instant instant = Instant.now();
System.out.println(instant);

輸出結果:

2020-06-10T08:22:13.759Z

細心的你會發現,這個時間比北京時間少了8個小時,如果要輸出北京時間,可以加上默認時區:

System.out.println(instant.atZone(ZoneId.systemDefault()));

輸出結果:

2020-06-10T16:22:13.759+08:00[Asia/Shanghai]

1.2 獲取時間戳

Instant instant = Instant.now();

// 當前時間戳:單位為秒
System.out.println(instant.getEpochSecond());
// 當前時間戳:單位為毫秒
System.out.println(instant.toEpochMilli());

輸出結果:

1591777752

1591777752613

當然,也可以通過System.currentTimeMillis()獲取當前毫秒數。

1.3 將long轉換為Instant

1)根據秒數時間戳轉換:

Instant instant = Instant.now();
System.out.println(instant);

long epochSecond = instant.getEpochSecond();
System.out.println(Instant.ofEpochSecond(epochSecond));
System.out.println(Instant.ofEpochSecond(epochSecond, instant.getNano()));

輸出結果:

2020-06-10T08:40:54.046Z

2020-06-10T08:40:54Z

2020-06-10T08:40:54.046Z

2)根據毫秒數時間戳轉換:

Instant instant = Instant.now();
System.out.println(instant);

long epochMilli = instant.toEpochMilli();
System.out.println(Instant.ofEpochMilli(epochMilli));

輸出結果:

2020-06-10T08:43:25.607Z

2020-06-10T08:43:25.607Z

1.4 將String轉換為Instant

String text = "2020-06-10T08:46:55.967Z";
Instant parseInstant = Instant.parse(text);
System.out.println("秒時間戳:" + parseInstant.getEpochSecond());
System.out.println("豪秒時間戳:" + parseInstant.toEpochMilli());
System.out.println("納秒:" + parseInstant.getNano());

輸出結果:

秒時間戳:1591778815

豪秒時間戳:1591778815967

納秒:967000000

如果字符串格式不對,比如修改成2020-06-10T08:46:55.967,就會拋出java.time.format.DateTimeParseException異常,如下圖所示:

2. LocalDate

2.1 獲取當前日期

使用LocalDate獲取當前日期非常簡單,如下所示:

LocalDate today = LocalDate.now();
System.out.println("today: " + today);

輸出結果:

today: 2020-06-10

不用任何格式化,輸出結果就非常友好,如果使用Date,輸出這樣的格式,還得配合SimpleDateFormat指定yyyy-MM-dd進行格式化,一不小心還會出個bug,比如去年年底很火的1個bug,我當時還是截了圖的:

這2個好友是2019/12/31關注我的,但我2020年1月2號查看時,卻显示成了2020/12/31,為啥呢?格式化日期時格式寫錯了,應該是yyyy/MM/dd,卻寫成了YYYY/MM/dd,剛好那周跨年,就显示成下一年,也就是2020年了,當時好幾個博主寫過文章解析原因,我這裏就不做過多解釋了。

划重點:都說到這了,給大家安利下我新註冊的公眾號「申城異鄉人」,歡迎大家關注,更多原創文章等着你哦,哈哈。

2.2 獲取年月日

LocalDate today = LocalDate.now();

int year = today.getYear();
int month = today.getMonthValue();
int day = today.getDayOfMonth();

System.out.println("year: " + year);
System.out.println("month: " + month);
System.out.println("day: " + day);

輸出結果:

year: 2020

month: 6

day: 10

獲取月份終於返回1到12了,不像java.util.Calendar獲取月份返回的是0到11,獲取完還得加1。

2.3 指定日期

LocalDate specifiedDate = LocalDate.of(2020, 6, 1);
System.out.println("specifiedDate: " + specifiedDate);

輸出結果:

specifiedDate: 2020-06-01

如果確定月份,推薦使用另一個重載方法,使用枚舉指定月份:

LocalDate specifiedDate = LocalDate.of(2020, Month.JUNE, 1);

2.4 比較日期是否相等

LocalDate localDate1 = LocalDate.now();
LocalDate localDate2 = LocalDate.of(2020, 6, 10);
if (localDate1.equals(localDate2)) {
    System.out.println("localDate1 equals localDate2");
}

輸出結果:

localDate1 equals localDate2

2.5 獲取日期是本周/本月/本年的第幾天

LocalDate today = LocalDate.now();

System.out.println("Today:" + today);
System.out.println("Today is:" + today.getDayOfWeek());
System.out.println("今天是本周的第" + today.getDayOfWeek().getValue() + "天");
System.out.println("今天是本月的第" + today.getDayOfMonth() + "天");
System.out.println("今天是本年的第" + today.getDayOfYear() + "天");

輸出結果:

Today:2020-06-11

Today is:THURSDAY

今天是本周的第4天

今天是本月的第11天

今天是本年的第163天

2.6 判斷是否為閏年

LocalDate today = LocalDate.now();

System.out.println(today.getYear() + " is leap year:" + today.isLeapYear());

輸出結果:

2020 is leap year:true

3. LocalTime

3.1 獲取時分秒

如果使用java.util.Date,那代碼是下面這樣的:

Date date = new Date();

int hour = date.getHours();
int minute = date.getMinutes();
int second = date.getSeconds();

System.out.println("hour: " + hour);
System.out.println("minute: " + minute);
System.out.println("second: " + second);

輸出結果:

注意事項:這幾個方法已經過期了,因此強烈不建議在項目中使用:

如果使用java.util.Calendar,那代碼是下面這樣的:

Calendar calendar = Calendar.getInstance();

// 12小時制
int hourOf12 = calendar.get(Calendar.HOUR);
// 24小時制
int hourOf24 = calendar.get(Calendar.HOUR_OF_DAY);
int minute = calendar.get(Calendar.MINUTE);
int second = calendar.get(Calendar.SECOND);
int milliSecond = calendar.get(Calendar.MILLISECOND);

System.out.println("hourOf12: " + hourOf12);
System.out.println("hourOf24: " + hourOf24);
System.out.println("minute: " + minute);
System.out.println("second: " + second);
System.out.println("milliSecond: " + milliSecond);

輸出結果:

注意事項:獲取小時時,有2個選項,1個返回12小時制的小時數,1個返回24小時制的小時數,因為現在是晚上8點,所以calendar.get(Calendar.HOUR)返回8,而calendar.get(Calendar.HOUR_OF_DAY)返回20。

如果使用java.time.LocalTime,那代碼是下面這樣的:

LocalTime localTime = LocalTime.now();
System.out.println("localTime:" + localTime);

int hour = localTime.getHour();
int minute = localTime.getMinute();
int second = localTime.getSecond();

System.out.println("hour: " + hour);
System.out.println("minute: " + minute);
System.out.println("second: " + second);

輸出結果:

可以看出,LocalTime只有時間沒有日期。

4. LocalDateTime

4.1 獲取當前時間

LocalDateTime localDateTime = LocalDateTime.now();
System.out.println("localDateTime:" + localDateTime);

輸出結果:

localDateTime: 2020-06-11T11:03:21.376

4.2 獲取年月日時分秒

LocalDateTime localDateTime = LocalDateTime.now();
System.out.println("localDateTime: " + localDateTime);

System.out.println("year: " + localDateTime.getYear());
System.out.println("month: " + localDateTime.getMonthValue());
System.out.println("day: " + localDateTime.getDayOfMonth());
System.out.println("hour: " + localDateTime.getHour());
System.out.println("minute: " + localDateTime.getMinute());
System.out.println("second: " + localDateTime.getSecond());

輸出結果:

4.3 增加天數/小時

LocalDateTime localDateTime = LocalDateTime.now();
System.out.println("localDateTime: " + localDateTime);

LocalDateTime tomorrow = localDateTime.plusDays(1);
System.out.println("tomorrow: " + tomorrow);

LocalDateTime nextHour = localDateTime.plusHours(1);
System.out.println("nextHour: " + nextHour);

輸出結果:

localDateTime: 2020-06-11T11:13:44.979

tomorrow: 2020-06-12T11:13:44.979

nextHour: 2020-06-11T12:13:44.979

LocalDateTime還提供了添加年、周、分鐘、秒這些方法,這裏就不一一列舉了:

4.4 減少天數/小時

LocalDateTime localDateTime = LocalDateTime.now();
System.out.println("localDateTime: " + localDateTime);

LocalDateTime yesterday = localDateTime.minusDays(1);
System.out.println("yesterday: " + yesterday);

LocalDateTime lastHour = localDateTime.minusHours(1);
System.out.println("lastHour: " + lastHour);

輸出結果:

localDateTime: 2020-06-11T11:20:38.896

yesterday: 2020-06-10T11:20:38.896

lastHour: 2020-06-11T10:20:38.896

類似的,LocalDateTime還提供了減少年、周、分鐘、秒這些方法,這裏就不一一列舉了:

4.5 獲取時間是本周/本年的第幾天

LocalDateTime localDateTime = LocalDateTime.now();
System.out.println("localDateTime: " + localDateTime);

System.out.println("DayOfWeek: " + localDateTime.getDayOfWeek().getValue());
System.out.println("DayOfYear: " + localDateTime.getDayOfYear());

輸出結果:

localDateTime: 2020-06-11T11:32:31.731

DayOfWeek: 4

DayOfYear: 163

5. DateTimeFormatter

JDK8中推出了java.time.format.DateTimeFormatter來處理日期格式化問題,《阿里巴巴Java開發手冊》中也是建議使用DateTimeFormatter代替SimpleDateFormat

5.1 格式化LocalDate

LocalDate localDate = LocalDate.now();

System.out.println("ISO_DATE: " + localDate.format(DateTimeFormatter.ISO_DATE));
System.out.println("BASIC_ISO_DATE: " + localDate.format(DateTimeFormatter.BASIC_ISO_DATE));
System.out.println("ISO_WEEK_DATE: " + localDate.format(DateTimeFormatter.ISO_WEEK_DATE));
System.out.println("ISO_ORDINAL_DATE: " + localDate.format(DateTimeFormatter.ISO_ORDINAL_DATE));

輸出結果:

如果提供的格式無法滿足你的需求,你還可以像以前一樣自定義格式:

LocalDate localDate = LocalDate.now();

System.out.println("yyyy/MM/dd: " + localDate.format(DateTimeFormatter.ofPattern("yyyy/MM/dd")));

輸出結果:

yyyy/MM/dd: 2020/06/11

5.2 格式化LocalTime

LocalTime localTime = LocalTime.now();
System.out.println(localTime);
System.out.println("ISO_TIME: " + localTime.format(DateTimeFormatter.ISO_TIME));
System.out.println("HH:mm:ss: " + localTime.format(DateTimeFormatter.ofPattern("HH:mm:ss")));

輸出結果:

14:28:35.230

ISO_TIME: 14:28:35.23

HH:mm:ss: 14:28:35

5.3 格式化LocalDateTime

LocalDateTime localDateTime = LocalDateTime.now();
System.out.println(localDateTime);
System.out.println("ISO_DATE_TIME: " + localDateTime.format(DateTimeFormatter.ISO_DATE_TIME));
System.out.println("ISO_DATE: " + localDateTime.format(DateTimeFormatter.ISO_DATE));

輸出結果:

2020-06-11T14:33:18.303

ISO_DATE_TIME: 2020-06-11T14:33:18.303

ISO_DATE: 2020-06-11

6. 類型相互轉換

6.1 Instant轉Date

JDK8中,Date新增了from()方法,將Instant轉換為Date,代碼如下所示:

Instant instant = Instant.now();
System.out.println(instant);

Date dateFromInstant = Date.from(instant);
System.out.println(dateFromInstant);

輸出結果:

2020-06-11T06:39:34.979Z

Thu Jun 11 14:39:34 CST 2020

6.2 Date轉Instant

JDK8中,Date新增了toInstant方法,將Date轉換為Instant,代碼如下所示:

Date date = new Date();
Instant dateToInstant = date.toInstant();
System.out.println(date);
System.out.println(dateToInstant);

輸出結果:

Thu Jun 11 14:46:12 CST 2020

2020-06-11T06:46:12.112Z

6.3 Date轉LocalDateTime

Date date = new Date();
Instant instant = date.toInstant();
LocalDateTime localDateTimeOfInstant = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
System.out.println(date);
System.out.println(localDateTimeOfInstant);

輸出結果:

Thu Jun 11 14:51:07 CST 2020

2020-06-11T14:51:07.904

6.4 Date轉LocalDate

Date date = new Date();
Instant instant = date.toInstant();
LocalDateTime localDateTimeOfInstant = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
LocalDate localDate = localDateTimeOfInstant.toLocalDate();
System.out.println(date);
System.out.println(localDate);

輸出結果:

Thu Jun 11 14:59:38 CST 2020

2020-06-11

可以看出,Date是先轉換為Instant,再轉換為LocalDateTime,然後通過LocalDateTime獲取LocalDate

6.5 Date轉LocalTime

Date date = new Date();
Instant instant = date.toInstant();
LocalDateTime localDateTimeOfInstant = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
LocalTime toLocalTime = localDateTimeOfInstant.toLocalTime();
System.out.println(date);
System.out.println(toLocalTime);

輸出結果:

Thu Jun 11 15:06:14 CST 2020

15:06:14.531

可以看出,Date是先轉換為Instant,再轉換為LocalDateTime,然後通過LocalDateTime獲取LocalTime

6.6 LocalDateTime轉Date

LocalDateTime localDateTime = LocalDateTime.now();

Instant toInstant = localDateTime.atZone(ZoneId.systemDefault()).toInstant();
Date dateFromInstant = Date.from(toInstant);
System.out.println(localDateTime);
System.out.println(dateFromInstant);

輸出結果:

2020-06-11T15:12:11.600

Thu Jun 11 15:12:11 CST 2020

6.7 LocalDate轉Date

LocalDate today = LocalDate.now();

LocalDateTime localDateTime = localDate.atStartOfDay();
Instant toInstant = localDateTime.atZone(ZoneId.systemDefault()).toInstant();
Date dateFromLocalDate = Date.from(toInstant);
System.out.println(dateFromLocalDate);

輸出結果:

Thu Jun 11 00:00:00 CST 2020

6.8 LocalTime轉Date

LocalDate localDate = LocalDate.now();
LocalTime localTime = LocalTime.now();

LocalDateTime localDateTime = LocalDateTime.of(localDate, localTime);
Instant instantFromLocalTime = localDateTime.atZone(ZoneId.systemDefault()).toInstant();
Date dateFromLocalTime = Date.from(instantFromLocalTime);

System.out.println(dateFromLocalTime);

輸出結果:

Thu Jun 11 15:24:18 CST 2020

7. 總結

JDK8推出了全新的日期時間類,如InstantLocaleDateLocalTimeLocalDateTimeDateTimeFormatter,設計比之前更合理,也是線程安全的。

《阿里巴巴Java開發規範》中也推薦使用Instant代替DateLocalDateTime 代替 CalendarDateTimeFormatter 代替 SimpleDateFormat

因此,如果條件允許,建議在項目中使用,沒有使用的,可以考慮升級下。

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

【其他文章推薦】

※超省錢租車方案

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

※回頭車貨運收費標準

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

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

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

看完這些黑科技,我覺得寶馬的真實身份其實是科技公司!

寶馬通過一台基於全新BMW 5系的原型車展示了未來高度自動駕駛汽車極為個性化、智能化的駕駛感受。在高度自動駕駛狀態下,駕駛者的雙手和雙腳都得到了解放,無需手握方向盤和控制油門、剎車。同時,BMW實時交通信號系統還可以預測前方下一組信號燈情況,讓駕駛者更好地進行選擇。

(拉斯維加斯/北京)2017年1月5日,一年一度的科技盛會、2017國際消費电子展在拉斯維加斯拉開帷幕。寶馬集團攜一系列前瞻理念和科技成果亮相,展現了其在智能互聯、未來汽車內部設計、控制與显示系統、自動駕駛等創新領域蓬勃的創造力。

寶馬集團正在成為引領数字出行生活的重要力量。在其願景中,自動駕駛技術將為駕駛者帶來更多選擇和自由;豐富的智能互聯服務圍繞人的需求,將車輛與用戶的数字生活無縫對接,讓出行和生活更高效、便捷、充滿樂趣。重要的是,這些離我們的生活並不遙遠,近年在CES上亮相的手勢控制、遠程3D環視影像等,均迅速應用於量產車型,體現了寶馬在行業內的領先地位。

這不是科幻大片,是你未來的汽車

在CES展台,寶馬集團通過BMW i Inside Future未來內室研究項目展示出,未來配備自動駕駛技術的汽車,其座艙將根據用戶需求,在休息室、辦公室和娛樂室之間實現自由切換。

未來,車內空間的氛圍和控制方式將取決於駕駛模式。在主動駕駛模式下,主動駕駛的功能將處於車內中心位置。在高度自動駕駛模式下,系統將显示更多的舒適、信息娛樂和通訊功能。導航系統可以推薦適合高度或完全自動駕駛的行車路線,並在到達路口時發出提醒。未來,自動駕駛將首先應用於高速或單向行駛道路。

對人機交互模式的創新是未來車輛的重要課題。寶馬在最近的兩屆CES上都帶來了創新的人機交互系統——手勢控制和AirTouch手勢控制系統。其中手勢控制已經應用於量產的新BMW 7系和全新BMW 5系車型。今年,寶馬又帶來了BMW HoloActive觸控系統,將人機交互體驗提升至新的高度。

BMW HoloActive觸控系統的原理與平視显示系統類似,通過反射原理在中控台位置投射出一塊“懸浮”屏幕,駕駛者通過指尖“點擊”虛擬屏幕來控制車輛,呈現出科幻大片的既視感。這套系統通過高敏感度的攝像頭捕捉駕駛者指尖的動作。在超聲波裝置的配合下,客戶的手指可以感受到輕微的壓力,模擬了傳統觸控屏的體驗,讓操作更符合人們的習慣。

未來的車輛內,駕駛者和乘客可以各自享受音樂而不互相干擾。首次展出的音效裝置BMW Sound Curtain通過座椅頭枕發射出不同的聲音信號,為座位上的用戶提供個性化的專屬娛樂信息。一個可摺疊的大尺寸屏幕可從車內頂篷延伸出來,進一步豐富後排乘客的車內互聯生活。

擁有一台高度互聯的自動駕駛汽車是一種怎樣的體驗?

寶馬通過一台基於全新BMW 5系的原型車展示了未來高度自動駕駛汽車極為個性化、智能化的駕駛感受。在高度自動駕駛狀態下,駕駛者的雙手和雙腳都得到了解放,無需手握方向盤和控制油門、剎車。同時,BMW實時交通信號系統還可以預測前方下一組信號燈情況,讓駕駛者更好地進行選擇。

在自動泊車功能展示中,抵達停車場時,車輛自動與泊車管理服務進行連接,显示屏會提示駕駛者可以使用預約的停車位。駕駛者與乘客下車后,車輛隨即啟動自動泊車功能。BMW雲端互聯將在車輛停好後向駕駛者推送提示信息。通過全新BMW 5系中首次配備的環視影像系統,用戶還可以通過BMW雲端互聯應用實時查看車輛情況。

在自動駕駛帶來的閑暇時間里,駕駛者可以盡情享受豐富的智能互聯功能。例如,在車輛行進途中,前排乘客可以通過BMW增強手勢控制系統、獲取途經場所的信息,如娛樂場所的節目單,還可以直接訂票。

在開放式移動雲的支持下,BMW 雲端互聯可以整合豐富的應用,讓人們在車內便捷地處理各種事務。比如已經在家用電腦上推出的個人数字助理“微軟小娜(Cortana)”也可以在BMW汽車上使用。在駕駛過程中,用戶可以通過語音控制讓小娜推薦就餐地點並預定位置。

通過與亞馬遜prime Now速遞服務合作,未來在行車途中預約收取快遞也成為可能。設想一下,用戶正前往一個生日聚會,但忘了購買禮物,通過這一功能,用戶可以從容地在車內在線購物。prime Now與開放式移動雲將根據車輛位置、路線和實時交通信息計算出最佳交付地點;車輛到達交付點之後,prime Now的員工將把貨物送到用戶的手中。BMW 雲端互聯無疑為智能、便捷的数字生活帶來了巨大的想象空間。BMW 雲端互聯已於2016年12月在中國正式上線,未來這些功能也將逐步升級到現有版本中。

BMW不僅在車庫,還可以在客廳——智能出行和智能生活相結合

在2017 CES上,寶馬集團還將全新的智能互聯科技從車內延伸到用戶的家中,標志著BMW 雲端互聯與家居環境的結合。未來,通過創新的智能終端BMW Conncted Window,用戶在家也可以享受BMW雲端互聯的豐富功能。

當用戶開始新的一天時,BMW Connected Window的界面將显示溫暖的問候,與此同時,用戶每天的出行日程會按照時間軸進行显示,出行目的地,建議出發時間、天氣情況等信息一目瞭然。BMW Connected Window的虛擬界面通過手勢來操作,與觸摸屏一樣直觀。更新信息也可以通過BMW 雲端互聯方便地添加到日程中並與其他智能設備保持同步,幫助用戶胸有成竹地開始新一天的生活。

本站聲明:網站內容來源於http://www.auto6s.com/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※帶您來了解什麼是 USB CONNECTOR  ?

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

※如何讓商品強力曝光呢? 網頁設計公司幫您建置最吸引人的網站,提高曝光率!

※綠能、環保無空污,成為電動車最新代名詞,目前市場使用率逐漸普及化

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

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