機器學習——打開集成方法的大門,手把手帶你實現AdaBoost模型

本文始發於個人公眾號:TechFlow,原創不易,求個關注

今天是機器學習專題的第25篇文章,我們一起來聊聊AdaBoost。

我們目前為止已經學過了好幾個模型,光決策樹的生成算法就有三種。但是我們每次進行分類的時候,每次都是採用一個模型進行訓練和預測。我們日常在做一個決策的時候,往往會諮詢好幾個人,綜合採納他們的意見。那麼有沒有可能把這個思路照搬到機器學習領域當中,創建多個模型來綜合得出結果呢?

這當然是可以的,這樣的思路就叫做集成方法(ensemble method)。

集成方法

集成方法本身並不是某種具體的方法或者是算法,只是一種訓練機器學習模型的思路。它的含義只有一點,就是訓練多個模型,然後將它們的結果匯聚在一起。

根據這個思路,業內又衍生出了三種特定的方法,分別是Bagging、Boosting和Stacking。

Bagging

Bagging是bootstrap aggregating的縮寫,我們從字面上很難理解它的含義。我們記住這個名字即可,在Bagging方法當中,我們會通過有放回隨機採樣的方式創建K個數據集。對於每一個數據集來說,可能有一些單個的樣本重複出現,也可能有一些樣本從沒有出現過,但整體而言,每個樣本出現的概率是相同的。

之後,我們用抽樣出來的K個數據集訓練K個模型,這裏的模型沒有做限制,我們可以使用任何機器學習方模型。K個模型自然會得到K個結果,那麼我們採取民主投票的方式對這K個模型進行聚合。

舉個例子說,假設K=25,在一個二分類問題當中。有10個模型預測結果是0,15個模型預測結果是1。那麼最終整個模型的預測結果就是1,相當於K個模型民主投票,每個模型投票權一樣。大名鼎鼎的隨機森林就是採取的這種方式。

Boosting

Boosting的思路和Bagging非常相似,它們對於樣本的採樣邏輯是一致的。不同的是,在Boosting當中,這K個模型並不是同時訓練的,而是串行訓練的。每一個模型在訓練的時候都會基於之前模型的結果,更加關注於被之前模型判斷錯誤的樣本。同樣,樣本也會有一個權值,錯誤判斷率越大的樣本擁有越大的權值。

並且每一個模型根據它能力的不同,會被賦予不同的權重,最後會對所有模型進行加權求和,而不是公平投票。由於這個機制,使得模型在訓練的時候的效率也有差異。因為Bagging所有模型之間是完全獨立的,我們是可以採取分佈式訓練的。而Boosting中每一個模型會依賴之前模型的效果,所以只能串行訓練。

Stacking

Stacking是Kaggle比賽當中經常使用的方法,它的思路也非常簡單。我們選擇K種不同的模型,然後通過交叉驗證的方式,在訓練集上進行訓練和預測。保證每個模型都對所有的訓練樣本產出一個預測結果。那麼對於每一條訓練樣本,我們都能得到K個結果。

之後,我們再創建一個第二層的模型,它的訓練特徵就是這K個結果。也就是說Stacking方法當中會用到多層模型的結構,最後一層模型的訓練特徵是上層模型預測的結果。由模型自己去訓練究竟哪一個模型的結果更值得採納,以及如何組合模型之間的特長。

我們今天介紹的AdaBoost顧名思義,是一個經典的Boosting算法。

模型思路

AdaBoost的核心思路是通過使用Boosting的方法,通過一些弱分類器構建出強分類器來。

強分類器我們都很好理解,就是性能很強的模型,那麼弱分類器應該怎麼理解呢?模型的強弱其實是相對於隨機結果來定義的,比隨機結果越好的模型,它的性能越強。從這點出發,弱分類器也就是只比隨機結果略強的分類器。我們的目的是通過設計樣本和模型的權重,使得可以做出最佳決策,將這些弱分類器的結果綜合出強分類器的效果來。

首先我們會給訓練樣本賦予一個權重,一開始的時候,每一條樣本的權重均相等。根據訓練樣本訓練出一個弱分類器並計算這個分類器的錯誤率。然後在同一個數據集上再次訓練弱分類器,在第二次的訓練當中,我們將會調整每個樣本的權重。其中正確的樣本權重會降低,錯誤的樣本權重會升高

同樣每一個分類器也會分配到一個權重值,權重越高說明它的話語權越大。這些是根據模型的錯誤率來計算的。錯誤率定義為:

這裏的D表示數據集表示分類錯誤的集合,它也就等於錯誤分類的樣本數除以總樣本數。

有了錯誤率之後,我們可以根據下面這個公式得到

得到了之後,我們利用它對樣本的權重進行更新,其中分類正確的權重更改為:

分類錯誤的樣本權重更改為:

這樣,我們所有的權重都更新完了,這也就完成了一輪迭代。AdaBoost會反覆進行迭代和調整權重,直到訓練錯誤率為0或者是弱分類器的數量達到閾值。

代碼實現

首先,我們來獲取數據,這裏我們選擇了sklearn數據集中的乳腺癌預測數據。和之前的例子一樣,我們可以直接import進來使用,非常方便:

import numpy as np
import pandas as pd
from sklearn.datasets import load_breast_cancer

breast = load_breast_cancer()
X, y = breast.data, breast.target
# reshape,將一維向量轉成二維
y = y.reshape((-1, 1))

接着,我們將數據拆分成訓練數據和測試數據,這個也是常規做法了,沒有難度:

from sklearn.model_selection import train_test_split

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=23)

在AdaBoost模型當中,我們選擇的弱分類器是決策樹的樹樁。所謂的樹樁就是樹深為1的決策樹。樹深為1顯然不論我們怎麼選擇閾值,都不會得到特別好的結果,但是由於我們依然會選擇閾值和特徵,所以結果也不會太差,至少要比隨機選擇要好。所以這就保證了,我們可以得到一個比隨機選擇效果略好一些的弱分類器,並且它的實現非常簡單。

在我們實現模型之前,我們先來實現幾個輔助函數。

def loss_error(y_pred, y, weight):
    return weight.T.dot((y_pred != y_train))

def stump_classify(X, idx, threshold, comparator):
    if comparator == 'lt':
        return X[:, idx] <= threshold
    else:
        return X[:, idx] > threshold
    
def get_thresholds(X, i):
    min_val, max_val = X[:, i].min(), X[:, i].max()
    return np.linspace(min_val, max_val, 10)

這三個函數應該都不難理解,第一個函數當中我們計算了模型的誤差。由於我們每一個樣本擁有一個自身的權重,所以我們對誤差進行加權求和。第二個函數是樹樁分類器的預測函數,邏輯非常簡單,根據閾值比較大小。這裡有兩種情況,有可能小於閾值的樣本是正例,也有可能大於閾值的樣本是正例,所以我們還需要第三個參數記錄這個信息。第三個函數是生成閾值的函數,由於我們並不需要樹樁的性能特別好,所以我們也沒有必要去遍歷閾值的所有取值,簡單地把特徵的範圍劃分成10段即可。

接下來是單個樹樁的生成函數,它等價於決策樹當中選擇特徵進行數據拆分的函數,邏輯大同小異,只需要稍作修改即可。

def build_stump(X, y, weight):
    m, n = X.shape
    ret_stump, ret_pred = None, []
    best_error = float('inf')

    # 枚舉特徵
    for i in range(n):
        # 枚舉閾值
        for j in get_thresholds(X, i):
            # 枚舉正例兩種情況
            for c in ['lt', 'gt']:
                # 預測並且求誤差
                pred = stump_classify(X, i, j, c).reshape((-1, 1))
                err = loss_error(pred, y, weight)
                # 記錄下最好的樹樁
                if err < best_error:
                    best_error, ret_pred = err, pred.copy()
                    ret_stump = {'idx': i, 'threshold': j, 'comparator': c} 
    return ret_stump, best_error, ret_pred

接下來要做的就是重複生成樹樁的操作,計算,並且更新每一條樣本的權重。整個過程也沒有太多的難點,基本上就是照着實現公式:

def adaboost_train(X, y, num_stump):
    stumps = []
    m = X.shape[0]
    # 樣本權重初始化,一開始全部相等
    weight = np.ones((y_train.shape[0], 1)) / y_train.shape[0]
    # 生成num_stump個樹樁
    for i in range(num_stump):
        best_stump, err, pred = build_stump(X, y, weight)
        # 計算alpha
        alpha = 0.5 * np.log((1.0 - err) / max(err, 1e-10))
        best_stump['alpha'] = alpha
        stumps.append(best_stump)

        # 更新每一條樣本的權重
        for j in range(m):
            weight[j] = weight[j] * (np.exp(-alpha) if pred[j] == y[j] else np.exp(alpha))
        weight = weight / weight.sum()
        # 如果當前的準確率已經非常高,則退出
        if err < 1e-8:
            break
    return stumps

樹樁生成結束之後,最後就是預測的部分了。整個預測過程依然非常簡單,就是一個加權求和的過程。這裏要注意一下,我們在訓練的時候為了突出錯誤預測的樣本,讓模型擁有更好的能力,維護了樣本的權重。然而在預測的時候,我們是不知道預測樣本的權重的,所以我們只需要對模型的結果進行加權即可。

def adaboost_classify(X, stumps):
    m = X.shape[0]
    pred = np.ones((m, 1))
    alphs = 0.0
    for i, stump in enumerate(stumps):
        y_pred = stump_classify(X, stump['idx'], stump['threshold'], stump['comparator'])
        # 根據alpha加權求和
        pred = y_pred * stump['alpha']
        alphs += stump['alpha']
    pred /= alphs
    # 根據0.5劃分0和1類別
    return np.sign(pred).reshape((-1, 1))

到這裏,我們整個模型就實現完了,我們先來看下單個樹樁在訓練集上的表現:

可以看到準確率只有0.54,只是比隨機預測略好一點點而已。

然而當我們綜合了20個樹樁的結果之後,在訓練集上我們可以得到0.9的準確率。在預測集上,它的表現更好,準確率有接近0.95!

這是因為AdaBoost當中,每一個分類器都是弱分類器,它根本沒有過擬合的能力,畢竟在訓練集的表現都很差,這就保證了分類器學到的都是實在的泛化能力,在訓練集上適用,在測試集上很大概率也適用。這也是集成方法最大的優點之一。

總結

集成方法可以說是機器學習領域一個非常重要的飛躍,集成方法的出現,讓設計出一個強分類器這件事的難度大大降低,並且還保證了模型的效果。

因為在一些領域當中,設計一個強分類器可能非常困難,然而設計一個弱一些的分類器則簡單得多,再加上模型本身性能很好,不容易陷入過擬合。使得在深度學習模型流行之前,集成方法廣泛使用,幾乎所有機器學習領域的比賽的冠軍,都使用了集成學習。

集成學習當中具體的思想或許各有不同,但是核心的思路是一致的。我們理解了AdaBoost之後,再去學習其他的集成模型就要容易多了。

如果喜歡本文,可以的話,請點個關注,給我一點鼓勵,也方便獲取更多文章。

本文使用 mdnice 排版

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

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

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

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

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

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

※超省錢租車方案

聚甘新

從linux源碼看epoll

從linux源碼看epoll

前言

在linux的高性能網絡編程中,繞不開的就是epoll。和select、poll等系統調用相比,epoll在需要監視大量文件描述符並且其中只有少數活躍的時候,表現出無可比擬的優勢。epoll能讓內核記住所關注的描述符,並在對應的描述符事件就緒的時候,在epoll的就緒鏈表中添加這些就緒元素,並喚醒對應的epoll等待進程。
本文就是筆者在探究epoll源碼過程中,對kernel將就緒描述符添加到epoll並喚醒對應進程的一次源碼分析(基於linux-2.6.32內核版本)。由於篇幅所限,筆者聚焦於tcp協議下socket可讀事件的源碼分析。

簡單的epoll例子

下面的例子,是從筆者本人用c語言寫的dbproxy中的一段代碼。由於細節過多,所以做了一些刪減。

int init_reactor(int listen_fd,int worker_count){
	......
	// 創建多個epoll fd,以充分利用多核
	for(i=0;i<worker_count;i++){
		reactor->worker_fd = epoll_create(EPOLL_MAX_EVENTS);
	}
	/* epoll add listen_fd and accept */
	// 將accept后的事件加入到對應的epoll fd中
	int client_fd = accept(listen_fd,(struct sockaddr *)&client_addr,&client_len)));
	// 將連接描述符註冊到對應的worker裏面
	epoll_ctl(reactor->client_fd,EPOLL_CTL_ADD,epifd,&event);
}
// reactor的worker線程
static void* rw_thread_func(void* arg){
	......

	for(;;){
		  // epoll_wait等待事件觸發
        int retval = epoll_wait(epfd,events,EPOLL_MAX_EVENTS,500);
        if(retval > 0){
        	for(j=0; j < retval; j++){
        		// 處理讀事件
        	   if(event & EPOLLIN){
                 handle_ready_read_connection(conn);
                 continue;
             }
             /* 處理其它事件 */
        	}
        }
	}
	......
}

上述代碼事實上就是實現了一個reactor模式中的accept與read/write處理線程,如下圖所示:

epoll_create

Unix的萬物皆文件的思想在epoll裏面也有體現,epoll_create調用返回一個文件描述符,此描述符掛載在anon_inode_fs(匿名inode文件系統)的根目錄下面。讓我們看下具體的epoll_create系統調用源碼:

SYSCALL_DEFINE1(epoll_create, int, size)
{
	if (size <= 0)
		return -EINVAL;

	return sys_epoll_create1(0);
}

由上述源碼可見,epoll_create的參數是基本沒有意義的,kernel簡單的判斷是否為0,然後就直接就調用了sys_epoll_create1。由於linux的系統調用是通過(SYSCALL_DEFINE1,SYSCALL_DEFINE2……SYSCALL_DEFINE6)定義的,那麼sys_epoll_create1對應的源碼即是SYSCALL_DEFINE(epoll_create1)。
(注:受限於寄存器數量的限制,(80×86下的)kernel限制系統調用最多有6個參數。據ulk3所述,這是由於32位80×86寄存器的限制)
接下來,我們就看下epoll_create1的源碼:

