從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地圖已可更新顯示潭子電動車充電站設置地點!!

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

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

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

※聚甘新

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

元素添加初始參數

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

添加初始數據，默認透明度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 .....");
    }
}

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

【其他文章推薦】

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

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

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

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

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

※聚甘新

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會假設地球為完美的球形, 所以可能有一些位置計算偏差，據說<=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加入key為cities:locs的 sorted 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 半徑長度，必選項。後面的m、km、ft、mi、是長度單位選項，四選一。
• 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中的WITHCOORD、COUNT、ASC等，例如我們需要在返回結果中包含坐標、中心距離、由近到遠排序的前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 刪除元素

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

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

4. 總結

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

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

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

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

【其他文章推薦】

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

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

※超省錢租車方案

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

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

※聚甘新

原文鏈接：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的動態。

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

【其他文章推薦】

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

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

※回頭車貨運收費標準

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

※超省錢租車方案

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

※聚甘新

調試是應用程序開發周期不可或缺的一部分。用Java或任何其他語言編寫程序時，每個開發人員應解決的首要問題之一是可靠的調試工具的可用性。

所使用的工具類型可能影響或破壞應用程序的調試過程，因此至關重要的是，要了解根據用例而定最佳選擇。

在這篇文章中，我們概述了2020年最好的7種Java調試工具。在開發，生產環境中查找，診斷和修復問題時，這些工具中的大多數將派上用場。

NetBeans

NetBeans是運行在Linux，Windows，MacOS和Solaris上的頂級，使用最廣泛的Java IDE之一。正如預期的那樣，它具有可視化調試器和代碼概要分析器，使開發人員可以調試可執行的Java類，單元測試和整個項目。

NetBeans調試器允許您在Java代碼中放置斷點，運行方法，添加字段監視，監視執行以及在調試會話期間拍攝快照。

Rookout

Rookout是一個很棒的Java調試選項，可以在開發和生產中很好地工作。它在包括無服務器和容器的各種環境中提供了強大的調試功能。

通過收集和流水線化關鍵數據，淘汰工作超越了標準調試功能。這使開發人員無需編寫代碼，重新部署或重新啟動應用程序即可了解軟件執行問題並解決錯誤。

藉助Rookout，開發人員可以消除冗長，複雜且資源密集的數據探索和錯誤查找過程。

Eclipse

Eclipse是帶有內置Java調試器的著名開源IDE。自成立以來，Eclipse一直保持其作為開發現代應用程序最強大的跨平台IDE之一的聲譽。

它提供了標準的調試功能，例如設置斷點，執行步驟執行，檢查變量和值，掛起和恢複線程等功能。

Eclipse平台還方便了遠程調試。儘管它主要是Java IDE，但Eclipse調試視圖還支持PHP，C，C ++和許多其他編程語言。

IntelliJ IDEA

IntelliJ IDEA是具有功能強大的調試器的高度流行的Java IDE。該工具使開發人員可以輕鬆調試簡單代碼以及多線程Java應用程序。

使用IntelliJ調試器，您可以設置斷點，單步執行代碼，評估表達式，檢查變量以及執行一系列其他調試過程。它可以更輕鬆地檢測意外的流量和狀態，死鎖，活動鎖等。

IntelliJ IDEA的核心旨在改善Java開發團隊的工作流程和生產力。

Java調試器（JDB）

Java調試器（JDB）是允許開發人員在命令行中調試Java代碼的工具。通過Java調試接口（JDI）（高級前端接口），開發人員可以檢測並修復程序中的錯誤。該工具還可用於檢查和調試遠程Java虛擬機中的代碼。

像大多數命令行調試器一樣，JDB具有學習曲線，因此新用戶需要花費一些時間來適應JDB。但是，一旦掌握了JDB命令，就可以輕鬆設置斷點，單步執行代碼並執行其他調試操作。

Fusion Reactor(聚變反應堆??)

Fusion Reactor是針對開發，測試和生產環境中的Java應用程序的創新性能監視解決方案。該工具配備了一組令人印象深刻的功能，這些功能可提供Java開發人員在APM工具中所需的一切。

Fusion Reactor開發版，您可以開發，測試，並在非生產環境分析應用。使用此工具，在將應用程序部署到生產環境之前，更容易發現問題並提高代碼質量。

另一個值得注意的功能是生產調試器，它使開發人員在與代碼交互並修復錯誤時獲得最大的控制權。Fusion Reactor還支持遠程調試。

JDeveloper

Oracle的JDeveloper是一種免費的IDE，可解決應用程序開發生命周期中從編碼到設計，性能分析，調試，優化和部署的每個步驟。

使用JDeveloper進行調試時，可以設置斷點和觀察點，分析調用堆棧，檢查和操作變量，並逐步研究代碼執行情況。除了Java，它還可以用於調試HTML，PHP，JavaScript，SQL和XML。

現在，您可以繼續使用上述工具，以更高的速度和效率來檢測，診斷和解決Java應用程序中的問題。

翻譯原文

Top Java Debugging Tools for 2020

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

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

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

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

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

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

※超省錢租車方案

※聚甘新