SYSCALL_DEFINE1(epoll_create1, int, flags)
{
	// kzalloc(sizeof(*ep), GFP_KERNEL),用的是內核空間
	error = ep_alloc(&ep);
	// 獲取尚未被使用的文件描述符,即描述符數組的槽位
	fd = get_unused_fd_flags(O_RDWR | (flags & O_CLOEXEC));
	// 在匿名inode文件系統中分配一個inode,並得到其file結構體
	// 且file->f_op = &eventpoll_fops
	// 且file->private_data = ep;
	file = anon_inode_getfile("[eventpoll]", &eventpoll_fops, ep,
				 O_RDWR | (flags & O_CLOEXEC));
	// 將file填入到對應的文件描述符數組的槽裏面
	fd_install(fd,file);			 
	ep->file = file;
	return fd;
}

最後epoll_create生成的文件描述符如下圖所示:

struct eventpoll

所有的epoll系統調用都是圍繞eventpoll結構體做操作,現簡要描述下其中的成員:

/*
 * 此結構體存儲在file->private_data中
 */
struct eventpoll {
	// 自旋鎖,在kernel內部用自旋鎖加鎖,就可以同時多線(進)程對此結構體進行操作
	// 主要是保護ready_list
	spinlock_t lock;
	// 這個互斥鎖是為了保證在eventloop使用對應的文件描述符的時候,文件描述符不會被移除掉
	struct mutex mtx;
	// epoll_wait使用的等待隊列,和進程喚醒有關
	wait_queue_head_t wq;
	// file->poll使用的等待隊列,和進程喚醒有關
	wait_queue_head_t poll_wait;
	// 就緒的描述符隊列
	struct list_head rdllist;
	// 通過紅黑樹來組織當前epoll關注的文件描述符
	struct rb_root rbr;
	// 在向用戶空間傳輸就緒事件的時候,將同時發生事件的文件描述符鏈入到這個鏈表裡面
	struct epitem *ovflist;
	// 對應的user
	struct user_struct *user;
	// 對應的文件描述符
	struct file *file;
	// 下面兩個是用於環路檢測的優化
	int visited;
	struct list_head visited_list_link;
};

本文講述的是kernel是如何將就緒事件傳遞給epoll並喚醒對應進程上,因此在這裏主要聚焦於(wait_queue_head_t wq)等成員。

epoll_ctl(add)

我們看下epoll_ctl(EPOLL_CTL_ADD)是如何將對應的文件描述符插入到eventpoll中的。
藉助於spin_lock(自旋鎖)和mutex(互斥鎖),epoll_ctl調用可以在多個KSE(內核調度實體,即進程/線程)中併發執行。

SYSCALL_DEFINE4(epoll_ctl, int, epfd, int, op, int, fd,
		struct epoll_event __user *, event)
{
	/* 校驗epfd是否是epoll的描述符 */
	// 此處的互斥鎖是為了防止併發調用epoll_ctl,即保護內部數據結構
	// 不會被併發的添加修改刪除破壞
	mutex_lock_nested(&ep->mtx, 0);
	switch (op) {
		case EPOLL_CTL_ADD:
			...
			// 插入到紅黑樹中
			error = ep_insert(ep, &epds, tfile, fd);
			...
			break;
		......
	}
	mutex_unlock(&ep->mtx);	
}		

上述過程如下圖所示:

ep_insert

在ep_insert中初始化了epitem,然後初始化了本文關注的焦點,即事件就緒時候的回調函數,代碼如下所示:

static int ep_insert(struct eventpoll *ep, struct epoll_event *event,
		     struct file *tfile, int fd)
{
	/* 初始化epitem */
	// &epq.pt->qproc = ep_ptable_queue_proc
	init_poll_funcptr(&epq.pt, ep_ptable_queue_proc);
	// 在這裏將回調函數注入
	revents = tfile->f_op->poll(tfile, &epq.pt);
	// 如果當前有事件已經就緒,那麼一開始就會被加入到ready list
	// 例如可寫事件
	// 另外,在tcp內部ack之後調用tcp_check_space,最終調用sock_def_write_space來喚醒對應的epoll_wait下的進程
	if ((revents & event->events) && !ep_is_linked(&epi->rdllink)) {
		list_add_tail(&epi->rdllink, &ep->rdllist);
		// wake_up ep對應在epoll_wait下的進程
		if (waitqueue_active(&ep->wq)){
			wake_up_locked(&ep->wq);
		}
		......
	}	
	// 將epitem插入紅黑樹
	ep_rbtree_insert(ep, epi);
	......
}

tfile->f_op->poll的實現

向kernel更底層註冊回調函數的是tfile->f_op->poll(tfile, &epq.pt)這一句,我們來看一下對於對應的socket文件描述符,其fd=>file->f_op->poll的初始化過程:

    // 將accept后的事件加入到對應的epoll fd中
    int client_fd = accept(listen_fd,(struct sockaddr *)&client_addr,&client_len)));
    // 將連接描述符註冊到對應的worker裏面
    epoll_ctl(reactor->client_fd,EPOLL_CTL_ADD,epifd,&event);

回顧一下上述user space代碼,fd即client_fd是由tcp的listen_fd通過accept調用而來,那麼我們看下accept調用鏈的關鍵路徑:

accept
      |->accept4
            |->sock_attach_fd(newsock, newfile, flags & O_NONBLOCK);
                  |->init_file(file,...,&socket_file_ops);
                        |->file->f_op = fop;
                              /* file->f_op = &socket_file_ops */
            |->fd_install(newfd, newfile); // 安裝fd

那麼,由accept獲得的client_fd的結構如下圖所示:

(注:由於是tcp socket,所以這邊sock->ops=inet_stream_ops,這個初始化的過程在我的另一篇博客<<從linux源碼看socket的阻塞和非阻塞>>中,博客地址如下:
https://my.oschina.net/alchemystar/blog/1791017)
既然知道了tfile->f_op->poll的實現,我們就可以看下此poll是如何將安裝回調函數的。

回調函數的安裝

kernel的調用路徑如下:

sock_poll /*tfile->f_op->poll(tfile, &epq.pt)*/;
	|->sock->ops->poll
		|->tcp_poll
			/* 這邊重要的是拿到了sk_sleep用於KSE(進程/線程)的喚醒 */
			|->sock_poll_wait(file, sk->sk_sleep, wait);
				|->poll_wait
					|->p->qproc(filp, wait_address, p);
					/* p為&epq.pt,而且&epq.pt->qproc= ep_ptable_queue_proc*/
						|-> ep_ptable_queue_proc(filp,wait_address,p);

繞了一大圈之後,我們的回調函數的安裝其實就是調用了eventpoll.c中的ep_ptable_queue_proc,而且向其中傳遞了sk->sk_sleep作為其waitqueue的head,其源碼如下所示:

static void ep_ptable_queue_proc(struct file *file, wait_queue_head_t *whead,
				 poll_table *pt)
{
	// 取出當前client_fd對應的epitem
	struct epitem *epi = ep_item_from_epqueue(pt);
	// &pwq->wait->func=ep_poll_callback,用於回調喚醒
	// 注意,這邊不是init_waitqueue_entry,即沒有將當前KSE(current,當前進程/線程)寫入到
	// wait_queue當中,因為不一定是從當前安裝的KSE喚醒,而應該是喚醒epoll\_wait的KSE
	init_waitqueue_func_entry(&pwq->wait, ep_poll_callback);
	// 這邊的whead是sk->sk_sleep,將當前的waitqueue鏈入到socket對應的sleep列表
	add_wait_queue(whead, &pwq->wait);	
}	

這樣client_fd的結構進一步完善,如下圖所示:

ep_poll_callback函數是喚醒對應epoll_wait的地方,我們將在後面一起講述。

epoll_wait

epoll_wait主要是調用了ep_poll:

SYSCALL_DEFINE4(epoll_wait, int, epfd, struct epoll_event __user *, events,
		int, maxevents, int, timeout)
{
	/* 檢查epfd是否是epoll\_create創建的fd */
	// 調用ep_poll
	error = ep_poll(ep, events, maxevents, timeout);
	...
}

緊接着,我們看下ep_poll函數:

static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,
		   int maxevents, long timeout)
{
	......
retry:
	// 獲取spinlock
	spin_lock_irqsave(&ep->lock, flags);
	// 將當前task_struct寫入到waitqueue中以便喚醒
	// wq_entry->func = default_wake_function;
	init_waitqueue_entry(&wait, current);
	// WQ_FLAG_EXCLUSIVE,排他性喚醒,配合SO_REUSEPORT從而解決accept驚群問題
	wait.flags |= WQ_FLAG_EXCLUSIVE;
	// 鏈入到ep的waitqueue中
	__add_wait_queue(&ep->wq, &wait);
	for (;;) {
		// 設置當前進程狀態為可打斷
		set_current_state(TASK_INTERRUPTIBLE);
		// 檢查當前線程是否有信號要處理,有則返回-EINTR
		if (signal_pending(current)) {
			res = -EINTR;
			break;
		}
		spin_unlock_irqrestore(&ep->lock, flags);
		// schedule調度,讓出CPU
		jtimeout = schedule_timeout(jtimeout);
		spin_lock_irqsave(&ep->lock, flags);
	}
	// 到這裏,表明超時或者有事件觸發等動作導致進程重新調度
	__remove_wait_queue(&ep->wq, &wait);
	// 設置進程狀態為running
	set_current_state(TASK_RUNNING);
	......
	// 檢查是否有可用事件
	eavail = !list_empty(&ep->rdllist) || ep->ovflist != EP_UNACTIVE_PTR;
	......
	// 向用戶空間拷貝就緒事件
	ep_send_events(ep, events, maxevents)
}		   

上述邏輯如下圖所示:

ep_send_events

ep_send_events函數主要就是調用了ep_scan_ready_list,顧名思義ep_scan_ready_list就是掃描就緒列表:

static int ep_scan_ready_list(struct eventpoll *ep,
			      int (*sproc)(struct eventpoll *,
					   struct list_head *, void *),
			      void *priv,
			      int depth)
{
	...
	// 將epfd的rdllist鏈入到txlist
	list_splice_init(&ep->rdllist, &txlist);
	...
	/* sproc = ep_send_events_proc */
	error = (*sproc)(ep, &txlist, priv);
	...
	// 處理ovflist,即在上面sproc過程中又到來的事件
	...
}

其主要調用了ep_send_events_proc:

static int ep_send_events_proc(struct eventpoll *ep, struct list_head *head,
			       void *priv)
{
	for (eventcnt = 0, uevent = esed->events;
	     !list_empty(head) && eventcnt < esed->maxevents;) {
	   // 遍歷ready list 
		epi = list_first_entry(head, struct epitem, rdllink);
		list_del_init(&epi->rdllink);
		// readylist只是表明當前epi有事件,具體的事件信息還是得調用對應file的poll
		// 這邊的poll即是tcp_poll,根據tcp本身的信息設置掩碼(mask)等信息 & 上興趣事件掩碼,則可以得知當前事件是否是epoll_wait感興趣的事件
		revents = epi->ffd.file->f_op->poll(epi->ffd.file, NULL) &
			epi->event.events;
		if(revents){
			/* 將event放入到用戶空間 */
			/* 處理ONESHOT邏輯 */
			// 如果不是邊緣觸發,則將當前的epi重新加回到可用列表中,這樣就可以下一次繼續觸發poll,如果下一次poll的revents不為0,那麼用戶空間依舊能感知 */
			else if (!(epi->event.events & EPOLLET)){
				list_add_tail(&epi->rdllink, &ep->rdllist);
			}
			/* 如果是邊緣觸發,那麼就不加回可用列表,因此只能等到下一個可用事件觸發的時候才會將對應的epi放到可用列表裡面*/
			eventcnt++
		}
		/* 如poll出來的revents事件epoll_wait不感興趣(或者本來就沒有事件),那麼也不會加回到可用列表 */
		......
	}
	return eventcnt;
}			    

上述代碼邏輯如下所示:

事件到來添加到epoll就緒隊列(rdllist)的過程

經過上述章節的詳述之後,我們終於可以闡述,tcp在數據到來時是怎麼加入到epoll的就緒隊列的了。

可讀事件到來

首先我們看下tcp數據包從網卡驅動到kernel內部tcp協議處理調用鏈:

step1:

網絡分組到來的內核路徑,網卡發起中斷後調用netif_rx將事件掛入CPU的等待隊列,並喚起軟中斷(soft_irq),再通過linux的軟中斷機制調用net_rx_action,如下圖所示:

注:上圖來自PLKA(<<深入Linux內核架構>>)

step2:

緊接着跟蹤next_rx_action

next_rx_action
	|-process_backlog
		......
			|->packet_type->func 在這裏我們考慮ip_rcv
					|->ipprot->handler 在這裏ipprot重載為tcp_protocol
						(handler 即為tcp_v4_rcv)					

我們再看下對應的tcp_v4_rcv

tcp_v4_rcv
      |->tcp_v4_do_rcv
            |->tcp_rcv_state_process
                  |->tcp_data_queue
                        |-> sk->sk_data_ready(sock_def_readable)
                              |->wake_up_interruptible_sync_poll(sk->sleep,...)
                                    |->__wake_up
                                          |->__wake_up_common
                                                |->curr->func
                                                /* 這裏已經被ep_insert添加為ep_poll_callback,而且設定了排它標識WQ_FLAG_EXCLUSIVE*/
                                                      |->ep_poll_callback

這樣,我們就看下最終喚醒epoll_wait的ep_poll_callback函數:

static int ep_poll_callback(wait_queue_t *wait, unsigned mode, int sync, void *key)
{
	// 獲取wait對應的epitem	
	struct epitem *epi = ep_item_from_wait(wait);
	// epitem對應的eventpoll結構體
	struct eventpoll *ep = epi->ep;
	// 獲取自旋鎖,保護ready_list等結構
	spin_lock_irqsave(&ep->lock, flags);
	// 如果當前epi沒有被鏈入ep的ready list,則鏈入
	// 這樣,就把當前的可用事件加入到epoll的可用列表了
	if (!ep_is_linked(&epi->rdllink))
		list_add_tail(&epi->rdllink, &ep->rdllist);
	// 如果有epoll_wait在等待的話,則喚醒這個epoll_wait進程
	// 對應的&ep->wq是在epoll_wait調用的時候通過init_waitqueue_entry(&wait, current)而生成的
	// 其中的current即是對應調用epoll_wait的進程信息task_struct
	if (waitqueue_active(&ep->wq))
		wake_up_locked(&ep->wq);
}

上述過程如下圖所示:

最後wake_up_locked調用__wake_up_common,然後調用了在init_waitqueue_entry註冊的default_wake_function,調用路徑為:

wake_up_locked
	|->__wake_up_common
		|->default_wake_function
			|->try_wake_up (wake up a thread)
				|->activate_task
					|->enqueue_task    running

將epoll_wait進程推入可運行隊列,等待內核重新調度進程,然後epoll_wait對應的這個進程重新運行后,就從schedule恢復,繼續下面的ep_send_events(向用戶空間拷貝事件並返回)。
wake_up過程如下圖所示:

可寫事件到來

可寫事件的運行過程和可讀事件大同小異:
首先,在epoll_ctl_add的時候預先會調用一次對應文件描述符的poll,如果返回事件里有可寫掩碼的時候直接調用wake_up_locked以喚醒對應的epoll_wait進程。
然後,在tcp在底層驅動有數據到來的時候可能攜帶了ack從而可以釋放部分已經被對端接收的數據,於是觸發可寫事件,這一部分的調用鏈為:

tcp_input.c
tcp_v4_rcv
	|-tcp_v4_do_rcv
		|-tcp_rcv_state_process
			|-tcp_data_snd_check
				|->tcp_check_space
					|->tcp_new_space
						|->sk->sk_write_space
						/* tcp下即是sk_stream_write_space*/

最後在此函數裏面sk_stream_write_space喚醒對應的epoll_wait進程

void sk_stream_write_space(struct sock *sk)
{
	// 即有1/3可寫空間的時候才觸發可寫事件
	if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk) && sock) {
		clear_bit(SOCK_NOSPACE, &sock->flags);

		if (sk->sk_sleep && waitqueue_active(sk->sk_sleep))
			wake_up_interruptible_poll(sk->sk_sleep, POLLOUT |
						POLLWRNORM | POLLWRBAND)
		......
	}
}

關閉描述符(close fd)

值得注意的是,我們在close對應的文件描述符的時候,會自動調用eventpoll_release將對應的file從其關聯的epoll_fd中刪除,kernel關鍵路徑如下:

close fd
      |->filp_close
            |->fput
                  |->__fput
                        |->eventpoll_release
                              |->ep_remove

所以我們在關閉對應的文件描述符后,並不需要通過epoll_ctl_del來刪掉對應epoll中相應的描述符。

總結

epoll作為linux下非常優秀的事件觸發機製得到了廣泛的運用。其源碼還是比較複雜的,本文只是闡述了epoll讀寫事件的觸發機制,探究linux kernel源碼的過程非常快樂_

公眾號

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

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

【其他文章推薦】

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

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

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

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

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

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

聚甘新

html/css 滾動到元素位置,显示加載動畫

每次滾動到元素時,都显示加載動畫,如何添加?

 

元素添加初始參數

以上圖中的動畫為例,添加倆個左右容器,將內容放置在容器內部。

添加初始數據,默認透明度0、左右分別移動100px。

 1   //左側容器
 2   .item-leftContainer {
 3     opacity: 0;
 4     transform: translateX(-100px);
 5   }
 6   //右側容器
 7   .item-rightContainer {
 8     opacity: 0;
 9     transform: translateX(100px);
10   }

添加動畫數據

在less中添加動畫數據。這裏只設置了to,也可以省略第1步中的初始參數設置而在動畫里設置from。

執行后,透明度由0到1,倆個容器向中間移動100px回到原處。

 1   //動畫
 2   @keyframes showLeft {
 3     to {
 4       opacity: 1;
 5       transform: translateX(0px);
 6     }
 7   }
 8   @keyframes showRight {
 9     to {
10       opacity: 1;
11       transform: translateX(0px);
12     }
13   }
14   @keyframes hideLeft {
15     to {
16       opacity: 0;
17       transform: translateX(-100px);
18     }
19   }
20   @keyframes hideRight {
21     to {
22       opacity: 0;
23       transform: translateX(100px);
24     }
25   }

觸發動畫

頁面加載/刷新觸發 – 在componentDidMount中執行

頁面滾動時觸發 – 在componentDidMount、componentWillUnmount添加監聽/註銷頁面滾動的事件

校驗當前滾動高度與元素的位置差異:

window.pageYOffset(滾動距離) + windowHeight(窗口高度) > leftElement.offsetTop (元素的相對位置)+ parentOffsetTop(父元素的相對位置) + 200

  1. 真正的滾動視覺位置 – window.pageYOffset(滾動距離) + windowHeight(窗口高度)
  2. 元素距離頂部的高度 – 這裏使用了leftElement.offsetTop + parentOffsetTop,原因是父容器使用了absolute絕對定位。如果是正常布局的話,使用元素當前的位置leftElement.offsetTop即可
  3. 額外添加200高度,是為了優化視覺體驗。當超出200高度時才觸發動畫

當頁面滾動到下方,觸發显示動畫;當頁面重新滾動到上方,觸發隱藏動畫。

 1     componentDidMount() {
 2         this.checkScrollHeightAndLoadAnimation();
 3         window.addEventListener('scroll', this.bindHandleScroll);
 4     }
 5     componentWillUnmount() {
 6         window.removeEventListener('scroll', this.bindHandleScroll);
 7     }
 8     bindHandleScroll = (event) => {
 9         this.checkScrollHeightAndLoadAnimation();
10     }
11     checkScrollHeightAndLoadAnimation() {
12         const windowHeight = window.innerHeight;
13         let parentEelement = document.getElementById("softwareUsingWays-container") as HTMLElement;
14         const parentOffsetTop = parentEelement.offsetTop;
15         let leftElement = (parentEelement.getElementsByClassName("item-leftContainer") as HTMLCollectionOf<HTMLElement>)[0];
16         if (window.pageYOffset + windowHeight > leftElement.offsetTop + parentOffsetTop + 200) {
17             leftElement.style.animation = "showLeft .6s forwards" //添加動畫  
18         } else {
19             leftElement.style.animation = "hideLeft 0s forwards" //隱藏動畫 
20         }
21         let rightElement = (parentEelement.getElementsByClassName(".item-rightContainer") as HTMLCollectionOf<HTMLElement>)[0];
22         if (window.pageYOffset + windowHeight > rightElement.offsetTop + parentOffsetTop + 200) {
23             rightElement.style.animation = "showRight .6s forwards" //添加動畫  
24         } else {
25             rightElement.style.animation = "hideRight 0s forwards" //隱藏動畫 
26         }
27     }

 

關鍵字:React 滾動、加載/出現動畫

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

【其他文章推薦】

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

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

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

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

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

聚甘新

六、線程池(一)

線程池

通過建立池可以有效的利用系統資源,節約系統性能。Java 中的線程池就是一種非常好的實現,從 JDK1.5 開始 Java 提供了一個線程工廠 Executors 用來生成線程池,通過 Executors 可以方便的生成不同類型的線程池。

線程池的優點

  • 降低資源消耗。線程的開啟和銷毀會消耗資源,通過重複利用已創建的線程降低線程創建和銷毀造成的消耗。
  • 提高響應速度。當任務到達時,任務可以不需要的等到線程創建就能立即執行。
  • 提高線程的可管理性。線程是稀缺資源,如果無限制的創建,不僅會消耗系統資源,還會降低系統的穩定性,使用線程池可以進行統一的分配,調優和監控。

常見的線程池

  • CachedThreadPool:可緩存的線程池,該線程池中沒有核心線程,非核心線程的數量為 Integer.max_value,就是無限大,當有需要時創建線程來執行任務,沒有需要時回收線程,適用於耗時少,任務量大的情況。
  • SecudleThreadPool:周期性執行任務的線程池,按照某種特定的計劃執行線程中的任務,有核心線程,但也有非核心線程,非核心線程的大小也為無限大。適用於執行周期性的任務。
  • SingleThreadPool:只有一條線程來執行任務,適用於有順序的任務的應用場景。
  • FixedThreadPool:定長的線程池,有核心線程,核心線程的即為最大的線程數量,沒有非核心線程
  • Executors.newFixedThreadPool()、Executors.newSingleThreadExecutor() 和 Executors.newCachedThreadPool() 等方法的底層都是通過 ThreadPoolExecutor 實現的。

ThreadPoolExecutor

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        // maximumPoolSize 必須大於 0,且必須大於 corePoolSize
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.acc = System.getSecurityManager() == null ?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

參數介紹:

  • corePoolSize

    • 線程池的核心線程數。在沒有設置 allowCoreThreadTimeOut 為 true 的情況下,核心線程會在線程池中一直存活,即使處於閑置狀態。
    • 如果設置為 0,則表示在沒有任何任務時,銷毀線程池;如果大於 0,即使沒有任務時也會保證線程池的線程數量等於此值。
    • 但需要注意,此值如果設置的比較小,則會頻繁的創建和銷毀線程,如果設置的比較大,則會浪費系統資源,所以需要根據自己的實際業務來調整此值。
  • maximumPoolSize

    • 線程池所能容納的最大線程數。當活動線程(核心線程+非核心線程)達到這個數值后,後續任務將會根據 RejectedExecutionHandler 來進行拒絕策略處理。
    • 官方規定此值必須大於 0,也必須大於等於 corePoolSize,此值只有在任務比較多,且不能存放在任務隊列時,才會用到。
  • keepAliveTime

    • 非核心線程閑置時的超時時長。超過該時長,非核心線程就會被回收。
    • 若線程池通過 allowCoreThreadTimeOut() 方法設置 allowCoreThreadTimeOut 屬性為 true,則該時長同樣會作用於核心線程,AsyncTask 配置的線程池就是這樣設置的。
  • unit

    • keepAliveTime 時長對應的單位。
  • workQueue

    • 表示線程池執行的任務隊列,當線程池的所有線程都在處理任務時,如果來了新任務就會緩存到此任務隊列中排隊等待執行。
    • 是一個阻塞隊列 BlockingQueue,雖然它是 Queue 的子接口,但是它的主要作用並不是容器,而是作為線程同步的工具,他有一個特徵,當生產者試圖向 BlockingQueue 放入(put)元素,如果隊列已滿,則該線程被阻塞;當消費者試圖從 BlockingQueue 取出(take)元素,如果隊列已空,則該線程被阻塞。
  • ThreadFactory

    • 線程的創建工廠,功能很簡單,就是為線程池提供創建新線程的功能。
    • 也可以自定義一個線程工廠,通過實現 ThreadFactory 接口來完成,這樣就可以自定義線程的名稱或線程執行的優先級了。
    • 通常在創建線程池時不指定此參數,它會使用默認的線程創建工廠的方法來創建線程,源代碼如下:
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        // Executors.defaultThreadFactory() 為默認的線程創建工廠
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }
    public static ThreadFactory defaultThreadFactory() {
        return new DefaultThreadFactory();
    }
    // 默認的線程創建工廠,需要實現 ThreadFactory 接口
    static class DefaultThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;
    
        DefaultThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                                  Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }
        // 創建線程
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                                  namePrefix + threadNumber.getAndIncrement(),
                                  0);
            if (t.isDaemon()) 
                t.setDaemon(false); // 創建一個非守護線程
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY); // 線程優先級設置為默認值
            return t;
        }
    }
    
  • RejectedExecutionHandler

    • 表示指定線程池的拒絕策略,當線程池的任務已經在緩存隊列 workQueue 中存儲滿了之後,並且不能創建新的線程來執行此任務時,就會用到此拒絕策略.
    • 它屬於一種限流保護的機制,這裡有四種任務拒絕類型:
      1. AbortPolicy: 不執行新任務,直接拋出異常,提示線程池已滿,涉及到該異常的任務也不會被執行,線程池默認的拒絕策略就是該策略。
      2. DisCardPolicy: 不執行新任務,也不拋出異常,即忽略此任務;
      3. DisCardOldSetPolicy: 將消息隊列中的第一個任務(即等待時間最久的任務)替換為當前新進來的任務執行,忽略最早的任務(最先加入隊列的任務);
      4. CallerRunsPolicy: 把任務交給當前線程來執行;
    /**
     * 線程池的拒絕策略
     */
    @Test
    public void test1() {
        // 創建線程池 核心線程為1,最大線程為3,任務隊列大小為2
        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(1, 3, 10,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(2),
                new ThreadPoolExecutor.AbortPolicy() // 添加 AbortPolicy 拒絕策略
        );
    
    
        for (int i = 0; i < 6; i++) {
            poolExecutor.execute(() -> {
                System.out.println(Thread.currentThread().getName());
            });
        }
        
    }
    
    • 自定義線程池拒絕策略
    /**
     * 自定義線程池的拒絕策略
     * 實現接口 RejectedExecutionHandler
     */
    @Test
    public void test2() {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 3, 10,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(2),
                new RejectedExecutionHandler() {
    
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        // 業務處理方法
                        System.out.println("執行自定義拒絕策略");
                    }
                }
        );
    
        for (int i = 0; i < 6; i++) {
            executor.execute(() -> {
                System.out.println(Thread.currentThread().getName());
            });
        }
    
    }
    

線程池工作原理

線程池的工作流程要從它的執行方法 execute() 說起,源碼如下:

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    int c = ctl.get();
    // 當前工作的線程數小於核心線程數
    if (workerCountOf(c) < corePoolSize) {
        // 創建新的線程執行此任務
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 檢查線程池是否處於運行狀態,如果是則把任務添加到隊列
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 再出檢查線程池是否處於運行狀態,防止在第一次校驗通過後線程池關閉
        // 如果是非運行狀態,則將剛加入隊列的任務移除
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 如果線程池的線程數為 0 時(當 corePoolSize 設置為 0 時會發生)
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false); // 新建線程執行任務
    }
    // 核心線程都在忙且隊列都已爆滿,嘗試新啟動一個線程執行失敗
    else if (!addWorker(command, false)) 
        // 執行拒絕策略
        reject(command);
}

execute() VS submit()

  • execute() 和 submit() 都是用來執行線程池任務的,它們最主要的區別是,submit() 方法可以接收線程池執行的返回值,而 execute() 不能接收返回值。
  • sumbit 之所以可以接收返回值,是因為參數中可以傳遞:Callable task,而通過 callable 創建的線程任務有返回值並且可以拋出異常。
/**
 * execute VS sumbin
 * execute 提交任務沒有返回值
 * submit 提交任務有返回值
 */
@Test
public void test3() throws ExecutionException, InterruptedException {
    ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(20));
    // execute
    executor.execute(new Runnable() {
        @Override
        public void run() {
            System.out.println("Hello, execute");
        }
    });

    // submit 使用
    Future<String> future = executor.submit(new Callable<String>() {
        @Override
        public String call() throws Exception {
            System.out.println("Hello, submit");
            return "submit success";
        }
    });
    System.out.println(future.get());
}
  • 它們的另一個區別是 execute() 方法屬於 Executor 接口的方法,而 submit() 方法則是屬於 ExecutorService 接口的方法。

線程池的使用:

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author xiandongxie
 */
public class ThreadPool {

    //參數初始化 返回Java虛擬機可用的處理器數量
//    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
    private static final int CPU_COUNT = 2;
    //核心線程數量大小
    private static final int corePoolSize = Math.max(2, Math.min(CPU_COUNT - 1, 4));
    //線程池最大容納線程數
    private static final int maximumPoolSize = CPU_COUNT * 2 + 1;
    //線程空閑后的存活時長
    private static final int keepAliveTime = 30;

    //任務過多后,存儲任務的一個阻塞隊列
    BlockingQueue<Runnable> workQueue = new SynchronousQueue<>();

    //線程的創建工廠
    ThreadFactory threadFactory = new ThreadFactory() {
        private final AtomicInteger mCount = new AtomicInteger(1);

        public Thread newThread(Runnable r) {
            return new Thread(r, "AdvacnedAsyncTask #" + mCount.getAndIncrement());
        }
    };

    //線程池任務滿載后採取的任務拒絕策略: 不執行新任務,直接拋出異常,提示線程池已滿
    RejectedExecutionHandler rejectHandler = new ThreadPoolExecutor.AbortPolicy();

    //線程池對象,創建線程
    ThreadPoolExecutor mExecute = new ThreadPoolExecutor(
            corePoolSize,
            maximumPoolSize,
            keepAliveTime,
            TimeUnit.SECONDS,
            workQueue,
            threadFactory,
            rejectHandler
    );

    public static void main(String[] args) {
        System.out.println("main start ..... \nCPU_COUNT = " + CPU_COUNT + "\tcorePoolSize=" + corePoolSize + "\tmaximumPoolSize=" + maximumPoolSize);
        
        ThreadPool threadPool = new ThreadPool();
        ThreadPoolExecutor execute = threadPool.mExecute;
        // 預啟動所有核心線程
        execute.prestartAllCoreThreads();

        for (int i = 0; i < 5; i++) {
            execute.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName() + "\tstart..." + System.currentTimeMillis());
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName() + "\tend..." + System.currentTimeMillis());
                }
            });
        }
        execute.shutdown();
        
        System.out.println("main end .....");
    }
}

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

【其他文章推薦】

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

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

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

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

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

聚甘新

Java 從入門到進階之路(二十四)

在之前的文章我們介紹了一下 Java 中的  集合框架中的Collection 的泛型,本章我們來看一下 Java 集合框架中的Collection 的子接口 List。

Collection 接口有 3 種子類型,List、Set 和 Queue,其中 List 和 Set 的區別是 Set 中不能存放相同的元素,而 List 中可以,本章我們就來介紹一下 List。

 

 

 從上圖我們可以知道 List 有兩個實例類,ArrayList 和 LinkedList,

ArrayList 是數組實現,查找快,增上慢,由於是數組實現,在增和刪的時候牽扯到數組的增容,以及靠背元素,所以慢,數組是可以直接按索引查找,所以查找時較快。

LinkedList 是鏈表實現,增刪快,查找慢,由於鏈表實現,增加時只要讓前一個元素記住自己就可以了,刪除時讓前一個元素記住后一個元素,后一個元素記住前一個元素,這樣的增刪效率高但查詢時需要一個一個遍歷,所以效率低。

LinkedList 我們可以形象的比作老式手錶的鏈子,一節扣一節,增刪時只需要打開兩個之間的節扣即可,不需要牽扯到其他節扣。

ArrayList 和 LinkedList 都有各自的優缺點,在用的時候可以根據需求自行選擇,避免性能消耗。在現在計算機計算能力越來越強,做的也不是大型項目的時候,這兩個之間的性能差異我們其實是可以忽略的。

接下來我們就來看一下 List 接口的一些基礎用法,如下:

 1 import java.util.ArrayList;
 2 import java.util.List;
 3 
 4 /**
 5  * java.util.List
 6  * 可重複集合,並且有序
 7  * 特點是可以根據下錶操作元素
 8  * ArrayList:使用數組實現,查詢更快
 9  * LinkedList:使用鏈表實現,增刪更快(收尾增刪效果更明顯)
10  */
11 
12 public class Main {
13     public static void main(String[] args) {
14         List<String> list = new ArrayList<String>();
15         list.add("one");
16         list.add("two");
17         list.add("three");
18         list.add("four");
19         /**
20          * E set(int index, E e)
21          * 將給定元素設置到制定位置上,返回原位置的元素
22          * 所以是替換元素操作
23          * 如果超出了元素的長度,則使用 add 添加,否則編譯錯誤
24          * */
25         String old = list.set(1, "2"); // 將下標為 1 的元素改為 2,返回值則是被替換的元素
26         System.out.println(old); // two
27         System.out.println(list); // [one, 2, three, four]
28 
29         /**
30          * E get(int index)
31          * 獲取給定下標對應的元素
32          * */
33         String two = list.get(1); // 獲取第二個元素
34         System.out.println(two); // 2
35 
36         /**
37          * 可以通過傳統的循環遍歷 List 集合
38          * */
39         for (int i = 0; i < list.size(); i++) {
40             System.out.println(list.get(i)); // one 2 three four
41         }
42     }
43 }

在上面的代碼中,我們通過 set 和 get 方法來設置和獲取我們想要的下標的元素,當然還有其他方法,如下:

 1 /**
 2  * List 集合提供了一對重載的 add,remove 方法
 3  * void add(int index, E e)
 4  * 將給定元素插入到指定位置,
 5  * 如果不指定下標,則插入到末尾
 6  * <p>
 7  * E remove(int index)
 8  * 從集合中刪除指定位置的元素,並將其返回
 9  */
10 
11 public class Main {
12     public static void main(String[] args) {
13         List<String> list = new ArrayList<String>();
14         list.add("one");
15         list.add("two");
16         list.add("three");
17         list.add("four");
18 
19         list.add(1, "2"); // 將下標為 1 的元素插入 2
20         System.out.println(list); // [one, 2, two, three, four]
21 
22         String string = list.remove(1); // 將下標為 1 的元素刪除,返回值為該元素
23         System.out.println(string); // 2
24         System.out.println(list); // [one, two, three, four]
25     }
26 }

我們在將 Collection 的時候講過 add 和 remove 方法,在 List 中這兩個方法被重載了,可以根據需求插入和刪除想要刪除的下標的元素,那如果我們想要獲取兩個下標之間的元素和刪除兩個下標之間的元素該怎麼辦呢,如下:

 1 import java.util.ArrayList;
 2 import java.util.List;
 3 
 4 public class Main {
 5     public static void main(String[] args) {
 6         List<Integer> list = new ArrayList<Integer>();
 7         for (int i = 0; i < 10; i++) {
 8             list.add(i);
 9         }
10         System.out.println(list); // [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
11         List<Integer> subList = list.subList(2, 5); // 獲取下標從 2 到 5 的元素,含 2 不含 5
12         System.out.println(subList); // [2, 3, 4]
13         // 將 subList 中每個元素擴大 10 倍
14         for (int i = 0; i < subList.size(); i++) {
15             subList.set(i, subList.get(i) * 10);
16         }
17         System.out.println(subList); // [20, 30, 40]
18         /**
19          * 對子集的修改,就是修改原集合相應內容
20          * */
21         System.out.println(list); // [0, 1, 20, 30, 40, 5, 6, 7, 8, 9]
22         /**
23          * 刪除集合中 2-5 的元素
24          * */
25         list.subList(2, 5).clear();
26         System.out.println(list); // [0, 1, 5, 6, 7, 8, 9]
27     }
28 }

我們說集合和數組有很多相似的地方,那課可以進行相互轉換呢,當然是可以的,如下:

 1 import java.util.ArrayList;
 2 import java.util.Collection;
 3 
 4 public class Main {
 5     public static void main(String[] args) {
 6         Collection<String> collection = new ArrayList<String>();
 7         collection.add("one");
 8         collection.add("two");
 9         collection.add("three");
10         collection.add("four");
11         System.out.println(collection); // [one, two, three, four]
12         /**
13          * 集合提供了一個 toArray,可以將當前集合轉換為數組
14          * */
15         // Object[] array = collection.toArray(); // 不常用
16         /**
17          * collection.size() 表示要轉換的數組的 length
18          * 如果大於給定的 collection 的 size,則自動填充完整 array
19          * 如果小於給定的 collection 的 size,則自動創建給你一樣長度的 size
20          * */
21         String[] array = collection.toArray(new String[collection.size()]);
22         System.out.println(array.length); // 4
23         for (String string : array) {
24             System.out.println(string); // one two three four
25         }
26 
27         String[] array1 = collection.toArray(new String[6]);
28         System.out.println(array.length); // 4
29         for (String string : array1) {
30             System.out.println(string); // one two three four null null
31         }
32 
33         String[] array2 = collection.toArray(new String[1]);
34         System.out.println(array.length); // 4
35         for (String string : array2) {
36             System.out.println(string); // one two three four
37         }
38     }
39 }

在上面的代碼中我們實現了集合轉換為數組的方法,接下來我們再看一下數組轉換為集合的方法:

 1 import java.util.ArrayList;
 2 import java.util.Arrays;
 3 import java.util.List;
 4 
 5 /**
 6  * 數組轉換為集合
 7  * 需要注意,轉換隻能轉換為 List 集合
 8  * 使用的是數組的工具類 Arrays 的靜態方法 asList
 9  * 只能轉換為 List 集合的主要原因是:Set 不能存放重複元素
10  * 所以若轉換為 Set 集合可能會出現丟失元素的情況
11  */
12 public class Main {
13     public static void main(String[] args) {
14         String[] array = {"one", "two", "three", "four"};
15         List<String> list = Arrays.asList(array);
16         System.out.println(list); // [one, two, three, four]
17 
18         /**
19          * 向集合中添加元素,會出現編譯錯誤
20          * 相當於在原數組添加元素
21          * 該集合時由數組轉換過來的,那麼該集合就表示原來的數組
22          * 所以對集合的操作就是對數組的操作
23          * 那麼添加元素會導致原數組擴容
24          * 那麼久不能表示原來的數組了
25          * 所以不允許向該集合添加元素
26          */
27         // list.add("five"); // 編譯錯誤 Exception in thread "main" java.lang.UnsupportedOperationException
28 
29         /**
30          * 若希望增刪元素,需要另外創建一個集合
31          * */
32         /**
33          * 所有的集合都提供了一個帶有 Collection 類型參數的構造方法
34          * 該構造方法稱為:複製構造器
35          * 作用是在創建當前集合的同時,
36          * 集合中包含給定集合中的所有元素
37          * */
38         // List<String> list1 = new ArrayList<String>(list); // 複製構造器,一步到位
39         List<String> list1 = new ArrayList<String>();
40         list1.addAll(list);
41         list1.add("five");
42         System.out.println(list1); // [one, 2, three, four, five]
43 
44         /**
45          * 修改集合元素,數組元素也會改變
46          * */
47         list.set(1, "2");
48         System.out.println(list); // [one, 2, three, four]
49         for (String string : array) {
50             System.out.println(string); // one 2 three four
51         }
52     }
53 }

    

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

【其他文章推薦】

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

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

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

※超省錢租車方案

FB行銷專家,教你從零開始的技巧

聚甘新

Spring Boot 2 實戰:利用Redis的Geo功能實現查找附近的位置

1. 前言

老闆突然要上線一個需求,獲取當前位置方圓一公里的業務代理點。明天上線!當接到這個需求的時候我差點吐血,這時間也太緊張了。趕緊去查相關的技術選型。經過一番折騰,終於在晚上十點完成了這個需求。現在把大致實現的思路總結一下。

2. MySQL 不合適

遇到需求,首先要想到現有的東西能不能滿足,成本如何。

MySQL是我首先能夠想到的,畢竟大部分數據要持久化到MySQL。但是使用MySQL需要自行計算Geohash。需要使用大量數學幾何計算,並且需要學習地理相關知識,門檻較高,短時間內不可能完成需求,而且長期來看這也不是MySQL擅長的領域,所以沒有考慮它。

Geohash 參考 https://www.cnblogs.com/LBSer/p/3310455.html

2. Redis 中的GEO

Redis是我們最為熟悉的K-V數據庫,它常被拿來作為高性能的緩存數據庫來使用,大部分項目都會用到它。從3.2版本開始它開始提供了GEO能力,用來實現諸如附近位置、計算距離等這類依賴於地理位置信息的功能。GEO相關的命令如下:

Redis命令 描述
GEOHASH 返回一個或多個位置元素的 Geohash 表示
GEOPOS 從key里返回所有給定位置元素的位置(經度和緯度)
GEODIST 返回兩個給定位置之間的距離
GEORADIUS 以給定的經緯度為中心, 找出某一半徑內的元素
GEOADD 將指定的地理空間位置(緯度、經度、名稱)添加到指定的key中
GEORADIUSBYMEMBER 找出位於指定範圍內的元素,中心點是由給定的位置元素決定

Redis會假設地球為完美的球形, 所以可能有一些位置計算偏差,據說<=0.5%,對於有嚴格地理位置要求的需求來說要經過一些場景測試來檢驗是否能夠滿足需求。

2.1 寫入地理信息

那麼如何實現目標單位半徑內的所有元素呢?我們可以將所有的位置的經緯度通過上表中的GEOADD將這些地理信息轉換為52位的Geohash寫入Redis

該命令格式:

geoadd key longitude latitude member [longitude latitude member ...]

對應例子:

redis> geoadd cities:locs 117.12 39.08 tianjin 114.29 38.02  shijiazhuang 
(integer) 2

意思是將經度為117.12緯度為39.08的地點tianjin和經度為114.29緯度為38.02的地點shijiazhuang加入keycities:locssorted set集合中。可以添加一到多個位置。然後我們就可以藉助於其他命令來進行地理位置的計算了。

有效的經度從-180度到180度。有效的緯度從-85.05112878度到85.05112878度。當坐標位置超出上述指定範圍時,該命令將會返回一個錯誤。

2.2 統計單位半徑內的地區

我們可以藉助於GEORADIUS來找出以給定經緯度,某一半徑內的所有元素。

該命令格式:

georadius key longtitude latitude radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count] [ASC|DESC] 

這個命令比GEOADD要複雜一些:

  • radius 半徑長度,必選項。後面的mkmftmi、是長度單位選項,四選一。
  • WITHCOORD 將位置元素的經度和維度也一併返回,非必選。
  • WITHDIST 在返回位置元素的同時, 將位置元素與中心點的距離也一併返回。 距離的單位和查詢單位一致,非必選。
  • WITHHASH 返回位置的52位精度的Geohash值,非必選。這個我反正很少用,可能其它一些偏向底層的LBS應用服務需要這個。
  • COUNT 返回符合條件的位置元素的數量,非必選。比如返回前10個,以避免出現符合的結果太多而出現性能問題。
  • ASC|DESC 排序方式,非必選。默認情況下返回未排序,但是大多數我們需要進行排序。參照中心位置,從近到遠使用ASC ,從遠到近使用DESC

例如,我們在 cities:locs 中查找以(115.03,38.44)為中心,方圓200km的城市,結果包含城市名稱、對應的坐標和距離中心點的距離(km),並按照從近到遠排列。命令如下:

redis> georadius cities:locs 115.03 38.44 200 km WITHCOORD WITHDIST ASC
1) 1) "shijiazhuang"
   2) "79.7653"
   3) 1) "114.29000169038772583"
      2) "38.01999994251037407"
2) 1) "tianjin"
   2) "186.6937"
   3) 1) "117.02000230550765991"
      2) "39.0800000535766543"

你可以加上 COUNT 1來查找最近的一個位置。

3. 基於Redis GEO實戰

大致的原理思路說完了,接下來就是實操了。結合Spring Boot應用我們應該如何做?

3.1 開發環境

需要具有GEO特性的Redis版本,這裏我使用的是Redis 4 。另外我們客戶端使用 spring-boot-starter-data-redis 。這裏我們會使用到 RedisTemplate對象。

3.2 批量添加位置信息

第一步,我們需要將位置數據初始化到Redis中。在Spring Data Redis中一個位置坐標(lng,lat) 可以封裝到org.springframework.data.geo.Point對象中。然後指定一個名稱,就組成了一個位置Geo信息。RedisTemplate提供了批量添加位置信息的方法。我們可以將章節2.1中的添加命令轉換為下面的代碼:

   Map<String, Point> points = new HashMap<>();
   points.put("tianjin", new Point(117.12, 39.08));
   points.put("shijiazhuang", new Point(114.29, 38.02));
   // RedisTemplate 批量添加 Geo
   redisTemplate.boundGeoOps("cities:locs").add(points);

可以結合Spring Boot 提供的ApplicationRunner接口來實現初始化。

@Bean
public ApplicationRunner cacheActiveAppRunner(RedisTemplate<String, String> redisTemplate) {

    return args -> {
        final String GEO_KEY = "cities:locs";

        // 清理緩存
        redisTemplate.delete(GEO_KEY);
        
        Map<String, Point> points = new HashMap<>();
        points.put("tianjin", new Point(117.12, 39.08));
        points.put("shijiazhuang", new Point(114.29, 38.02));
        // RedisTemplate 批量添加 GeoLocation
        BoundGeoOperations<String, String> geoOps = redisTemplate.boundGeoOps(GEO_KEY);
        geoOps.add(points);
    };
}

地理數據持久化到MySQL,然後同步到Redis中。

3.3 查詢附近的特定位置

RedisTemplate 針對GEORADIUS命令也有封裝:

GeoResults<GeoLocation<M>> radius(K key, Circle within, GeoRadiusCommandArgs args)

Circle對象是封裝覆蓋的面積(圖1),需要的要素為中心點坐標Point對象、半徑(radius)、計量單位(metric), 例如:

Point point = new Point(115.03, 38.44);

Metric metric = RedisGeoCommands.DistanceUnit.KILOMETERS;
Distance distance = new Distance(200, metric);

Circle circle = new Circle(point, distance);

GeoRadiusCommandArgs用來封裝GEORADIUS的一些可選命令參數,參見章節2.2中的WITHCOORDCOUNTASC等,例如我們需要在返回結果中包含坐標、中心距離、由近到遠排序的前5條數據:

RedisGeoCommands.GeoRadiusCommandArgs args = RedisGeoCommands
        .GeoRadiusCommandArgs
        .newGeoRadiusArgs()
        .includeDistance()
        .includeCoordinates()
        .sortAscending()
        .limit(limit);

然後執行 radius方法就會拿到GeoResults<RedisGeoCommands.GeoLocation<String>>封裝的結果,我們對這個可迭代對象進行解析就可以拿到我們想要的數據:

GeoResults<RedisGeoCommands.GeoLocation<String>> radius = redisTemplate.opsForGeo()
        .radius(GEO_STAGE, circle, args);

if (radius != null) {
    List<StageDTO> stageDTOS = new ArrayList<>();
    radius.forEach(geoLocationGeoResult -> {
        RedisGeoCommands.GeoLocation<String> content = geoLocationGeoResult.getContent();
        //member 名稱  如  tianjin 
        String name = content.getName();
        // 對應的經緯度坐標
        Point pos = content.getPoint();
        // 距離中心點的距離
        Distance dis = geoLocationGeoResult.getDistance();
    });
}

3.4 刪除元素

有時候我們可能需要刪除某個位置元素,但是RedisGeo並沒有刪除成員的命令。不過由於它的底層是zset,我們可以藉助zrem命令進行刪除,對應的Java代碼為:

redisTemplate.boundZSetOps(GEO_STAGE).remove("tianjin");

4. 總結

今天我們使用RedisGeo特性實現了常見的附近的地理信息查詢需求,簡單易上手。其實使用另一個Nosql數據庫MongoDB也可以實現。在數據量比較小的情況下Redis已經能很好的滿足需要。如果數據量大可使用MongoDB來實現。 文中涉及的DEMO可通過我個人博客獲取。

關注公眾號:Felordcn 獲取更多資訊

個人博客:https://felord.cn

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

【其他文章推薦】

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

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

※超省錢租車方案

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

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

聚甘新

跨雲廠商部署 k3s 集群

原文鏈接:https://fuckcloudnative.io/posts/deploy-k3s-cross-public-cloud/

最近一兩年各大雲服務商都出了各種福利活動,很多小夥伴薅了一波又一波羊毛,比如騰訊雲 1C2G 95/年 真香系列,華為雲和阿里雲也都有類似的活動,薅個兩三台就能搭建一個 Kubernetes 集群。但是跨雲服務商搭建 Kubernetes 集群並不像我們想象中的那麼容易,首先就是原生的 Kubernetes 組件本身對資源的消耗量很大,而雲服務器的資源非常有限,經不起這麼大傢伙的折騰,對此我們可以選擇使用輕量級 Kubernetes 發行版:k3s

k3s 將安裝 Kubernetes 所需的一切打包進僅有 60MB 大小的二進制文件中,並且完全實現了 Kubernetes API。為了減少運行 Kubernetes 所需的內存,k3s 刪除了很多不必要的驅動程序,並用附加組件對其進行替換。由於它只需要極低的資源就可以運行,因此它能夠在任何 512MB 內存以上的設備上運行集群。

其實 k3s 的安裝非常簡單,分分鐘就能搞定,但對於公有雲來說,還是有很多坑的,比如內網不通、公網 IP 不在服務器上該咋辦?本文就為你一一解決這些難題,讓天下的雲羊毛都成為 k3s 的後宮!

1. 下載二進制文件

首先來解決第一個難題:k3s 二進制文件的下載。國內下載 GitHub 速度基本都是以幾個 kb 為單位,不忍直視,如果下載內容都是代碼,有很多辦法可以解決,比如通過碼雲中轉啊、直接通過 CDN 下載啊,什麼?你不知道可以通過 CDN 下載?好吧沒關係,現在我告訴你了:https://cdn.con.sh/。

但是上面的 CDN 並不能下載 release 里的內容,要想下載 release 里的內容,可以使用這個網站:https://toolwa.com/github/。打開網站,輸入 release 裏面的文件下載鏈接,點擊起飛即可加速下載。

當然,如果你會魔法上網的話,上面的所有花里胡哨的方法都可以無視,直接下載就好啦(本文選擇使用版本 v1.17.6+k3s1):

$ wget https://github.com/rancher/k3s/releases/download/v1.17.6+k3s1/k3s -O /usr/local/bin/k3s
$ chmod +x /usr/local/bin/k3s

需要在所有節點中下載上述二進制文件。

2. 升級內核

k3s 的默認網絡插件是 flannel,默認模式是 vxlan 模式,建議使用 wireguard 模式,原因不解釋了,不知道 wireguard 是啥的自己去搜一下。

wireguard 對內核的要求比較高,而 CentOS 7.x 的默認內核是不滿足要求的,需要升級內核(如果你的操作系統是 CentOS 7.x 的話)。步驟如下:

① 載入公鑰

$ rpm --import https://www.elrepo.org/RPM-GPG-KEY-elrepo.org

② 升級安裝 elrepo

$ rpm -Uvh http://www.elrepo.org/elrepo-release-7.0-3.el7.elrepo.noarch.rpm

③ 載入 elrepo-kernel 元數據

$ yum --disablerepo=\* --enablerepo=elrepo-kernel repolist

④ 安裝最新版本的內核

$ yum --disablerepo=\* --enablerepo=elrepo-kernel install  kernel-ml.x86_64  -y

⑤ 刪除舊版本工具包

$ yum remove kernel-tools-libs.x86_64 kernel-tools.x86_64  -y

⑥ 安裝新版本工具包

$ yum --disablerepo=\* --enablerepo=elrepo-kernel install kernel-ml-tools kernel-ml-devel kernel-ml-headers -y

⑦ 查看內核插入順序

$ grep "^menuentry" /boot/grub2/grub.cfg | cut -d "'" -f2

CentOS Linux (3.10.0-1127.10.1.el7.x86_64) 7 (Core)
CentOS Linux (5.7.2-1.el7.elrepo.x86_64) 7 (Core)
CentOS Linux (0-rescue-96820b9851c24560b5f942f2496b9aeb) 7 (Core)

默認新內核是從頭插入,默認啟動順序也是從 0 開始。

⑧ 查看當前實際啟動順序

$ grub2-editenv list

saved_entry=CentOS Linux (3.10.0-1127.10.1.el7.x86_64) 7 (Core)

⑨ 設置默認啟動

$ grub2-set-default 'CentOS Linux (5.7.2-1.el7.elrepo.x86_64) 7 (Core)'

最後重啟檢查:

$ reboot
$ uname -r

注意:集群中的所有節點都需要升級內核。

3. 安裝 wireguard

內核升級了之後,就可以安裝 wireguard 了,也很簡單,步驟如下:

$ yum install epel-release https://www.elrepo.org/elrepo-release-7.el7.elrepo.noarch.rpm
$ yum install yum-plugin-elrepo
$ yum install kmod-wireguard wireguard-tools

注意:集群中的所有節點都需要安裝。

4. 部署控制平面

下面就可以在控制節點上啟動控制平面的組件了,這裏我們選擇手動部署,這樣比較方便修改參數。先創建一個 Service Unit 文件:

$ cat > /etc/systemd/system/k3s.service <<EOF
[Unit]
Description=Lightweight Kubernetes
Documentation=https://k3s.io
Wants=network-online.target

[Install]
WantedBy=multi-user.target

[Service]
Type=notify
EnvironmentFile=/etc/systemd/system/k3s.service.env
KillMode=process
Delegate=yes
# Having non-zero Limit*s causes performance problems due to accounting overhead
# in the kernel. We recommend using cgroups to do container-local accounting.
LimitNOFILE=1048576
LimitNPROC=infinity
LimitCORE=infinity
TasksMax=infinity
TimeoutStartSec=0
Restart=always
RestartSec=5s
ExecStartPre=-/sbin/modprobe br_netfilter
ExecStartPre=-/sbin/modprobe overlay
ExecStart=/usr/local/bin/k3s \
    server \
    --tls-san <public_ip> \
    --node-ip <public_ip> \
    --node-external-ip <public_ip> \
    --no-deploy servicelb \
    --flannel-backend wireguard \
    --kube-proxy-arg "proxy-mode=ipvs" "masquerade-all=true" \
    --kube-proxy-arg "metrics-bind-address=0.0.0.0"
EOF
  • <public_ip> 替換成控制節點的公網 IP。
  • flannel 使用 wireguard 協議來跨主機通信。
  • kube-proxy 使用 ipvs 模式。

啟動 k3s 控制平面並設置開機自啟:

$ systemctl enable k3s --now

查看集群組件健康狀況:

$ kubectl get cs

NAME                 STATUS    MESSAGE   ERROR
scheduler            Healthy   ok
controller-manager   Healthy   ok

這裏的輸出沒有 etcd,因為 k3s 的默認數據存儲是 Sqlite,對於小型數據庫十分友好。Kubernetes 控制平面中發生的更改更多是與頻繁更新部署、調度 Pod 等有關,因此對於幾個節點的小型集群而言,數據庫不會造成太大負載,能省下不少資源,真香!

5. 加入計算節點

部署好控制平面之後,就可以加入計算節點了。首先在計算節點上創建 Service Unit 文件:

$ cat > /etc/systemd/system/k3s-agent.service <<EOF
[Unit]
Description=Lightweight Kubernetes
Documentation=https://k3s.io
Wants=network-online.target

[Install]
WantedBy=multi-user.target

[Service]
Type=exec
EnvironmentFile=/etc/systemd/system/k3s-agent.service.env
KillMode=process
Delegate=yes
LimitNOFILE=infinity
LimitNPROC=infinity
LimitCORE=infinity
TasksMax=infinity
TimeoutStartSec=0
Restart=always
RestartSec=5s
ExecStartPre=-/sbin/modprobe br_netfilter
ExecStartPre=-/sbin/modprobe overlay
ExecStart=/usr/local/bin/k3s agent \
    --node-external-ip <public_ip> \
    --node-ip <public_ip> \
    --kube-proxy-arg "proxy-mode=ipvs" "masquerade-all=true" \
    --kube-proxy-arg "metrics-bind-address=0.0.0.0"
EOF

環境變量文件 /etc/systemd/system/k3s-agent.service.env 中需要加入兩個環境變量:

  • K3S_URL : API Server 的 URL,一般格式為:https://<master_ip>:6443。其中 <master_ip> 是控制節點的公網 IP。
  • K3S_TOKEN : 加入集群所需的 token,可以在控制節點上查看 /var/lib/rancher/k3s/server/node-token 文件。

/etc/systemd/system/k3s-agent.service.env 內容如下:

K3S_URL=https://<master_ip>:6443
K3S_TOKEN=xxxxxxxx

啟動 k3s-agent 並設置開啟自啟:

$ systemctl enable k3s-agent --now

查看節點狀態:

$ kubectl get node

NAME         STATUS   ROLES    AGE     VERSION
blog-k3s01   Ready    master   3d6h    v1.17.6+k3s1
blog-k3s02   Ready    <none>   3d3h    v1.17.6+k3s1

6. 內網不互通的解決辦法

這裡會遇到一個問題,不同節點的 flannel 使用的是內網 IP 來進行通信,而我們的雲服務器是內網不互通的,而且公網 IP 也不在服務器上。可以看一下 node 的 annotations

$ kubectl get node blog-k3s02 -o yaml

apiVersion: v1
kind: Node
metadata:
  annotations:
    flannel.alpha.coreos.com/backend-data: '"xxxxx"'
    flannel.alpha.coreos.com/backend-type: extension
    flannel.alpha.coreos.com/kube-subnet-manager: "true"
    flannel.alpha.coreos.com/public-ip: 192.168.0.11
    ...

可以看到 flannel 給節點打的註解中的節點 IP 是內網 IP。要想讓 flannel 使用公網 IP 進行通信,需要額外添加一個註解 public-ip-overwrite,然後 flannel 會基於這個 IP 配置網絡。按照官方文檔的說法,如果你的 node 設置了 ExternalIP,flannel 會自動給 node 添加一個註解 public-ip-overwrite,但我不知道該如何給 node 設置 ExternalIP,乾脆就直接手動加註解吧:

$ kubectl annotate nodes <master> flannel.alpha.coreos.com/public-ip-overwrite=<master_pub_ip>
$ kubectl annotate nodes <node> flannel.alpha.coreos.com/public-ip-overwrite=<node_pub_ip>

加了註解之後,flannel 的 public-ip 就會被修改為公網 IP。然後在各個節點上重啟各自的 k3s 服務,查看 wireguard 連接狀況:

$ wg show flannel.1

interface: flannel.1
  public key: ONDgJCwxxxxxxxJvdWpoOKTxQA=
  private key: (hidden)
  listening port: 51820
  
peer: MKKaanTxxxxxxxV8VpcHq4CSRISshw=
  endpoint: <pub_ip>:51820
  allowed ips: 10.42.4.0/24
  latest handshake: 26 seconds ago
  transfer: 133.17 KiB received, 387.44 KiB sent
  persistent keepalive: every 25 seconds

可以看到通信端點被改成了公網 IP,大功告成!

7. metrics-server 問題解決

還有一個問題就是 metrics-server 無法獲取 cpu、內存等利用率核心指標。需要修改 metrics-server 的 manifests,使用以下命令在線編輯 metrics-server 的 manifests:

$ kubectl -n kube-system edit deploy metrics-server

然後加入以下執行參數后保存退出:

      -command:
        - /metrics-server
        - --kubelet-preferred-address-types=ExternalIP
        - --kubelet-insecure-tls

這樣就可以讓 metrics-server 使用公網 IP 來和 node 通信了。修改成功后就可以看到核心指標了:

$ kubectl top nodes
NAME         CPU(cores)   CPU%   MEMORY(bytes)   MEMORY%
blog-k3s01   193m         9%     886Mi           22%
blog-k3s02   41m          2%     1292Mi          32%

$ kubectl top pod -n kube-system
NAME                                      CPU(cores)   MEMORY(bytes)
coredns-848b6cc76f-zq576                  8m           14Mi
local-path-provisioner-58fb86bdfd-bzdfl   2m           9Mi
metrics-server-bdfc79c97-djmzk            1m           12Mi

到這裏跨雲服務商部署 k3s 基本上就大功告成了,下一篇文章將會教你如何打通家裡到雲上 k3s 的網絡,讓你家中所有設備都可以直接訪問 Pod IP、svc IP,甚至可以直接訪問 svc 域名,敬請期待。

Kubernetes 1.18.2 1.17.5 1.16.9 1.15.12離線安裝包發布地址http://store.lameleg.com ,歡迎體驗。 使用了最新的sealos v3.3.6版本。 作了主機名解析配置優化,lvscare 掛載/lib/module解決開機啟動ipvs加載問題, 修復lvscare社區netlink與3.10內核不兼容問題,sealos生成百年證書等特性。更多特性 https://github.com/fanux/sealos 。歡迎掃描下方的二維碼加入釘釘群 ,釘釘群已經集成sealos的機器人實時可以看到sealos的動態。

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

【其他文章推薦】

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

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

※回頭車貨運收費標準

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

※超省錢租車方案

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

聚甘新

Flutter學習筆記(35)–通知Notification,Flutter學習筆記(35)–通知Notification

如需轉載,請註明出處:Flutter學習筆記(35)–通知Notification

通知的NotificationListener和我們之前寫的事件的Listener一樣,都是功能性的組件,而且也都是從子節點順着widget樹向上冒泡,不同的是,事件的Listener不可以被終止,但是通知的NotificationListener是可以被終止的。

是否終止根據NotificationListener的返回值來決定。

說一下我個人的理解:

通知Notification的發送是通過disPatch進行分發的,就好像Android裏面的事件分發,當NotificationListener監聽到了通知事件,這時候會走到其onNotification回調中,根據回調中的返回值類型(true還是false)來決定是否還繼續向父親節點發送通知。

返回true就是繼續分發,返回false就是終止分發,返回false就意味着上層節點的NotificationListener就不會接收到通知事件了。

舉個例子就是:

兩層NotificationListener嵌套,子節點的NotificationListener返回true,那麼父親節點的NotificationListener可以接收到通知事件,反之如果返回false,那麼父親節點的NotificationListener就不會接收到通知事件了。

下面看一下demo示例:

demo就是簡單的發送通知,監聽到通知事件后改變text的內容。

1.創建一個事件通知類,要繼承Notification,它其實就是一個數據載體,在裏面定義通知數據的類型和內容。

import 'package:flutter/material.dart';

class MyNotification extends Notification{
  String notificationStr;

  MyNotification(this.notificationStr);
}

2.NotificationListener的使用和通知事件的分發

import 'package:flutter/material.dart';
import 'package:study_app/util/MyNotification.dart';

class NotificationDemo extends StatefulWidget {
  @override
  State<StatefulWidget> createState() {
    return _NotificationDemoState();
  }
}

class _NotificationDemoState extends State {
  String _notificationData = 'default_data';

  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      title: 'NotificationDemo',
      home: new Scaffold(
          appBar: AppBar(
            title: Text('NotificationDemo'),
          ),
          body: NotificationListener<MyNotification>(
            onNotification: (notification) {
              setState(() {
                _notificationData = notification.notificationStr;
              });
              return true;
            },
            child: Column(
              children: <Widget>[
                Text(_notificationData),
                Builder(
                  builder: (context) {
                    return Container(
                      width: double.infinity,
                      child: RaisedButton(
                          child: Text('發送通知'),
                          onPressed: () {
                            MyNotification('notification_data')
                                .dispatch(context);
                          }),
                    );
                  },
                )
              ],
            ),
          )),
    );
  }
}

在看書的時候,作者強調了一種錯誤的寫法,如下圖註釋的部分:

原因是通知在分發的時候,需要一個context參數,這個參數指的是Notification監聽的子widget的context,如果按照註釋部分的寫法的話,context是根widget的,這樣會導致監聽不到子widget了。

所以需要我們通過Builder構建出我們子widget的context,這裏需要特別注意一下。

最後看一下效果截圖:

   

以上!有任何疑問歡迎留言!

 

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

【其他文章推薦】

※超省錢租車方案

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

※回頭車貨運收費標準

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

FB行銷專家,教你從零開始的技巧

聚甘新

003.OpenShift網絡

一 OpenShift網絡實現

1.1 軟件定義網絡(SDN)


默認情況下,Docker網絡使用僅使用主機虛機網橋bridge,主機內的所有容器都連接至該網橋。連接到此橋的所有容器都可以彼此通信,但不能與不同主機上的容器通信。通常,這種通信使用端口映射來處理,其中容器端口綁定到主機上的端口,所有通信都通過物理主機上的端口路由。

當有大量主機和容器時,使用此模式,需要手動管理所有端口綁定非常不現實。

為了支持跨集群的容器之間的通信,OpenShift容器平台使用了軟件定義的網絡(SDN)方法。軟件定義的網絡是一種網絡模型,它通過幾個網絡層的抽象來管理網絡服務。SDN將處理流量的軟件(稱為控制平面)和路由流量的底層機制(稱為數據平面)解耦。SDN支持控制平面和數據平面之間的通信。

在OpenShift Container Platform 3.9中(之後簡稱OCP),管理員可以為pod網絡配置三個SDN插件:

  1. ovs-subnet:默認插件,子網提供了一個flat pod網絡,其中每個pod可以與其他pod和service通信。
  2. ovs-multitenant:該為pod和服務提供了額外的隔離層。當使用此插件時,每個project接收一個惟一的虛擬網絡ID (VNID),該ID標識來自屬於該project的pod的流量。通過使用VNID,來自不同project的pod不能與其他project的pod和service通信。
  3. ovs-networkpolicy:此插件允許管理員使用NetworkPolicy對象定義自己的隔離策略。


cluster network由OpenShift SDN建立和維護,它使用Open vSwitch創建overlay網絡,master節點不能通過集群網絡訪問容器,除非master同時也為node節點。

注意:VNID為0的project可以與所有其他pod通信,在OpenShift容器平台中,默認項目的VNID為0。

1.2 Kubernetes SDN Pod




在默認的OpenShift容器平台安裝中,每個pod都有一個惟一的IP地址。pod中的所有容器都對外表現在相同的主機上。給每個pod提供自己的IP地址意味着,在端口分配、網絡、DNS、負載平衡、應用程序配置和遷移方面,pod被視為物理主機或虛擬機的獨立節點(僅從網絡層面看待)。

Kubernetes提供了service的概念,在任何OpenShift應用程序中,service都是必不可少的資源。service充當一個或多個pod前的負載平衡器。該service提供一個固定的IP地址,並且允許與pod通信,而不必跟蹤單獨的pod IP地址。



大多數實際應用程序都不是作為單個pod運行的。它們需要水平伸縮,這樣應用程序就可以在許多pod上運行,以滿足不斷增長的用戶需求。在OpenShift集群中,pod不斷地在集群中的節點之間創建和銷毀。每次創建pod時,它們都會獲得一個不同的IP地址。一個service提供一個單獨的、惟一的IP地址供其他pod使用,而不依賴於pod運行的節點,因此一個pod不必一定需要發現另一個pod的IP地址。客戶端通過service的請求在不同pod之間實現負載均衡。

1.3 Kubernetes SDN Service


service背後運行的一組pod由OpenShift容器平台自動管理。每個service都被分配了一個唯一的IP地址供客戶端連接。這個IP地址也來自OpenShift SDN,它與pod的內部網絡不同,也只在集群中可見。每個與selector匹配的pod都作為endpoint添加到service資源中。當創建和銷毀pods時,service後面的endpoint將自動更新。

service yaml語法:

  1 - apiVersion: v1
  2   kind: Service			#聲明資源類型
  3   metadata:
  4     labels:
  5       app: hello-openshift
  6       name: hello-openshift	#服務的唯一名稱
  7   spec:
  8     ports:,
  9     - name: 8080-tcp
 10       port: 8080		#服務對外公開的端口客戶機連接到服務端口
 11       protocol: TCP
 12       targetPort: 8080		#targetPort屬性必須匹配pod容器定義中的containerPort,服務將數據包轉發到pod中定義的目標端口。
 13     selector:			#該服務使用selector屬性查找要轉發數據包的pod。目標pod的元數據中需要有匹配的標籤。如果服務發現多個具有匹配標籤的pod,它將在它們之間實現負載
 14       app: hello-openshift
 15       deploymentconfig: hello-openshift


1.4 service對外暴露


默認情況下,pod和service IP地址不能從OpenShift集群外部訪問。對於需要從OpenShift集群外部訪問服務的應用程序,可以通過以下三種方式。

HostPort/HostNetwork:在這種方法中,client可以通過主機上的網絡端口直接訪問集群中的應用程序pod。應用程序pod中的端口被綁定到運行該pod的主機上的端口。這種方法在集群中運行大量pod時,存在端口衝突的風險。

NodePort:這是一種較老的基於Kubernetes的方法,通過綁定到node主機上的可用端口,將service公開給外部客戶端,然後node主機代理到service IP地址的連接。使用oc edit svc命令編輯服務屬性,指定NodePort的類型,併為NodePort屬性提供端口值。OpenShift然後通過node主機的公共IP地址和nodePort中設置的端口值代理到服務的連接。這種方法支持非http通信。

OpenShift routes:OpenShift中的推薦方式。它使用唯一的URL公開服務。使用oc expose命令公開用於外部訪問的服務,或者從OpenShift web控制台公開服務。在這種方法中,目前只支持HTTP、HTTPS、TLS whit SNI和WebSockets。

附圖:显示了NodePort服務如何允許外部訪問Kubernetes服務。



service nodeport yaml語法:

  1 apiVersion: v1
  2 kind: Service
  3 metadata:
  4 ...
  5 spec:
  6   ports:
  7   - name: 3306-tcp
  8     port: 3306
  9     protocol: TCP
 10     targetPort: 3306	#pod目標端口,即需要和pod定義的端口匹配
 11     nodePort: 30306	#OpenShift集群中主機上的端口,暴露給外部客戶端
 12   selector:
 13     app: mysqldb
 14     deploymentconfig: mysqldb
 15     sessionAffinity: None
 16   type: NodePort	#服務的類型,如NodePort
 17 ...



OpenShift將服務綁定到服務定義的nodePort屬性中定義的值,併為集群中所有node(包括master)上的流量打開該端口。外部客戶端可以連接到node端口上的任何節點的公共IP地址來訪問服務。請求會在服務後面的各個pod之間實現輪詢的負載平衡。

OpenShift route主要限於HTTP和HTTPS流量,但是節點端口可以處理非HTTP流量,當設置好公開的端口后,客戶機可以使用TCP或UDP的協議連接到該端口。

提示:缺省情況下,NodePort屬性的端口號限制在30000-32767之間,可通過在OpenShift主配置文件中配置範圍。

node port在集群中的所有node上都是打開的,包括master節點。如果沒有提供node端口值,OpenShift將自動在配置範圍內分配一個隨機端口

1.5 pod訪問外部網絡


pod可以使用其主機的地址與外部網絡通信。只要主機能夠解析pod需要到達的服務器,pod就可以使用網絡地址轉換(network address translation, NAT)機制與目標服務器通信。

二 OpenShift SDN練習

2.1 前置準備


[student@workstation ~]$ lab install-prepare setup

[student@workstation ~]$ cd /home/student/do280-ansible

[student@workstation do280-ansible]$ ./install.sh

提示:以上準備為部署一個正確的OpenShift平台。

2.2 本練習準備


[student@workstation ~]$ lab openshift-network setup #準備本實驗環境

2.3 創建應用


[student@workstation ~]$ oc login -u developer -p redhat https://master.lab.example.com

[student@workstation ~]$ oc new-project network-test #創建project

[student@workstation ~]$ oc new-app –name=hello -i php:7.0 http://registry.lab.example.com/scaling

[student@workstation ~]$ oc get pods

NAME READY STATUS RESTARTS AGE

hello-1-build 1/1 Running 0 8s

2.4 擴展應用


[student@workstation ~]$ oc scale –replicas=2 dc hello

[student@workstation ~]$ oc get pods -o wide

NAME READY STATUS RESTARTS AGE IP NODE

hello-1-kszfh 1/1 Running 0 11m 10.128.0.21 node1.lab.example.com

hello-1-q7wk2 1/1 Running 0 11m 10.129.0.37 node2.lab.example.com

2.5 測試訪問


[student@workstation ~]$ curl http://10.128.0.21:8080

curl: (7) Failed connect to 10.128.0.21:8080; Network is unreachable

[root@node1 ~]# curl http://10.128.0.21:8080

  1 <html>
  2  <head>
  3   <title>PHP Test</title>
  4  </head>
  5  <body>
  6  <br/> Server IP: 10.128.0.21
  7  </body>
  8 </html>
  9 [root@node1 ~]# curl http://10.129.0.37:8080
 10 <html>
 11  <head>
 12   <title>PHP Test</title>
 13  </head>
 14  <body>
 15  <br/> Server IP: 10.129.0.37
 16  </body>
 17 </html>



提示:默認情況下,pod的ip屬於內部,集群內部節點可以使用pod ip訪問,集群外部(如workstation)無法訪問。

[student@workstation ~]$ oc get svc hello

NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE

hello ClusterIP 172.30.253.212 <none> 8080/TCP 14m

[student@workstation ~]$ curl http://172.30.253.212:8080

curl: (7) Failed connect to 172.30.253.212:8080; Network is unreachable

[root@node1 ~]# curl http://172.30.253.212:8080 #驗證負載均衡

  1 <html>
  2  <head>
  3   <title>PHP Test</title>
  4  </head>
  5  <body>
  6  <br/> Server IP: 10.128.0.21
  7  </body>
  8 </html>
  9 [root@node1 ~]# curl http://172.30.253.212:8080		#驗證負載均衡
 10 <html>
 11  <head>
 12   <title>PHP Test</title>
 13  </head>
 14  <body>
 15  <br/> Server IP: 10.129.0.37
 16  </body>
 17 </html>



提示:默認情況下,cluster的ip屬於內部,集群內部節點可以使用cluster ip訪問,集群外部(如workstation)無法訪問。

2.6 檢查服務


[student@workstation ~]$ oc describe svc hello

Name: hello

Namespace: network-test

Labels: app=hello

Annotations: openshift.io/generated-by=OpenShiftNewApp

Selector: app=hello,deploymentconfig=hello

Type: ClusterIP

IP: 172.30.253.212

Port: 8080-tcp 8080/TCP

TargetPort: 8080/TCP

Endpoints: 10.128.0.21:8080,10.129.0.37:8080

Session Affinity: None

Events: <none>

解釋:

endpoint:显示請求路由到的pod IP地址列表。當pod有更新后,endpoint將自動更新。

Selector:OpenShift使用為pods定義的選擇器和標籤來使用給定的集群IP,以便實現應用的負載均衡。如上所示為OpenShift將此服務的請求路由到所有標記為app=hello和deploymentconfig=hello的pod。

2.7 檢查pod


[student@workstation ~]$ oc describe pod hello-1-kszfh

2.8 設置外部訪問


使用NodePort方式設置外部訪問。

[student@workstation ~]$ oc edit svc hello

  1 apiVersion: v1
  2 kind: Service
  3 metadata:
  4   annotations:
  5     openshift.io/generated-by: OpenShiftNewApp
  6   creationTimestamp: 2019-07-19T15:50:09Z
  7   labels:
  8     app: hello
  9   name: hello
 10   namespace: network-test
 11   resourceVersion: "24496"
 12   selfLink: /api/v1/namespaces/network-test/services/hello
 13   uid: e348e2a3-aa3c-11e9-b230-52540000fa0a
 14 spec:
 15   clusterIP: 172.30.253.212
 16   ports:
 17   - name: 8080-tcp
 18     port: 8080
 19     protocol: TCP
 20     targetPort: 8080
 21     nodePort: 30800
 22   selector:
 23     app: hello
 24     deploymentconfig: hello
 25   sessionAffinity: None
 26   type: NodePort
 27 status:



[student@workstation ~]$ oc describe svc hello

Name: hello

Namespace: network-test

Labels: app=hello

Annotations: openshift.io/generated-by=OpenShiftNewApp

Selector: app=hello,deploymentconfig=hello

Type: NodePort #驗證是否為NodePort

IP: 172.30.253.212

Port: 8080-tcp 8080/TCP

TargetPort: 8080/TCP

NodePort: 8080-tcp 30800/TCP

Endpoints: 10.128.0.21:8080,10.129.0.37:8080

Session Affinity: None

External Traffic Policy: Cluster

Events: <none>

2.9 驗證外部訪問


[student@workstation ~]$ curl http://node1.lab.example.com:30800

  1 <html>
  2  <head>
  3   <title>PHP Test</title>
  4  </head>
  5  <body>
  6  <br/> Server IP: 10.128.0.21
  7  </body>
  8 </html>



[student@workstation ~]$ curl http://node2.lab.example.com:30800

  1 <html>
  2  <head>
  3   <title>PHP Test</title>
  4  </head>
  5  <body>
  6  <br/> Server IP: 10.129.0.37
  7  </body>
  8 </html>


2.10 使用pod shell


[student@workstation ~]$ oc rsh hello-1-kszfh #使用pod的shell

sh-4.2$ curl http://services.lab.example.com

三 OpenShift router

3.1 OpenShift route概述


OpenShift service允許在OpenShift中的pod之間進行網絡訪問;

OpenShift routes允許從OpenShift外部對pods進行網絡訪問。



路由概念上是通過連接公網IP和DNS主機名訪問內網service IP。在實踐中,為了提高性能和減少延遲,OpenShift route通過OpenShift創建的網絡直接連接到pod,使用該服務只查找endpoint,service只是協助查詢Pod地址。

OpenShift 路由功能由router service提供,該服務在OpenShift實例中作為一個pod運行,可以像任何其他常規pod一樣伸縮和複製。router service基於開源軟件HAProxy實現。

OpenShift route配置的公共DNS主機名需要指向運行router的節點的公共IP地址。route pod與常規應用程序pod不同,它綁定到節點的公共IP地址,而不是內部pod網絡。這通常使用DNS通配符配置。

  1 - apiVersion: v1
  2   kind: Route				#聲明為route類型
  3   metadata:
  4     creationTimestamp: null
  5     labels:
  6       app: quoteapp
  7     name: quoteapp				#路由器名字
  8   spec:
  9     host: quoteapp.apps.lab.example.com	#與route關聯的FQDN,必須預先配置,以解析到OpenShift route pod運行的節點的IP地址
 10     port:
 11       targetPort: 8080-tcp
 12   to:					#一個對象,該對象聲明此route指向的資源類型(在本例中是OpenShift service),以及該資源的名稱(quoteapp)
 13     kind: Service
 14     name: quoteapp



提示:不同資源類型可以使用相同的名稱,如一個名為quoteapp的route可以指向一個名為quoteapp的service。

service通過selector與pod的label進行匹配,router通過name與service的name匹配。

3.2 創建route


創建route最簡單和推薦的方法是使用oc expose命令,將service資源名稱作為輸入參數。–name選項可用於控制route資源的名稱,–hostname選項可用於為route提供自定義主機名。

示例:

[user@demo ~]$ oc expose service quote \

–name quote –hostname=quoteapp.apps.lab.example.com

從模板或不帶–hostname參數的oc expose命令創建的路由,命名方式為:

<route-name>-<project-name>.<default-domain>

解釋

route-name:route的名稱,或原始資源的名稱;

project-name:包含資源的項目的名稱;

default-domain:該值是在OpenShift master上配置的,它對應於作為安裝OpenShift先決條件列出的通配符DNS域。

例如,在OpenShift集群中名為test的project中創建一條名為quote的路由,其中子域為apps.example.com,則FQDN為quote-test.apps.example.com

注意:承載通配符域的DNS服務器不知道任何route的主機名,它只將任何名稱解析為已配置的ip。只有OpenShift route知道route主機名,將每個主機都當作HTTP虛擬主機。無效的通配符域主機名,即不與任何route對應的主機名,將被OpenShift路由器阻塞。

通過向oc create提供JSON或YAML資源定義文件,也可以像其他OpenShift資源一樣創建route資源。

oc new-app命令在從容器映像、Dockerfiles或應用程序源代碼構建pod時不創建route資源。

oc new-app命令不知道pod是否打算從OpenShift實例外部訪問。當oc new-app命令從模板創建一組pod時,沒有什麼可以阻止模板將路由資源包含到應用程序中。

3.3 查找默認subdomain


默認路由子域是在OpenShift配置文件master-config.yaml中的routingConfig字段中定義,使用關鍵字subdomain。

routingConfig:

subdomain: apps.example.com

默認情況下,OpenShift HAProxy route綁定到主機端口80 (HTTP)和443 (HTTPS)。route必須放置在這些端口不使用的節點上。或者,可以通過設置ROUTER_SERVICE_HTTP_PORT和ROUTER_SERVICE_HTTPS_PORT環境變量來配置路由器監聽其他端口.

路由器支持以下協議:

  • HTTP


  • HTTPS with SNI
  • WebSockets
  • TLS with SNI

3.4 routing類型和選項


路由可以是安全的,也可以是非安全的。安全route提供了使用幾種類型的TLS方式來向客戶端提供證書的能力。不安全路由是最容易配置的,因為它們不需要密鑰或證書,但是安全路由會加密進出pod的流量。

在創建安全路由之前,需要生成TLS證書。

示例:如下步驟創建名為test.example.com的路由創建一個簡單的自簽名證書。

  • 使用openssl命令創建私鑰:


[user@demo ~]$ openssl genrsa -out example.key 2048

  • 使用生成的私鑰創建證書籤名請求(CSR):


[user@demo ~]$ openssl req -new -key example.key -out example.csr -subj “/C=US/ST=CA/L=Los Angeles/O=Example/OU=IT/CN=test.example.com”

  • 使用密鑰和CSR生成證書


[user@demo ~]$ openssl x509 -req -days 366 -in example.csr -signkey example.key -out example.crt

  • 當證書準備好時,創建一個edge-terminated的路由


[user@demo ~]$ oc create route edge –service=test \

–hostname=test.example.com \

–key=example.key –cert=example.crt

3.5 通配符子域


wildcard policy允許用戶定義domain中所有主機的route。route可以使用wildcardPolicy字段將wildcard policy指定為其配置的一部分。OpenShift路由器支持通配符路由,通過設置路由器部署配置中的ROUTER_ALLOW_WILDCARD_ROUTES環境變量為true,從而可將wildcardPolicy屬性設置為子域的任何route都由路由器提供服務。路由器根據route的通配符策略暴露相關的service。

示例:如下下示例表示對於三個不同的路由,a.lab.example.com、b.lab.example.com和c.lab.example.com,它們應該路由到一個名為test的OpenShift服務,可以使用通配符策略配置路由。

  • 將路由器作為集群管理用戶處理通配符路由


[user@demo ~]$ oc scale dc/router –replicas=0

[user@demo ~]$ oc set env dc/router ROUTER_ALLOW_WILDCARD_ROUTES=true

[user@demo ~]$ oc scale dc/router –replicas=1

  • 使用通配符策略創建新路由


[user@demo ~]$ oc expose svc test –wildcard-policy=Subdomain \

–hostname=’www.lab.example.com’

3.6 管理route


在master節點上,在default中查找router

[root@master]# oc project default

[root@master]# oc get pods

在master節點上,檢查路由器環境變量,以找到運行在pod中的HAProxy進程的連接參數

[root@master]# oc env pod router-1-32toa –list | tail -n 6

提示:當創建路由器時,STATS_PASSWORD變量中的密碼是隨機生成的。STATS_USERNAME和STATS_PORT變量有固定的默認值,但是它們都可以在路由器創建時更改。

在router運行的節點上,配置firewall-cmd以打開STATS_PORT變量指定的端口。

[root@node ~]# firewall-cmd –permanent –zone=public –add-port=1936

[root@node ~]# firewall-cmd –reload

打開web瀏覽器並訪問HAProxy statistics URL 為 http://nodeIP:STATS_PORT/。

在User Name字段中輸入STATS_USERNAME的值,在Password字段中輸入STATS_PASSWORD的值,然後單擊OK。則會显示的HAProxy metrics頁面。

四 創建Route練習

4.1 前置準備


準備完整的OpenShift集群,參考《003.OpenShift網絡》2.1。

4.2 本練習準備


[student@workstation ~]$ lab secure-route setup #準備本實驗環境

4.3 創建應用


[student@workstation ~]$ oc login -u developer -p redhat https://master.lab.example.com

[student@workstation ~]$ oc new-project secure-route #創建project

[student@workstation ~]$ oc new-app –docker-image=registry.lab.example.com/openshift/hello-openshift –name=hello

[student@workstation ~]$ oc get pods -o wide

NAME READY STATUS RESTARTS AGE IP NODE

hello-1-wwgkr 1/1 Running 0 20s 10.129.0.38 node2.lab.example.com

4.4 創建TLS證書


[student@workstation ~]$ cd /home/student/DO280/labs/secure-route/ #使用環境中的腳本快速創建TLS自簽名證書

[student@workstation secure-route]$ ./create-cert.sh

4.5 創建route


[student@workstation secure-route]$ ll

-rw-r–r–. 1 student student 550 Aug 7 2018 commands.txt

-rwxr-xr-x. 1 student student 506 Jul 19 2018 create-cert.sh

-rw-rw-r–. 1 student student 1224 Jul 20 10:43 hello.apps.lab.example.com.crt

-rw-rw-r–. 1 student student 1017 Jul 20 10:43 hello.apps.lab.example.com.csr

-rw-rw-r–. 1 student student 1679 Jul 20 10:43 hello.apps.lab.example.com.key

[student@workstation secure-route]$ oc create route edge \

> –service=hello –hostname=hello.apps.lab.example.com \

> –key=hello.apps.lab.example.com.key \

> –cert=hello.apps.lab.example.com.crt

4.6 確認驗證


[student@workstation secure-route]$ oc get route

NAME HOST/PORT PATH SERVICES PORT TERMINATION WILDCARD

hello hello.apps.lab.example.com hello 8080-tcp edge None

[student@workstation secure-route]$ oc get route hello -o yaml #以yaml格式查看route

4.7 測試訪問


[student@workstation ~]$ curl http://hello.apps.lab.example.com #以http形式訪問會無法轉發至後端任何pod

  1 ……
  2       <h1>Application is not available</h1>
  3       <p>The application is currently not serving requests at this endpoint. It may not have been started or is still starting.</p>
  4 ……



[student@workstation ~]$ curl -k -vvv https://hello.apps.lab.example.com #以https形式訪問

  1 ……
  2 Hello OpenShift!
  3 * Connection #0 to host hello.apps.lab.example.com left intact
  4 ……


4.8 非安全形式訪問


由於加密的通信在路由器上終止,並且請求使用不安全的HTTP轉發到pods,所以可以使用pod IP地址通過普通HTTP訪問應用程序。為此,請使用oc get pods -o命令中指定的IP地址。

[student@workstation secure-route]$ oc get pod -o wide

NAME READY STATUS RESTARTS AGE IP NODE

hello-1-wwgkr 1/1 Running 0 21m 10.129.0.38 node2.lab.example.com

[root@node1 ~]# curl -vvv http://10.129.0.38:8080


五 OpenShift網絡綜合實驗

5.1 前置準備


準備完整的OpenShift集群,參考《003.OpenShift網絡》2.1。

5.2 本練習準備


[student@workstation ~]$ lab network-review setup

5.3 驗證所需資源


[student@workstation ~]$ oc login -u developer -p redhat \

https://master.lab.example.com

[student@workstation ~]$ oc get pod -o wide

NAME READY STATUS RESTARTS AGE IP NODE

hello-openshift-1-6ls8z 1/1 Running 0 2m 10.128.0.23 node1.lab.example.com

[student@workstation ~]$ oc get svc

NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE

hello-openshift ClusterIP 172.30.124.237 <none> 8080/TCP,8888/TCP 2m

[student@workstation ~]$ oc get route

NAME HOST/PORT PATH SERVICES PORT TERMINATION WILDCARD

hello-openshift hello.apps.lab.example.com hello-opensift 8080-tcp None

5.4 測試訪問


[student@workstation ~]$ curl http://hello.apps.lab.example.com #測試http訪問

  1 ……
  2       <h1>Application is not available</h1>
  3       <p>The application is currently not serving requests at this endpoint. It may not have been started or is still starting.</p>
  4 ……

[root@node1 ~]# curl http://10.128.0.23:8080 #測試使用pod ip訪問

Hello OpenShift!

[root@node1 ~]# curl http://172.30.124.237:8080 #測試使用cluster ip訪問

curl: (7) Failed connect to 172.30.124.237:8080; Connection refused

5.5 TS cluster故障


[student@workstation ~]$ oc describe svc hello-openshift -n network-review



提示:由上可知,沒有endpoint,endpoint是使用selector對pod的label進行匹配。

[student@workstation ~]$ oc describe pod hello-openshift-1-6ls8z #查看pod詳情



故障點:由上可知,Selector的label不一致,則沒有標記為hello_openshift的pod能進行匹配。

[student@workstation ~]$ oc edit svc hello-openshift

  1 ……
  2   selector:
  3     app: hello-openshift
  4     deploymentconfig: hello-openshift
  5   sessionAffinity: None
  6 ……


5.6 測試訪問


[root@node1 ~]# curl http://10.128.0.23:8080 #測試使用pod ip訪問

Hello OpenShift!

[root@node1 ~]# curl http://172.30.124.237:8080 #再次測試

Hello OpenShift!

[student@workstation ~]$ curl http://hello.apps.lab.example.com #測試http訪問

  1 ……
  2       <h1>Application is not available</h1>
  3       <p>The application is currently not serving requests at this endpoint. It may not have been started or is still starting.</p>
  4 ……


5.7 TS route故障


[student@workstation ~]$ oc describe route hello-openshift



故障點:由上可知,此路由沒有endpoint。即對route的URL請求沒有後端endpoint進行響應。路由器查詢service的endpoint,並註冊有效的endpoint來實現負載平衡。同時發現service名稱中有一個拼寫錯誤,它應該是hello-openshift。

[student@workstation ~]$ oc edit route hello-openshift

  1 ……
  2 spec:
  3   host: hello.apps.lab.example.com
  4   port:
  5     targetPort: 8080-tcp
  6   to:
  7     kind: Service
  8     name: hello-openshift
  9     weight: 100
 10   wildcardPolicy: None
 11 ……



[root@node1 ~]# curl http://hello.apps.lab.example.com #再次測試

Hello OpenShift!

5.8 確認驗證


[student@workstation ~]$ lab network-review grade #使用腳本判斷 本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

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

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

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

聚甘新

年僅 28 歲就宣布從字節跳動退休?

這两天,互聯網熱議最大的一個話題除了阿里 P8 程序員找生活助理的事,另外一個就是 28 歲的郭宇宣布從字節跳動退休,稱選擇經營溫泉旅行,選擇成為一名職業作家。

我看到這個話題的時候,情不自禁地“嘖嘖”了兩聲,真特么酸了,28 歲就退休了,我已經 31 歲了,還在辛苦打拚的路上,除了要忙工作,還要高產似母豬地更文,然而,即便我這麼努力,還是沒能成為一名“職業作家”,退休更是遙遙無期。

郭大佬非常牛逼的一點在於,高考之後就開始敲代碼,上了大學之後依然敲代碼,大三就在支付寶干過,然後創業的一家公司被字節跳動收購,再然後嘛,就財務自由退休了——28 歲,重新定義了退休的年紀。

字節跳動這家公司發展的真的是巨快,明星產品今日頭條和抖音,真的是國民級應用。反正我父母都是這兩款產品的忠實用戶,我妹妹雖然不玩今日頭條,但抖音玩得那叫一個熱火朝天。

我自己是不玩抖音也不看今日頭條的,因為覺得這種短視頻,或者說亂七八糟的新聞熱點有點浪費生命的感覺,所以一直很抵觸。

當然了,我如果說我一次也沒玩過,有點聖人的感覺,我做不到。但每次無聊到刷上倆小時的抖音,我就會噁心到把這款軟件卸載掉。尤其是聽到那些無厘頭的狂笑,我感覺到娛樂在致死。

這不是抖音的問題,是我的問題,是人性的問題,抖音就抓住了人性的弱點,讓你沉浸其中,忘乎所以。

抨擊歸抨擊,但我不能忽視的事實是,字節跳動是真的牛逼,郭大佬是真的有錢了。

每個人都有自己人生,郭大佬有實力又有運氣,他過的是一種極致的人生。

我是 2014 年回的洛陽,一回來就跟着一個老闆做創業項目,依稀還記得當初他給我許下的承諾:三年後讓你在洛陽買房買車,五年後帶你走上人生巔峰。

2015 年,我買了房,靠的是我和老婆辛苦攢下的一些積蓄,還有父母義無反顧的支持。老闆也借給了我兩三萬,一年後我就還他了,所以在買房這個承諾上,他有幫助,但遠非承諾中的那樣。

2016 年,我買了車,分期付款的那種,和老闆沒有一點關係。

至於五年後走上人生巔峰,更是瞎扯淡。我現在還是一名普通的程序員,生活的幸福指數也完全靠的是自己的付出。

這些年裡,老闆無數次胯下海口,聽得我耳朵都膩了。至於我為什麼還沒有離職,並不是我沉浸在溫柔故鄉,而是洛陽的軟件環境整體就這麼個樣,去哪都是打工,還不如自己踏踏實實做一些事情,比如說寫作。

對比我倆,就會發現一些很有意思的點,我來給同學們剖析一下。

1)學歷重不重要

很重要,郭大佬讀過深圳高級中學,深圳最好的高中之一,大學是暨南大學,211。

我呢,高中雖然是保送的,但那時候的學校已經走了下坡路,很動蕩,師資和校領導換了好幾波;然後我上的是一所大專。

所以我大學那會很自卑,即便專業是計算機網絡,也沒多少心思學習。而郭大佬就完全不一樣了,沉下一門心思學編程,為此還掛科了好幾門。因為他是非科班出身,專業是政治與行政管理。

假如,請允許我假如一下,給低學歷的同學們一點點信心。

假如我上大學那會一門心思撲倒編程上,大三也不至於出去參加培訓,真的,大把大把的時光我都浪費了。除了談戀愛是正事,我就只會打遊戲了。

假如拿現在的心態去過大學兩年的時光,我堅信,我一定能進阿里,因為拼過和沒拼過的人生差別巨大。

我就認識這樣一個初中小妹妹,平常老喊我二叔,搞得我都逆來順受了。她的成績非常優異,全年級第二名,為什麼不是第一名,因為語文成績拖了後腿。這不是關鍵,關鍵是小妹妹現在就開始學編程了,還去給初一的同學授過課。

後生可畏。

所以,我的結論就是,能通過學習改變命運,就下勁學,錯過這個年紀就真的沒機會了。如果真的上了大專,上了一般的本科,也不是沒有機會,別整天喊自己迷茫,誰的青春沒有迷茫過,關鍵是要發掘自己的興趣點,如果要從事程序員這個行業,就好好學編程。

2)要不要創業

十個創業九個坑,我只能這麼說,能進大廠進大廠,進不去進中廠,進不去中廠進小廠。如果非要創業,也得你自身實力夠硬,假如創業失敗,你還有出路,或者實在是沒有其他更好的選擇了,再選擇創業。

職場新人最好不要被忽悠去創業,太慘了。

你看,人家郭大佬在支付寶鍍了一層金,然後所在公司稀里糊塗被字節跳動收購了,這是運氣,沒得說。

大部分人的命運可能像我一樣,在日企待過三年半,有了一些資歷,然後作為技術大拿參与到創業公司,一開始老闆牛逼吹上天,最後,啥也沒撈着。

青春荒廢了,人際關係荒廢了,程序員的黃金年齡段也荒廢了。

3)要不要提前退休

有不少同學問過我這樣的話題,“二哥,我馬上到了結婚的年紀,雖然在一線城市掙得還可以,但遠沒到能買得起房子的水準,可能這輩子都不可能了,我想現在回二線城市或者三線城市,你看可行嗎?”

這種想法,其實就和郭大佬退休的想法是一致的,只不過郭財務自由了。

對於普通人來說,我的建議是這樣的,請認認真真做好筆記。

第一,不要盲目回二三線。

拿洛陽來說吧,一般程序員的極限工資就是一萬塊,撐死的那種。五險一金,包括獎金,能沒有公司就考慮沒有。

捫心自問一下,自己能否承受得起這份清心寡慾。另外,二三線城市也是會加班的,關鍵是不一定有加班工資。

第二,搞一份副業吧,同學們。

在一線城市,你可能沒有精力和時間搞副業,另外,主業的成長潛力並不比副業差,搞的意義不是特別大。但如果要回二三線,副業必須得搞,哪怕掙個零花錢,心裏不慌。

幸福指數高不高,離不開錢那,雖然很俗。粗茶淡飯沒問題,二三線城市的生活成本相對較低是真的,但你的掙錢能力也得匹配上吧,匹配不上的話,活得就會很累的。

我羡慕郭大佬,有些同學可能羡慕我,覺得我的幸福指數也很高。那我要告訴你的真相就是,我既要忙工作,還要讀書寫作,也是很拼的。

人生就是這樣,為別人的成功送上祝福的同時,不要忘記腳踏實地地活着。

如果覺得文章對你有點幫助,請微信搜索「 沉默王二 」第一時間閱讀。

本文已收錄 GitHub,傳送門~ ,裏面更有大廠面試完整考點,歡迎 Star。

我是沉默王二,一枚有顏值卻靠才華苟且的程序員。關注即可提升學習效率,別忘了三連啊,點贊、收藏、留言,我不挑,嘻嘻

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

【其他文章推薦】

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

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

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

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

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

聚甘新