分類: 3C資訊

MySQL 5.7 – 通過 BINLOG 恢複數據
日常開發，運維中，經常會出現誤刪數據的情況。誤刪數據的類型大致可分為以下幾類：
- 使用 delete 誤刪行
- 使用 drop table 或 truncate table 誤刪表
- 使用 drop database 語句誤刪數據庫
- 使用 rm 命令誤刪整個 MySQL 實例。
不同的情況，都會有其優先的解決方案：
- 針對誤刪行，可以通過 Flashback 工具將數據恢復
- 針對誤刪表或庫，一般採用通過 BINLOG 將數據恢復。
- 而對於誤刪 MySQL 實例，則需要我們搭建 HA 的 MySQL 集群，並保證我們的數據跨機房，跨城市保存。
本篇主要討論的內容是誤刪表或者庫，會先介紹有關 BINLOG 的操作命令，然後會對誤刪表的這種情況進行實際的模擬。

BINLOG 常見操作命令

BINLOG 的查詢方式一般分為兩種，一種是進入 MySQL 控制台進行查詢，另一種是通過 MySQL 提供的工具 mysqlbinlog 進行查詢，兩者的不同會在下面介紹。

通過 MySQL Cli 查詢 BINLOG 信息

在 cli 中，常見的命令如下：
```
# 查詢 BINLOG 格式
show VARIABLES like 'binlog_format';

# 查詢 BINLOG 位置
show VARIABLES like 'datadir';

# 查詢當前數據庫中 BINLOG 名稱及大小
show binary logs;

# 查看 master 正在寫入的 BINLOG 信息
show master status\G;

# 通過 offset 查看 BINLOG 信息
show BINLOG events in 'mysql-bin.000034' limit 9000,  10;

# 通過 position 查看 binlog 信息
show BINLOG events in 'mysql-bin.000034' from 1742635 limit 10;
```
使用 show BINLOG events 的問題：
- 使用該命令時，如果當前 binlog 文件很大，而且沒有指定 limit，會引發對資源的過度消耗。因為 MySQL 客戶端需要將 binlog 的全部內容處理，返回並显示出來。為了防止這種情況，mysqlbinlog 工具是一個很好的選擇。
通過 mysqlbinlog 查詢 BINLOG 信息

在介紹 mysqlbinlog 工具使用前，先來看下 BINLOG 文件的內容：
```
# 查詢 BINLOG 的信息
mysqlbinlog  --no-defaults mysql-bin.000034 | less
```
```
# at 141
#100309  9:28:36 server id 123  end_log_pos 245
  Query thread_id=3350  exec_time=11  error_code=0
```
- at 表示 offset 或者說事件開始的起始位置
- 100309 9:28:36 server id 123 表示 server 123 開始執行事件的日期
- end_log_pos 245 表示事件的結束位置 + 1，或者說是下一個事件的起始位置。
- exec_time 表示在 master 上花費的時間，在 salve 上，記錄的時間是從 Master 記錄開始，一直到 Slave 結束完成所花費的時間。
- rror_code=0 表示沒有錯誤發生。
在大致了解 binlog 的內容后，mysqlbinlog 的用途有哪些？：
- mysqlbinlog 可以作為代替 cli 讀取 binlog 的工具。
- mysqlbinlog 可以將執行過的 SQL 語句輸出，用於數據的恢復或備份。
查詢 BINLOG 日誌：
```
# 查詢規定時候后發生的 BINLOG 日誌
mysqlbinlog --no-defaults --base64-output=decode-rows -v  --start-datetime  "2019-11-22 14:00:00" --database sync_test  mysql-bin.000034 | less
```
導出 BINLOG 日誌，用於分析和排查 sql 語句：
```
mysqlbinlog --no-defaults --base64-output=decode-rows -v  --start-datetime  "2019-11-22 14:00:00" --database sync_test  mysql-bin.000034 > /home/mysql_backup/binlog_raw.sql
```
導入 BINLOG 日誌
```
# 通過 BINLOG 進行恢復。
mysqlbinlog --start-position=1038 --stop-position=1164 --database=db_name  mysql-bin.000034 | mysql  -u cisco -p db_name

# 通過 BINLOG 導出的 sql 進行恢復。
mysql -u cisco -p db_name < binlog_raw.sql.sql
```
mysqlbinlog 的常用參數：
- --database 僅僅列出配置的數據庫信息
- --no-defaults 讀取沒有選項的文件, 指定的原因是由於 mysqlbinlog 無法識別 BINLOG 中的 default-character-set=utf8 指令
- --offset 跳過 log 中 N 個條目
- --verbose 將日誌信息重建為原始的 SQL 陳述。
  - -v 僅僅解釋行信息
  - -vv 不但解釋行信息，還將 SQL 列類型的註釋信息也解析出來
- --start-datetime 显示從指定的時間或之後的時間的事件。
  - 接收 DATETIME 或者 TIMESTRAMP 格式。
- --base64-output=decode-rows 將 BINLOG 語句中事件以 base-64 的編碼显示，對一些二進制的內容進行屏蔽。
  - AUTO 默認參數，自動显示 BINLOG 中的必要的語句
  - NEVER 不會显示任何的 BINLOG 語句，如果遇到必須显示的 BINLOG 語言，則會報錯退出。
  - DECODE-ROWS 显示通過 -v 显示出來的 SQL 信息，過濾到一些 BINLOG 二進制數據。
MySQL Cli 和 mysqlbinlog 工具之間的比較

如果想知道當前 MySQL 中正在寫入的 BINLOG 的名稱，大小等基本信息時，可以通過 Cli 相關的命令來查詢。

但想查詢，定位，恢復 BINLOG 中具體的數據時，要通過 mysqlbinlog 工具，因為相較於 Cli 來說，mysqlbinlog 提供了 --start-datetime，--stop-position 等這樣更為豐富的參數供我們選擇。這時 Cli 中 SHOW BINLOG EVENTS 的簡要語法就變得相形見絀了。

使用 BINLOG 恢複數據

恢復的大致流程如下：
1. 會創建數據庫和表，並插入數據。
2. 誤刪一條數據。
3. 繼續插入數據。
4. 誤刪表。
5. 最後將原來以及之後插入的數據進行恢復。
準備數據

準備數據庫，表及數據：
```
# 創建臨時數據庫
CREATE DATABASE IF NOT EXISTS test_binlog default charset utf8 COLLATE utf8_general_ci; 


# 創建臨時表
CREATE TABLE `sync_test` (`id` int(11) NOT NULL AUTO_INCREMENT, `name` varchar(255) NOT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8;


# 添加數據
insert into sync_test (id, name) values (null, 'xiaoa');
insert into sync_test (id, name) values (null, 'xiaob');
insert into sync_test (id, name) values (null, 'xiaoc');

# 查看添加的數據
select * from sync_test;
```
刪除表或者數據

誤刪操作：
```
# 刪除 name=xiaob 的數據
delete from sync_test where id=3

# 插入幾條數據
insert into sync_test (id, name) values (null, 'xiaod');
insert into sync_test (id, name) values (null, 'xiaoe');
insert into sync_test (id, name) values (null, 'xiaof');

# 刪除表
DROP TABLE sync_test;
```
數據的恢復

在執行數據恢復前，如果操作的是生產環境，會有如下的建議：
- 使用 flush logs 命令，替換當前主庫中正在使用的 binlog 文件，好處如下：
  - 可將誤刪操作，定位在一個 BINLOG 文件中，便於之后的數據分析和恢復。
  - 避免操作正在被使用的 BINLOG 文件，防止發生意外情況。
- 數據的恢復不要在生產庫中執行，先在臨時庫恢復，確認無誤后，再倒回生產庫。防止對數據的二次傷害。
通常來說，恢復主要有兩個步驟：
1. 在臨時庫中，恢復定期執行的全量備份數據。
2. 然後基於全量備份的數據點，通過 BINLOG 來恢復誤操作和正常的數據。
使用 BINLOG 做數據恢復前：
```
# 查看正在使用的 Binlog 文件
show master status\G;
# 显示結果是： mysql-bin.000034

# 執行 flush logs 操作，生成新的 BINLOG
flush logs;

# 查看正在使用的 Binlog 文件
show master status\G;
# 結果是：mysql-bin.000035
```
確定恢複數據的步驟：

這裏主要是有兩條誤刪的操作，數據行的誤刪和表的誤刪。有兩種方式進行恢復。
- 方式一：首先恢復到刪除表操作之前的位置，然後再單獨恢復誤刪的數據行。
- 方式二：首先恢復到誤刪數據行的之前的位置，然後跳過誤刪事件再恢複數據表操作之前的位置。
這裏採用方式一的方案進行演示，由於是演示，就不額外找一個臨時庫進行全量恢復了，直接進行操作。

查詢創建表的事件位置和刪除表的事件位置
```
#  根據時間確定位置信息
mysqlbinlog --no-defaults --base64-output=decode-rows -v  --start-datetime  "2019-11-22 14:00:00" --database test_binlog  mysql-bin.000034 | less
```
創建表的開始位置：

刪除表的結束位置：

插入 name=’xiaob’ 的位置：
```
# 根據位置導出 SQL 文件
mysqlbinlog --no-defaults --base64-output=decode-rows -v --start-position "2508132" --stop-position "2511004" --database test_binlog  mysql-bin.000034 > /home/mysql_backup/test_binlog_step1.sql
 
 
mysqlbinlog --no-defaults --base64-output=decode-rows -v --start-position "2508813" --stop-position "2509187" --database test_binlog  mysql-bin.000034 > /home/mysql_backup/test_binlog_step2.sql
 

# 使用 mysql 進行恢復
mysql -u cisco -p < /home/mysql_backup/test_binlog_step1.sql
mysql -u cisco -p < /home/mysql_backup/test_binlog_step2.sql
```
MySQL 5.7 中無論是否打開 GTID 的配置，在每次事務開啟時，都首先會出 GTID 的一個事務，用於并行複製。所以在確定導出開始事務位置時，要算上這個事件。

在使用 –stop-position 導出時，會導出在指定位置的前一個事件，所以這裏要推后一個事務。

對於 DML 的語句，主要結束位置要算上 COMMIT 的位置。

總結

在文章開始時，我們熟悉了操作 BINLOG 的兩種方式 CLI 和 mysqlbinlog 工具，接着介紹了其間的區別和使用場景，對於一些大型的 BINLOG 文件，使用 mysqlbinlog 會更加的方便和效率。並對 mysqlbinlog 的一些常見參數進行了介紹。

接着通過使用 mysqlbinlog 實際模擬了數據恢復的過程，並在恢複數據時，提出了一些需要注意的事項，比如 flush logs 等。

最後在恢複數據時，要注意 start-position 和 end-position 的一些小細節，來保證找到合適的位置。

參考

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

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

※網頁設計一頭霧水??該從何著手呢? 找到專業技術的網頁設計公司，幫您輕鬆架站!

※想要讓你的商品成為最夯、最多人討論的話題?網頁設計公司讓你強力曝光

※想知道最厲害的台北網頁設計公司推薦、台中網頁設計公司推薦專業設計師”嚨底家”!!
2020-02-14
網絡相關的命令工具研究報告

主機配置：DHCP

　　DHCP（動態主機配置協議），是在一台主機啟動后，第一個運行的客戶/服務器應用程序。換言之，當一台主機啟動后，如果它認為自己當前應當連接到因特網上，但又不知道自己的IP地址時，DHCP就以引導程序的身份發揮作用。

　　每個連接到TCP/IP互聯網的計算機都必須知道自己的IP地址、一個路由器的IP地址、一個名字服務器的IP地址以及自己的子網掩碼這四種信息。

DHCP分組格式：

一、曾經使用過的協議

　　在DHCP成為正式的主機配置協議之前，還有過一些其他的協議。

1.RARP：

　　在因特網時代的初期，人們曾設計了一個稱為逆地址解析協議（Reverse Address Resolution Protocol，RARP）來向被引導的主機提供IP地址。實際上，RARP是ARP的一個版本。ARP將一個IP地址映射為一個物理地址，而RARP則將一個物理地址映射成為一個IP地址。但是RARP已經被淘汰了，原因有兩個：首先，RARP利用了數據鏈路層的廣播服務，這也就表示每個網絡上都必須存在一台RARP服務器。第二，RARP只能提供計算機的IP地址，但如今的計算機需要前面提到的所有四種信息。

2.BOOTP：

　　引導程序協議（BOOTstrap Protocol，BOOTP）是DHCP的先驅。它是一個客戶/服務器協議，被設計用來克服RARP協議存在的缺陷。但是BOOTP是一個靜態配置協議，當客戶請求自己的IP地址時，BOOTP服務器就諮詢一張表，將客戶的物理地址映射成相應的IP地址。這就意味着客戶的物理地址和IP地址之間的綁定是已經存在的。這個綁定關係是事先設定好的。

　　在某些場合，我們需要的是一個動態配置協議。例如，當一台主機從一個物理網絡移動到另一個物理網絡時，它的物理地址就改變了。再比如，有時候主機需要在某一段時間內使用一個臨時的IP地址。BOOTP無法處理這種狀況，因為物理地址和IP地址之間的綁定是靜態的，是固定存放在一張表中的，除非管理員更改這張表。

　　而DHCP的設計就是為了解決這些不足之處。

3. DHCP：

　　動態主機配置協議（Dynamic Host Configuration Protocol，DHCP）是一種客戶/服務器協議，設計這個協議是為了將上述四種信息傳遞給無盤計算機或者第一次啟動的計算機。DHCP是BOOTP的繼承者，並且能夠兼容BOOTP。

二、DHCP操作

　　DHCP客戶和DHCP服務器可以在同一個網絡上，也可以位於不同的網絡。

1.DHCP客戶和DHCP服務器在同一個網絡

　　雖然這種情況不是很常見，不過管理員可以把客戶和服務器放在同一個網絡中。如圖所示：

這種情況的操作如下：

（1）DHCP服務器在UDP端口67發出被動打開命令，等待客戶請求。

（2）被引導的客戶在UDP端口68發出主動打開命令。這個報文被封裝成UDP用戶報，其目的端口是67，源端口號是68。這個UDP用戶數據報在封裝成IP數據包。客戶使用的是全0的源地址和全1的目的地址。

（3）服務器或者用廣播報文，或者用單播報文來響應這個用戶，它使用了UDP源端口號67和目的端口68.這個響應可以是單播的，因為服務器知道客戶的IP地址，同時也知道客戶的物理地址，也就是說它不需要使用ARP的服務進行從邏輯地址到物理地址的映射。但是某些系統不允許旁路掉ARP，結果就要使用廣播地址。

2. DHCP客戶和DHCP服務器在不同的網絡

如圖所示：

　　像其他應用層的進程一樣，客戶可以在某個網絡上，而服務器可以在相隔好幾個網絡之外的另一網絡上。這就帶來了一個必須要解決的問題。DHCP請求是廣播發送的，因為客戶不知道服務器的IP地址。而廣播的IP數據報不能通過任何路由器。路由器收到這樣的分組就丟棄它。

　　要解決這個問題，就需要一个中介物。某台主機（或是一台能夠配置為在應用層工作的路由器）可以用來充當中繼。在這種情況下，該主機就稱為中繼代理。中繼代理知道DHCP服務器的單播地址，並在端口67監聽廣播報文。當它收到這種類型的分組后，就把它封裝成一個單播數據報，並且把此請求發送給DHCP服務器。攜帶了單播目的地址的分組可以被任何一個路由器轉發，最終到達DHCP服務器。DHCP服務器知道這個報文來自中繼代理，因為在請求報文中有一個字段定義了中繼代理的IP地址。中繼代理在收到回答后，再把它發送給DHCP客戶。

三、配置

　　人們設計DHCP是為了提供靜態和動態的地址分配。

1.靜態地址分配

　　對於靜態地址分配，DHCP有一個專門的數據庫，可以靜態地吧物理地址綁定到IP地址。

2.動態地址分配

　　DHCP還有第二個數據庫，包括一個可用的IP地址池。第二個數據庫使DHCP成為動態的。當DHCP客戶請求臨時的IP地址時，DHCP服務器就從可用（即為使用的）IP地址池中取出一個IP地址進行指派，這個IP地址的使用時間長短可協商。

　　當DHCP客戶想DHCP服務器發送請求是，服務器首先檢查它的靜態數據庫。若靜態數據庫中存在所請求物理地址的表項，則返回給這個客戶的永久IP地址。反之，若靜態數據庫中沒有這個表項，服務器就從可用IP地址池中選擇一個IP地址，並把這個地址指派給客戶，然後再把相應的表項加入到動態數據庫中。

　　如果主機要從一個網絡移動到另一個網絡，或者與一個網絡時連時斷，那麼DHCP的這種動態特性就有了用武之地。DHCP可以在有限時間內提供一個臨時的IP地址。

從地址池指派的地址都是臨時地址。DHCP服務器向客戶授予某一段時間內對該地址池的租用權。當租用時效過期，客戶或者停止使用這個IP地址，或者續租。服務器有權力選擇同意或不同意續租。若服務器不同意，客戶就停止使用這個地址。

3.轉換狀態

　　為了提供動態的地址分配，DHCP客戶可以像狀態機那樣從一個狀態轉換到另一個狀態，狀態轉換取決於收到的報文和發送的報文。在這種情況下，報文的類型是由包含在DHCP分組中的標記為53的選項來定義的。標記為53 的選項如圖所示：

DHCP的不同狀態：

（1）INIT狀態

　　當DHCP客戶首次啟動時，它處於INIT狀態（初始化狀態）。客戶使用端口67廣播DHCPDISCOVER報文（一個帶有DHCPDISCOVER選項的請求報文）。

（2）SELECTING狀態

　　在發送DHCPDISCOVER報文後，客戶就進入SELECTING（選擇）狀態。能夠提供這種類型服務的服務器要用DHCPOFFER報文進行相應。在此類報文中，服務器提供了一個IP地址。它們還要提供租用時間長度，其默認值是1小時。在發送DHCPOFFER報文的服務器，把提供的IP地址鎖定，使這個地址不會再提供給任何其他的客戶。客戶選擇所提供的地址中的一個，並向所選擇的服務器發送DHCPREQUEST報文。然後就進入REQUESTING狀態。如果客戶沒有收到DHCPOFFER報文，它還要再嘗試四次，每一次間隔2秒。如果對這些DHCPDISCOVER都沒有收到回答，客戶就睡眠5分鐘后再試。

（3）REQUESTING狀態

　　客戶保持在這個REQUESTING（請求）狀態，直至它收到來自服務器的DHCPACK報文為止，這個報文創建了客戶物理地址和它的IP地址之間的綁定。客戶收到DHCPACK報文後進入BOUND狀態。

（4）BOUND狀態

　　在這種狀態下，客戶可以使用該IP地址，直到租用時間到期。當到達租用時間的50%時，客戶就再發送一個DHCPREQUEST報文以請求更新。於是，客戶進入RENEWING。在BOUND（綁定）狀態時，客戶也可以取消租用，並進入到初始化狀態。

（5）RENEWING狀態

　　客戶保持在RENEWING（更新）狀態，直至下面兩個事件之一發生。客戶可以收到更新租用協定的DHCPACK報文。在這種情況下，客戶把計時器複位，然後回到BOUND狀態。或者，如果沒有收到DHCPACK，同時到達租用時間的87.5%時，客戶就進入REBINDING狀態。

（6）REBINDING狀態

　　客戶保持在REBINDING（重新綁定）狀態，直至下面三個事件之一發生。若客戶收到一個DHCPNACK報文或者租用時間到期，則回到初始化狀態，並嘗試得到另一個IP地址。若客戶收到DHCPACK報文，它就進入綁定狀態，並把計時器複位。

DHCP不同狀態的轉換圖：

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

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

※評比前十大台北網頁設計、台北網站設計公司知名案例作品心得分享

※智慧手機時代的來臨，RWD網頁設計已成為網頁設計推薦首選

2020-02-14

MySQL 複製表結構和表數據

1、前言

　　在功能開發完畢，在本地或者測試環境進行測試時，經常會遇到這種情況：有專門的測試數據，測試過程會涉及到修改表中的數據，經常不能一次測試成功，所以，每次執行測試后，原來表中的數據其實已經被修改了，下一次測試，就需要將數據恢復。

　　我一般的做法是：先創建一個副本表，比如測試使用的user表，我在測試前創建副本表user_bak，每次測試后，將user表清空，然後將副本表user_bak的數據導入到user表中。

　　上面的操作是對一個table做備份，如果涉及到的table太多，可以創建database的副本。

　　接下來我將對此處的表結構複製以及表數據複製進行闡述，並非數據庫的複製原理！！！！

　　下面是staff表的表結構

create table staff (
	id int not null auto_increment comment '自增id',
	name char(20) not null comment '用戶姓名',
	dep char(20) not null comment '所屬部門',
	gender tinyint not null default 1 comment '性別：1男; 2女',
	addr char(30) not null comment '地址',
	primary key(id),
	index idx_1 (name, dep),
	index idx_2 (name, gender)
) engine=innodb default charset=utf8mb4 comment '員工表';

2、具體方式　

　　2.1、執行舊錶的創建SQL來創建表

　　如果原始表已經存在，那麼可以使用命令查看該表的創建語句：

mysql> show create table staff\G
*************************** 1. row ***************************
       Table: staff
Create Table: CREATE TABLE `staff` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `name` char(20) NOT NULL COMMENT '用戶姓名',
  `dep` char(20) NOT NULL COMMENT '所屬部門',
  `gender` tinyint(4) NOT NULL DEFAULT '1' COMMENT '性別：1男; 2女',
  `addr` char(30) NOT NULL,
  PRIMARY KEY (`id`),
  KEY `idx_1` (`name`,`dep`),
  KEY `idx_2` (`name`,`gender`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='員工表'
1 row in set (0.01 sec)

　　可以看到，上面show creat table xx的命令執行結果中，Create Table的值就是創建表的語句，此時可以直接複製創建表的SQL，然後重新執行一次就行了。

　　當數據表中有數據的時候，看到的創建staff表的sql就會稍有不同。比如，我在staff中添加了兩條記錄：

mysql> insert into staff values (null, '李明', 'RD', 1, '北京');
Query OK, 1 row affected (0.00 sec)

mysql> insert into staff values (null, '張三', 'PM', 0, '上海');
Query OK, 1 row affected (0.00 sec)

mysql> select * from staff;
+----+--------+-----+--------+--------+
| id | name   | dep | gender | addr   |
+----+--------+-----+--------+--------+
|  1 | 李明   | RD  |      1 | 北京   |
|  2 | 張三   | PM  |      0 | 上海   |
+----+--------+-----+--------+--------+
2 rows in set (0.00 sec)

　　此時在執行show create table命令：

mysql> show create table staff\G
*************************** 1. row ***************************
       Table: staff
Create Table: CREATE TABLE `staff` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `name` char(20) NOT NULL COMMENT '用戶姓名',
  `dep` char(20) NOT NULL COMMENT '所屬部門',
  `gender` tinyint(4) NOT NULL DEFAULT '1' COMMENT '性別：1男; 2女',
  `addr` char(30) NOT NULL,
  PRIMARY KEY (`id`),
  KEY `idx_1` (`name`,`dep`),
  KEY `idx_2` (`name`,`gender`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 COMMENT='員工表'
1 row in set (0.00 sec)

　　注意，上面結果中的倒數第二行

　　　　ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 COMMENT=’員工表’

　　因為staff表的id是自增的，且已經有了2條記錄，所以下一次插入數據的自增id應該為3，這個信息，也會出現在表的創建sql中。

　　2.2、使用like創建新表（僅包含表結構）

　　使用like根據已有的表來創建新表，特點如下：

　　1、方便，不需要查看原表的表結構定義信息；

　　2、創建的新表中，表結構定義、完整性約束，都與原表保持一致。

　　3、創建的新表是一個空表，全新的表，沒有數據。

　　用法如下：

mysql> select * from staff;　　#舊錶中已有2條數據
+----+--------+-----+--------+--------+
| id | name   | dep | gender | addr   |
+----+--------+-----+--------+--------+
|  1 | 李明   | RD  |      1 | 北京   |
|  2 | 張三   | PM  |      0 | 上海   |
+----+--------+-----+--------+--------+
2 rows in set (0.00 sec)

mysql> create table staff_bak_1 like staff;  # 直接使用like，前面指定新表名，後面指定舊錶（參考的表）
Query OK, 0 rows affected (0.02 sec)

mysql> show create table staff_bak_1\G
*************************** 1. row ***************************
       Table: staff_bak_1
Create Table: CREATE TABLE `staff_bak_1` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `name` char(20) NOT NULL COMMENT '用戶姓名',
  `dep` char(20) NOT NULL COMMENT '所屬部門',
  `gender` tinyint(4) NOT NULL DEFAULT '1' COMMENT '性別：1男; 2女',
  `addr` char(30) NOT NULL,
  PRIMARY KEY (`id`),
  KEY `idx_1` (`name`,`dep`),
  KEY `idx_2` (`name`,`gender`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='員工表'  # 注意沒有AUTO_INCREMENT=3
1 row in set (0.00 sec)

mysql> select * from staff_bak_1; # 沒有包含舊錶的數據
Empty set (0.00 sec)

　　2.3、使用as來創建新表（包含數據）

　　使用as來創建新表，有一下特點：

　　1、可以有選擇性的決定新表包含哪些字段；

　　2、創建的新表中，會包含舊錶的數據；

　　3、創建的新表不會包含舊錶的完整性約束（比如主鍵、索引等），僅包含最基礎的表結構定義。

　　用法如下：

mysql> create table staff_bak_2 as select * from staff;
Query OK, 2 rows affected (0.02 sec)
Records: 2  Duplicates: 0  Warnings: 0

mysql> select * from staff_bak_2;
+----+--------+-----+--------+--------+
| id | name   | dep | gender | addr   |
+----+--------+-----+--------+--------+
|  1 | 李明   | RD  |      1 | 北京   |
|  2 | 張三   | PM  |      0 | 上海   |
+----+--------+-----+--------+--------+
2 rows in set (0.00 sec)

mysql> show create table staff_bak_2\G
*************************** 1. row ***************************
       Table: staff_bak_2
Create Table: CREATE TABLE `staff_bak_2` (
  `id` int(11) NOT NULL DEFAULT '0' COMMENT '自增id',
  `name` char(20) CHARACTER SET utf8mb4 NOT NULL COMMENT '用戶姓名',
  `dep` char(20) CHARACTER SET utf8mb4 NOT NULL COMMENT '所屬部門',
  `gender` tinyint(4) NOT NULL DEFAULT '1' COMMENT '性別：1男; 2女',
  `addr` char(30) CHARACTER SET utf8mb4 NOT NULL
) ENGINE=InnoDB DEFAULT CHARSET=latin1
1 row in set (0.00 sec)

　　利用as創建表的時候沒有保留完整性約束，其實這個仔細想一下也能想明白。因為使用as創建表的時候，可以指定新表包含哪些字段呀，如果你創建新表時，忽略了幾個字段，這樣的話即使保留了完整約束，保存數據是也不能滿足完整性約束。

　　比如，staff表有一個索引idx1，由name和dep字段組成；但是我創建的新表中，沒有name和dep字段（只選擇了其他字段），那麼新表中保存idx1也沒有必要，對吧。

mysql> --  只選擇id、gender、addr作為新表的字段，那麼name和dep組成的索引就沒必要存在了
mysql> create table staff_bak_3 as (select id, gender, addr from staff);
Query OK, 2 rows affected (0.02 sec)
Records: 2  Duplicates: 0  Warnings: 0

mysql> show create table staff_bak_3\G
*************************** 1. row ***************************
       Table: staff_bak_3
Create Table: CREATE TABLE `staff_bak_3` (
  `id` int(11) NOT NULL DEFAULT '0' COMMENT '自增id',
  `gender` tinyint(4) NOT NULL DEFAULT '1' COMMENT '性別：1男; 2女',
  `addr` char(30) CHARACTER SET utf8mb4 NOT NULL
) ENGINE=InnoDB DEFAULT CHARSET=latin1
1 row in set (0.00 sec)

mysql> select * from staff_bak_3;
+----+--------+--------+
| id | gender | addr   |
+----+--------+--------+
|  1 |      1 | 北京   |
|  2 |      0 | 上海   |
+----+--------+--------+
2 rows in set (0.00 sec)

　　2.4、使用like+insert+select創建原表的副本（推薦）

　　使用like創建新表，雖然保留了舊錶的各種表結構定義以及完整性約束，但是如何將舊錶的數據導入到新表中呢？

　　最極端的方式：寫一個程序，先將舊錶數據讀出來，然後寫入到新表中，這個方式我就不嘗試了。

　　有一個比較簡單的命令：

mysql> select * from staff; #原表數據
+----+--------+-----+--------+--------+
| id | name   | dep | gender | addr   |
+----+--------+-----+--------+--------+
|  1 | 李明   | RD  |      1 | 北京   |
|  2 | 張三   | PM  |      0 | 上海   |
+----+--------+-----+--------+--------+
2 rows in set (0.00 sec)

mysql> select * from staff_bak_1; # 使用like創建的表，與原表相同的表結構和完整性約束（自增除外）
Empty set (0.00 sec)

mysql> insert into staff_bak_1 select * from staff;  # 將staff表的所有記錄的所有字段值都插入副本表中
Query OK, 2 rows affected (0.00 sec)
Records: 2  Duplicates: 0  Warnings: 0

mysql> select * from staff_bak_1;
+----+--------+-----+--------+--------+
| id | name   | dep | gender | addr   |
+----+--------+-----+--------+--------+
|  1 | 李明   | RD  |      1 | 北京   |
|  2 | 張三   | PM  |      0 | 上海   |
+----+--------+-----+--------+--------+
2 rows in set (0.00 sec)

　　其實這條SQL語句，是知道兩個表的表結構和完整性約束相同，所以，可以直接select *。

insert into staff_bak_1 select * from staff;

　　如果兩個表結構不相同，其實也是可以這個方式的，比如：

mysql> show create table demo\G
*************************** 1. row ***************************
       Table: demo
Create Table: CREATE TABLE `demo` (
  `_id` int(11) NOT NULL AUTO_INCREMENT,
  `_name` char(20) DEFAULT NULL,
  `_gender` tinyint(4) DEFAULT '1',
  PRIMARY KEY (`_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
1 row in set (0.00 sec)

# 只將staff表中的id和name字段組成的數據記錄插入到demo表中，對應_id和_name字段
mysql> insert into demo (_id, _name) select id,name from staff;
Query OK, 2 rows affected (0.00 sec)
Records: 2  Duplicates: 0  Warnings: 0

mysql> select * from demo;
+-----+--------+---------+
| _id | _name  | _gender |
+-----+--------+---------+
|   1 | 李明   |       1 |
|   2 | 張三   |       1 |
+-----+--------+---------+
2 rows in set (0.00 sec)

　　這是兩個表的字段數量不相同的情況，此時需要手動指定列名，否則就會報錯。

　　另外，如果兩個表的字段數量，以及相同順序的字段類型相同，如果是全部字段複製，即使字段名不同，也可以直接複製：

# staff_bak_5的字段名與staff表並不相同，但是字段數量、相同順序字段的類型相同，所以可以直接插入
mysql> show create table staff_bak_5\G
*************************** 1. row ***************************
       Table: staff_bak_5
Create Table: CREATE TABLE `staff_bak_5` (
  `_id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `_name` char(20) NOT NULL COMMENT '用戶姓名',
  `_dep` char(20) NOT NULL COMMENT '所屬部門',
  `_gender` tinyint(4) NOT NULL DEFAULT '1' COMMENT '性別：1男; 2女',
  `_addr` char(30) NOT NULL,
  PRIMARY KEY (`_id`),
  KEY `idx_1` (`_name`,`_dep`),
  KEY `idx_2` (`_name`,`_gender`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 COMMENT='員工表'
1 row in set (0.00 sec)

mysql> insert into staff_bak_5 select * from staff;
Query OK, 2 rows affected (0.00 sec)
Records: 2  Duplicates: 0  Warnings: 0

mysql> select * from staff_bak_5;
+-----+--------+------+---------+--------+
| _id | _name  | _dep | _gender | _addr  |
+-----+--------+------+---------+--------+
|   1 | 李明   | RD   |       1 | 北京   |
|   2 | 張三   | PM   |       0 | 上海   |
+-----+--------+------+---------+--------+
2 rows in set (0.00 sec)

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

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

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

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

2020-02-14

源碼分析RocketMQ消息軌跡

本文沿着的思路，從如下3個方面對其源碼進行解讀：

發送消息軌跡
消息軌跡格式
存儲消息軌跡數據

@(本節目錄)

1、發送消息軌跡流程

首先我們來看一下在消息發送端如何啟用消息軌跡，示例代碼如下：

public class TraceProducer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName",true);      // @1
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.start();
        for (int i = 0; i < 10; i++)
            try {
                {
                    Message msg = new Message("TopicTest",
                        "TagA",
                        "OrderID188",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                    SendResult sendResult = producer.send(msg);
                    System.out.printf("%s%n", sendResult);
                }

            } catch (Exception e) {
                e.printStackTrace();
            }
        producer.shutdown();
    }
}

從上述代碼可以看出其關鍵點是在創建DefaultMQProducer時指定開啟消息軌跡跟蹤。我們不妨瀏覽一下DefaultMQProducer與啟用消息軌跡相關的構造函數：

public DefaultMQProducer(final String producerGroup, boolean enableMsgTrace)
public DefaultMQProducer(final String producerGroup, boolean enableMsgTrace, final String customizedTraceTopic)

參數如下：

String producerGroup
生產者所屬組名。
boolean enableMsgTrace
是否開啟跟蹤消息軌跡，默認為false。
String customizedTraceTopic
如果開啟消息軌跡跟蹤，用來存儲消息軌跡數據所屬的主題名稱，默認為：RMQ_SYS_TRACE_TOPIC。

1.1 DefaultMQProducer構造函數

public DefaultMQProducer(final String producerGroup, RPCHook rpcHook, boolean enableMsgTrace,final String customizedTraceTopic) {      // @1
    this.producerGroup = producerGroup;
    defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
    //if client open the message trace feature
    if (enableMsgTrace) {                                                                                                                                                                                            // @2
        try {
            AsyncTraceDispatcher dispatcher = new AsyncTraceDispatcher(customizedTraceTopic, rpcHook);                                                         
            dispatcher.setHostProducer(this.getDefaultMQProducerImpl());
            traceDispatcher = dispatcher;
            this.getDefaultMQProducerImpl().registerSendMessageHook(
                new SendMessageTraceHookImpl(traceDispatcher));                                                                                                                             // @3
        } catch (Throwable e) {
            log.error("system mqtrace hook init failed ,maybe can't send msg trace data");
        }
    }
}

代碼@1：首先介紹一下其局部變量。

String producerGroup
生產者所屬組。
RPCHook rpcHook
生產者發送鈎子函數。
boolean enableMsgTrace
是否開啟消息軌跡跟蹤。
String customizedTraceTopic
定製用於存儲消息軌跡的數據。

代碼@2：用來構建AsyncTraceDispatcher，看其名：異步轉發消息軌跡數據，稍後重點關注。

代碼@3：構建SendMessageTraceHookImpl對象，並使用AsyncTraceDispatcher用來異步轉發。

1.2 SendMessageTraceHookImpl鈎子函數

1.2.1 SendMessageTraceHookImpl類圖

SendMessageHook
消息發送鈎子函數，用於在消息發送之前、發送之後執行一定的業務邏輯，是記錄消息軌跡的最佳擴展點。
TraceDispatcher
消息軌跡轉發處理器，其默認實現類AsyncTraceDispatcher，異步實現消息軌跡數據的發送。下面對其屬性做一個簡單的介紹：
- int queueSize
  異步轉發，隊列長度，默認為2048，當前版本不能修改。
- int batchSize
  批量消息條數，消息軌跡一次消息發送請求包含的數據條數，默認為100，當前版本不能修改。
- int maxMsgSize
  消息軌跡一次發送的最大消息大小，默認為128K，當前版本不能修改。
- DefaultMQProducer traceProducer
  用來發送消息軌跡的消息發送者。
- ThreadPoolExecutor traceExecuter
  線程池，用來異步執行消息發送。
- AtomicLong discardCount
  記錄丟棄的消息個數。
- Thread worker
  woker線程，主要負責從追加隊列中獲取一批待發送的消息軌跡數據，提交到線程池中執行。
- ArrayBlockingQueue< TraceContext> traceContextQueue
  消息軌跡TraceContext隊列，用來存放待發送到服務端的消息。
- ArrayBlockingQueue< Runnable> appenderQueue
  線程池內部隊列，默認長度1024。
- DefaultMQPushConsumerImpl hostConsumer
  消費者信息，記錄消息消費時的軌跡信息。
- String traceTopicName
  用於跟蹤消息軌跡的topic名稱。

1.2.2 源碼分析SendMessageTraceHookImpl

1.2.2.1 sendMessageBefore

public void sendMessageBefore(SendMessageContext context) { 
    //if it is message trace data,then it doesn't recorded
    if (context == null || context.getMessage().getTopic().startsWith(((AsyncTraceDispatcher) localDispatcher).getTraceTopicName())) {   // @1
        return;
    }
    //build the context content of TuxeTraceContext
    TraceContext tuxeContext = new TraceContext();
    tuxeContext.setTraceBeans(new ArrayList<TraceBean>(1));
    context.setMqTraceContext(tuxeContext);
    tuxeContext.setTraceType(TraceType.Pub);
    tuxeContext.setGroupName(context.getProducerGroup());                                                                                                                       // @2
    //build the data bean object of message trace
    TraceBean traceBean = new TraceBean();                                                                                                                                                // @3
    traceBean.setTopic(context.getMessage().getTopic());
    traceBean.setTags(context.getMessage().getTags());
    traceBean.setKeys(context.getMessage().getKeys());
    traceBean.setStoreHost(context.getBrokerAddr());
    traceBean.setBodyLength(context.getMessage().getBody().length);
    traceBean.setMsgType(context.getMsgType());
    tuxeContext.getTraceBeans().add(traceBean);
}

代碼@1：如果topic主題為消息軌跡的Topic，直接返回。

代碼@2：在消息發送上下文中，設置用來跟蹤消息軌跡的上下環境，裏面主要包含一個TraceBean集合、追蹤類型（TraceType.Pub）與生產者所屬的組。

代碼@3：構建一條跟蹤消息，用TraceBean來表示，記錄原消息的topic、tags、keys、發送到broker地址、消息體長度等消息。

從上文看出，sendMessageBefore主要的用途就是在消息發送的時候，先準備一部分消息跟蹤日誌，存儲在發送上下文環境中，此時並不會發送消息軌跡數據。

1.2.2.2 sendMessageAfter

public void sendMessageAfter(SendMessageContext context) {
    //if it is message trace data,then it doesn't recorded
    if (context == null || context.getMessage().getTopic().startsWith(((AsyncTraceDispatcher) localDispatcher).getTraceTopicName())     // @1
        || context.getMqTraceContext() == null) {
        return;
    }
    if (context.getSendResult() == null) {
        return;
    }

    if (context.getSendResult().getRegionId() == null
        || !context.getSendResult().isTraceOn()) {
        // if switch is false,skip it
        return;
    }

    TraceContext tuxeContext = (TraceContext) context.getMqTraceContext();
    TraceBean traceBean = tuxeContext.getTraceBeans().get(0);                                                                                                // @2
    int costTime = (int) ((System.currentTimeMillis() - tuxeContext.getTimeStamp()) / tuxeContext.getTraceBeans().size());     // @3
    tuxeContext.setCostTime(costTime);                                                                                                                                      // @4
    if (context.getSendResult().getSendStatus().equals(SendStatus.SEND_OK)) {                                                                    
        tuxeContext.setSuccess(true);
    } else {
        tuxeContext.setSuccess(false);
    }
    tuxeContext.setRegionId(context.getSendResult().getRegionId());                                                                                      
    traceBean.setMsgId(context.getSendResult().getMsgId());
    traceBean.setOffsetMsgId(context.getSendResult().getOffsetMsgId());
    traceBean.setStoreTime(tuxeContext.getTimeStamp() + costTime / 2);
    localDispatcher.append(tuxeContext);                                                                                                                                   // @5
}

代碼@1：如果topic主題為消息軌跡的Topic，直接返回。

代碼@2：從MqTraceContext中獲取跟蹤的TraceBean，雖然設計成List結構體，但在消息發送場景，這裏的數據永遠只有一條，及時是批量發送也不例外。

代碼@3：獲取消息發送到收到響應結果的耗時。

代碼@4：設置costTime(耗時)、success(是否發送成功)、regionId(發送到broker所在的分區)、msgId(消息ID，全局唯一)、offsetMsgId(消息物理偏移量，如果是批量消息，則是最後一條消息的物理偏移量)、storeTime，這裏使用的是(客戶端發送時間 + 二分之一的耗時)來表示消息的存儲時間，這裡是一個估值。

代碼@5：將需要跟蹤的信息通過TraceDispatcher轉發到Broker服務器。其代碼如下：

public boolean append(final Object ctx) {
    boolean result = traceContextQueue.offer((TraceContext) ctx);
    if (!result) {
        log.info("buffer full" + discardCount.incrementAndGet() + " ,context is " + ctx);
    }
    return result;
}

這裏一個非常關鍵的點是offer方法的使用，當隊列無法容納新的元素時會立即返回false，並不會阻塞。

接下來將目光轉向TraceDispatcher的實現。

1.3 TraceDispatcher實現原理

TraceDispatcher，用於客戶端消息軌跡數據轉發到Broker，其默認實現類：AsyncTraceDispatcher。

1.3.1 TraceDispatcher構造函數

public AsyncTraceDispatcher(String traceTopicName, RPCHook rpcHook) throws MQClientException {    
    // queueSize is greater than or equal to the n power of 2 of value
    this.queueSize = 2048;
    this.batchSize = 100;
    this.maxMsgSize = 128000;                                        
    this.discardCount = new AtomicLong(0L);         
    this.traceContextQueue = new ArrayBlockingQueue<TraceContext>(1024);
    this.appenderQueue = new ArrayBlockingQueue<Runnable>(queueSize);
    if (!UtilAll.isBlank(traceTopicName)) {
        this.traceTopicName = traceTopicName;
    } else {
        this.traceTopicName = MixAll.RMQ_SYS_TRACE_TOPIC;
    }                   // @1
    this.traceExecuter = new ThreadPoolExecutor(// :
        10, //
        20, //
        1000 * 60, //
        TimeUnit.MILLISECONDS, //
        this.appenderQueue, //
        new ThreadFactoryImpl("MQTraceSendThread_"));
    traceProducer = getAndCreateTraceProducer(rpcHook);      // @2
}

代碼@1：初始化核心屬性，該版本這些值都是“固化”的，用戶無法修改。

queueSize
隊列長度，默認為2048，異步線程池能夠積壓的消息軌跡數量。
batchSize
一次向Broker批量發送的消息條數，默認為100.
maxMsgSize
向Broker彙報消息軌跡時，消息體的總大小不能超過該值，默認為128k。
discardCount
整個運行過程中，丟棄的消息軌跡數據，這裏要說明一點的是，如果消息TPS發送過大，異步轉發線程處理不過來時，會主動丟棄消息軌跡數據。
traceContextQueue
traceContext積壓隊列，客戶端(消息發送、消息消費者)在收到處理結果后，將消息軌跡提交到噶隊列中，則會立即返回。
appenderQueue
提交到Broker線程池中隊列。
traceTopicName
用於接收消息軌跡的Topic，默認為RMQ_SYS_TRANS_HALF_TOPIC。
traceExecuter
用於發送到Broker服務的異步線程池，核心線程數默認為10，最大線程池為20，隊列堆積長度2048，線程名稱：MQTraceSendThread_。、
traceProducer
發送消息軌跡的Producer。

代碼@2：調用getAndCreateTraceProducer方法創建用於發送消息軌跡的Producer(消息發送者)，下面詳細介紹一下其實現。

1.3.2 getAndCreateTraceProducer詳解

private DefaultMQProducer getAndCreateTraceProducer(RPCHook rpcHook) {
        DefaultMQProducer traceProducerInstance = this.traceProducer;
        if (traceProducerInstance == null) {  //@1
            traceProducerInstance = new DefaultMQProducer(rpcHook);
            traceProducerInstance.setProducerGroup(TraceConstants.GROUP_NAME);
            traceProducerInstance.setSendMsgTimeout(5000);
            traceProducerInstance.setVipChannelEnabled(false);
            // The max size of message is 128K
            traceProducerInstance.setMaxMessageSize(maxMsgSize - 10 * 1000);
        }
        return traceProducerInstance;
    }

代碼@1：如果還未建立發送者，則創建用於發送消息軌跡的消息發送者，其GroupName為：_INNER_TRACE_PRODUCER，消息發送超時時間5s，最大允許發送消息大小118K。

1.3.3 start

public void start(String nameSrvAddr) throws MQClientException {
    if (isStarted.compareAndSet(false, true)) {     // @1
        traceProducer.setNamesrvAddr(nameSrvAddr);
        traceProducer.setInstanceName(TRACE_INSTANCE_NAME + "_" + nameSrvAddr);
        traceProducer.start();
    }
    this.worker = new Thread(new AsyncRunnable(), "MQ-AsyncTraceDispatcher-Thread-" + dispatcherId);   // @2
    this.worker.setDaemon(true);
    this.worker.start();                                                                                   
    this.registerShutDownHook();
}

開始啟動，其調用的時機為啟動DefaultMQProducer時，如果啟用跟蹤消息軌跡，則調用之。

代碼@1：如果用於發送消息軌跡的發送者沒有啟動，則設置nameserver地址，並啟動着。

代碼@2：啟動一個線程，用於執行AsyncRunnable任務，接下來將重點介紹。

1.3.4 AsyncRunnable

class AsyncRunnable implements Runnable {
         private boolean stopped;
    public void run() {
        while (!stopped) {
            List<TraceContext> contexts = new ArrayList<TraceContext>(batchSize);     // @1
            for (int i = 0; i < batchSize; i++) {
                TraceContext context = null;
                try {
                    //get trace data element from blocking Queue — traceContextQueue
                    context = traceContextQueue.poll(5, TimeUnit.MILLISECONDS);        // @2
                } catch (InterruptedException e) {
                }
                if (context != null) {
                    contexts.add(context);
                } else {
                    break;
                }
            }
            if (contexts.size() > 0) {                                                                               :
                AsyncAppenderRequest request = new AsyncAppenderRequest(contexts);  // @3
                traceExecuter.submit(request);                                                               
            } else if (AsyncTraceDispatcher.this.stopped) {
                this.stopped = true;
            }
        }
    }
}

代碼@1：構建待提交消息跟蹤Bean，每次最多發送batchSize，默認為100條。

代碼@2：從traceContextQueue中取出一個待提交的TraceContext，設置超時時間為5s，即如何該隊列中沒有待提交的TraceContext，則最多等待5s。

代碼@3：向線程池中提交任務AsyncAppenderRequest。

1.3.5 AsyncAppenderRequest#sendTraceData

public void sendTraceData(List<TraceContext> contextList) {
    Map<String, List<TraceTransferBean>> transBeanMap = new HashMap<String, List<TraceTransferBean>>();
    for (TraceContext context : contextList) {        //@1
        if (context.getTraceBeans().isEmpty()) {
            continue;
        }
        // Topic value corresponding to original message entity content
        String topic = context.getTraceBeans().get(0).getTopic();     // @2
        // Use  original message entity's topic as key
        String key = topic;
        List<TraceTransferBean> transBeanList = transBeanMap.get(key);
        if (transBeanList == null) {
            transBeanList = new ArrayList<TraceTransferBean>();
            transBeanMap.put(key, transBeanList);
        }
        TraceTransferBean traceData = TraceDataEncoder.encoderFromContextBean(context);    // @3
        transBeanList.add(traceData);
    }
    for (Map.Entry<String, List<TraceTransferBean>> entry : transBeanMap.entrySet()) {       // @4
        flushData(entry.getValue());
    }
}

代碼@1：遍歷收集的消息軌跡數據。

代碼@2：獲取存儲消息軌跡的Topic。

代碼@3：對TraceContext進行編碼，這裡是消息軌跡的傳輸數據，稍後對其詳細看一下，了解其上傳的格式。

代碼@4：將編碼后的數據發送到Broker服務器。

1.3.6 TraceDataEncoder#encoderFromContextBean

根據消息軌跡跟蹤類型，其格式會有一些不一樣，下面分別來介紹其合適。

1.3.6.1 PUB(消息發送)

case Pub: {
    TraceBean bean = ctx.getTraceBeans().get(0);
    //append the content of context and traceBean to transferBean's TransData
    sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getTimeStamp()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getRegionId()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getGroupName()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getTopic()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getTags()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getKeys()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getStoreHost()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getBodyLength()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getCostTime()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getMsgType().ordinal()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getOffsetMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
     .append(ctx.isSuccess()).append(TraceConstants.FIELD_SPLITOR);
}

消息軌跡數據的協議使用字符串拼接，字段的分隔符號為1，整個數據以2結尾，感覺這個設計還是有點“不可思議”，為什麼不直接使用json協議呢？

1.3.6.2 SubBefore(消息消費之前)

for (TraceBean bean : ctx.getTraceBeans()) {
    sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getTimeStamp()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getRegionId()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getGroupName()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getRequestId()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getRetryTimes()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getKeys()).append(TraceConstants.FIELD_SPLITOR);//
    }
}

軌跡就是按照上述順序拼接而成，各個字段使用1分隔，每一條記錄使用2結尾。

1.3.2.3 SubAfter（消息消費后）

case SubAfter: {
    for (TraceBean bean : ctx.getTraceBeans()) {
        sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)//
          .append(ctx.getRequestId()).append(TraceConstants.CONTENT_SPLITOR)//
          .append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
          .append(ctx.getCostTime()).append(TraceConstants.CONTENT_SPLITOR)//
          .append(ctx.isSuccess()).append(TraceConstants.CONTENT_SPLITOR)//
          .append(bean.getKeys()).append(TraceConstants.CONTENT_SPLITOR)//
          .append(ctx.getContextCode()).append(TraceConstants.FIELD_SPLITOR);
        }
    }
}

格式編碼一樣，就不重複多說。

經過上面的源碼跟蹤，消息發送端的消息軌跡跟蹤流程、消息軌跡數據編碼協議就清晰了，接下來我們使用一張序列圖來結束本部分的講解。

其實行文至此，只關注了消息發送的消息軌跡跟蹤，消息消費的軌跡跟蹤又是如何呢？其實現原理其實是一樣的，就是在消息消費前後執行特定的鈎子函數，其實現類為ConsumeMessageTraceHookImpl，由於其實現與消息發送的思路類似，故就不詳細介紹了。

2、消息軌跡數據如何存儲

其實從上面的分析，我們已經得知，RocketMQ的消息軌跡數據存儲在到Broker上，那消息軌跡的主題名如何指定？其路由信息又怎麼分配才好呢？是每台Broker上都創建還是只在其中某台上創建呢？RocketMQ支持系統默認與自定義消息軌跡的主題。

2.1 使用系統默認的主題名稱

RocketMQ默認的消息軌跡主題為：RMQ_SYS_TRACE_TOPIC，那該Topic需要手工創建嗎？其路由信息呢？

{
    if (this.brokerController.getBrokerConfig().isTraceTopicEnable()) {    // @1
        String topic = this.brokerController.getBrokerConfig().getMsgTraceTopicName();
        TopicConfig topicConfig = new TopicConfig(topic);
        this.systemTopicList.add(topic);
        topicConfig.setReadQueueNums(1);                                              // @2
        topicConfig.setWriteQueueNums(1);
        this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
    }
}

上述代碼出自TopicConfigManager的構造函數，在Broker啟動的時候會創建topicConfigManager對象，用來管理topic的路由信息。

代碼@1：如果Broker開啟了消息軌跡跟蹤(traceTopicEnable=true)時，會自動創建默認消息軌跡的topic路由信息，注意其讀寫隊列數為1。

2.2 用戶自定義消息軌跡主題

在創建消息發送者、消息消費者時，可以显示的指定消息軌跡的Topic，例如：

public DefaultMQProducer(final String producerGroup, RPCHook rpcHook, boolean enableMsgTrace,final String customizedTraceTopic)

public DefaultMQPushConsumer(final String consumerGroup, RPCHook rpcHook,
        AllocateMessageQueueStrategy allocateMessageQueueStrategy, boolean enableMsgTrace, final String customizedTraceTopic)

通過customizedTraceTopic來指定消息軌跡Topic。

溫馨提示：通常在生產環境上，將不會開啟自動創建主題，故需要RocketMQ運維管理人員提前創建好Topic。

好了，本文就介紹到這裏了，本文詳細介紹了RocktMQ消息軌跡的實現原理，下一篇，我們將進入到多副本的學習中。

作者介紹：
丁威，《RocketMQ技術內幕》作者，RocketMQ 社區佈道師，公眾號：維護者，目前已陸續發表源碼分析Java集合、Java 併發包(JUC)、Netty、Mycat、Dubbo、RocketMQ、Mybatis等源碼專欄。

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

※想知道網站建置、網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計及後台網頁設計

※不管是台北網頁設計公司、台中網頁設計公司，全省皆有專員為您服務

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

※帶您來看台北網站建置，台北網頁設計，各種案例分享

2020-02-14

【目標檢測實戰】目標檢測實戰之一–手把手教你LMDB格式數據集製作！

文章目錄

1 目標檢測簡介
2 lmdb數據製作
    2.1 VOC數據製作
    2.2 lmdb文件生成

lmdb格式的數據是在使用caffe進行目標檢測或分類時，使用的一種數據格式。這裏我主要以目標檢測為例講解lmdb格式數據的製作。

1 目標檢測簡介

【1】目標檢測主要有兩個任務：

判斷圖像中對象的類別
類別的位置

【2】目標檢測需要的數據：

訓練所需的圖像數據，可以是jpg、png等圖片格式
圖像數據對應的類別信息和類別框的位置信息。

2 lmdb數據製作

caffe一般使用lmdb格式的數據，在製作數據之前，我們需要對數據進行標註，可以使用labelImg對圖像進行標註（），這裏就不多贅述數據標註的問題。總之，我們得到了圖像的標註Annotations數據。lmdb數據製作，首先需要將annotations數據和圖像數據製作為VOC格式，然後將其生成LMDB文件即可。下邊是詳細的步驟：

2.1 VOC數據製作

這裏我以caffe環境的Mobilenet+YOLOv3模型的代碼為例（），進行lmdb數據製作，並且也假設你已經對其配置編譯成功（如沒成功，可以參考博文進行配置），所以我們的根目錄為:caffe-Mobilenet-YOLO-master，下邊為詳細步驟：

【1】VOC格式目錄建立：

VOC格式目錄主要包含為：

其中，Annotations里存儲的是xml標註信息，JPEGImages存儲的是圖片，ImageSets則是訓練和測試的txt列表等信息，下邊我們就要安裝如上的目錄進行建立我們自己的數據目錄。

創建Annotations、JPEGImages、ImageSets/Main等文件，命令如下（也可直接界面操作哈）：

注：建議新手按照我的名稱，對於後續文件修改容易！！！

cd ~/   # 進入home目錄
cd Documents/  # 進入Documents目錄
cd caffe-Mobilenet-YOLO-master/  # 進入我們的根目錄
cd data         # 進入data目錄內
mkdir VOCdevkit   # 創建存儲我們自己的數據的文件夾
cd VOCdevkit
mkdir MyDataSet  # 創建存儲voc的目錄
cd MyDataSet   
# 創建VOC格式目錄
mkdir Annotations
mkdir JPEGImages
mkdir ImageSets
cd ImageSets
mkdir Main

好啦，我們的文件夾就建立好了，如下圖所示：

【2】將所有xml文件考入至Annotations文件夾內
【3】將所有圖片考入至JPEGImages文件夾內
【4】劃分訓練接、驗證集合測試集，如下為Python代碼，需要修改的地方註釋已標明：

import os  
import random 
# 標註文件的路徑，需要你自己修改
xmlfilepath=r'/home/Documents/caffe-Mobilenet-YOLO-master/data/VOCdevkit/MyDataSet/Annotations/'      
# 這裡是存儲數據的本目錄，需要改為你自己的目錄              
saveBasePath=r"/home/Documents/caffe-Mobilenet-YOLO-master/data/VOCdevkit/"                        
trainval_percent=0.8            # 表示訓練集和驗證集所佔比例，你需要自己修改，也可選擇不修改
train_percent=0.8               # 表示訓練集所佔訓練集驗證集的比例，你需要自己修改，也可選擇不修改       
total_xml = os.listdir(xmlfilepath)
num=len(total_xml)    
list=range(num)    
tv=int(num*trainval_percent)    
tr=int(tv*train_percent)    
trainval= random.sample(list,tv)    
train=random.sample(trainval,tr)    
  
print("train and val size",tv)  
print("traub suze",tr)  
ftrainval = open(os.path.join(saveBasePath,'MyDataSet/ImageSets/Main/trainval.txt'), 'w')    
ftest = open(os.path.join(saveBasePath,'MyDataSet/ImageSets/Main/test.txt'), 'w')    
ftrain = open(os.path.join(saveBasePath,'MyDataSet/ImageSets/Main/train.txt'), 'w')    
fval = open(os.path.join(saveBasePath,'MyDataSet/ImageSets/Main/val.txt'), 'w')    
  
for i  in list:    
    name=total_xml[i][:-4]+'\n'    
    if i in trainval:    
        ftrainval.write(name)    
        if i in train:    
            ftrain.write(name)    
        else:    
            fval.write(name)    
    else:    
        ftest.write(name)    
    
ftrainval.close()    
ftrain.close()    
fval.close()    
ftest .close()

上述代碼修改之後，在根目錄caffe-Mobilenet-YOLO-master執行上述代碼即可，
在data/VOCdevkit/MyDataSet/ImageSets下生成trainval.txt、test.txt、train.txt、val.txt等所需的txt文件，如下圖所示：

這些TXT文件會包含圖片的名字，不帶路徑，如下圖所示：

2.2 lmdb文件生成

【1】執行如下命令，將生成lmdb所需的腳本複製至data/VOCdevkit/MyDataSet文件夾內：

cp data/VOC0712/create_* data/MyDataSet/                # 把create_list.sh和create_data.sh複製到MyDataSet目錄                  
cp data/VOC0712/labelmap_voc.prototxt data/MyDataSet/   # 把labelmap_voc.prototxt複製到MyDataSet目錄

【2】修改create_list.sh文件：

1 第3行修改目錄路徑，截止到VOCdevkit即可

2 第13行修改為for name in MyDataSet(VOCdevkit下自己建立的文件夾名字)

3 第15-18行註釋掉

4 第41行get_image_size修改為自己的路徑（注意，這裡是build caffe_mobilenet_yolo之後才會形成的）：

#!/bin/bash
# 如果嚴格安裝我上述的步驟，就可以不用修改路徑位置。
# 需要修改的位置也使用註釋進行了標註和解釋

# 這裏需要更改，你數據的根目錄位置，需要修改的地方！！！！
root_dir="/home/Documents/Caffe_Mobilenet_YOLO/data/VOCdevkit/"   
sub_dir=ImageSets/Main
bash_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
for dataset in trainval test
do
  dst_file=$bash_dir/$dataset.txt
  if [ -f $dst_file ]
  then
    rm -f $dst_file
  fi
  for name in MyDataSet  # 如果你建立的不是MyDataSet，這裏需要修改為你自己的名字
  do
    # 這裏需要修改，註釋掉即可
    #if [[ $dataset == "test" && $name == "VOC2012" ]]
    #then
    #  continue
    #fi
    echo "Create list for $name $dataset..."
    dataset_file=$root_dir/$name/$sub_dir/$dataset.txt

    img_file=$bash_dir/$dataset"_img.txt"
    cp $dataset_file $img_file
    sed -i "s/^/$name\/JPEGImages\//g" $img_file
    sed -i "s/$/.jpg/g" $img_file

    label_file=$bash_dir/$dataset"_label.txt"
    cp $dataset_file $label_file
    sed -i "s/^/$name\/Annotations\//g" $label_file
    sed -i "s/$/.xml/g" $label_file

    paste -d' ' $img_file $label_file >> $dst_file

    rm -f $label_file
    rm -f $img_file
  done

  # Generate image name and size infomation.
  if [ $dataset == "test" ]
  then
    home/Documents/Caffe_Mobilenet_YOLO/caffe-MobileNet-YOLO-master/build/tools/get_image_size $root_dir $dst_file $bash_dir/$dataset"_name_size.txt"

【3】creat_data.sh修改：

1 第2行修改為自己的路徑：root_dir=”/home/Documents/caffe-MobileNet-YOLO-master/”

2 第7行修改為：data_root_dir=”/home/Documents/caffe-MobileNet-YOLO-master/data/VOVdevkit/

3 第8行修改為：dataset_name=”MyDataSet”

4 第9行修改為：mapfile=”$root_dir/data/VOCdevkit/$dataset_name/labelmap_voc.prototxt”

5 第26行修改為$root_dir/data/VOCdevkit/$dataset_name/$subset.txt

cur_dir=$(cd $( dirname ${BASH_SOURCE[0]} ) && pwd )
# 修改為自己的路徑
root_dir="/home/Documents/Caffe_Mobilenet_YOLO/caffe-MobileNet-YOLO-master/"

cd $root_dir

redo=1
# 這裏需要修改為自己的路徑
data_root_dir="/home/Documents/Caffe_Mobilenet_YOLO/caffe-MobileNet-YOLO-master/data/VOCdevkit/"
dataset_name="MyDataSet"  # 修改為自己的名字
mapfile="$root_dir/data/VOCdevkit/$dataset_name/labelmap_voc.prototxt"  # 修改為自己的路徑
anno_type="detection"
db="lmdb"
min_dim=0
max_dim=0
width=0
height=0

extra_cmd="--encode-type=jpg --encoded"
if [ $redo ]
then
  extra_cmd="$extra_cmd --redo"
fi
for subset in test trainval
# subset.txt路徑需要修改
do
  python $root_dir/scripts/create_annoset.py --anno-type=$anno_type \
  --label-map-file=$mapfile --min-dim=$min_dim --max-dim=$max_dim --resize-width=$width \
  --resize-height=$height --check-label $extra_cmd $data_root_dir $root_dir/data/VOCdevkit/$dataset_name/$subset.txt \
  $data_root_dir/$dataset_name/$db/$dataset_name"_"$subset"_"$db examples/$dataset_name

【3】修改labelmap_voc.prototxt文件：

除了第一個背景標籤部分不要修改，其他改成自己的標籤就行，多的刪掉，少了添加進入就行

【4】最後在caffe-MobileNet-YOLO-master/examples文件夾內新建一個MyDataSet文件夾（空的）

【5】運行create_list.sh腳本: ./data/VOCdevkit/MyDataSet/create_list.sh，運行完后，會在自己建的VOCdevkit/MyDataSet/目錄內生成trainval.txt, test.txt, test_name_size.txt。

【6】運行create_data.sh腳本: ./data/VOCdevkit/MyDataSet/create_data.sh

運行此命令時，提示：bash：./data/VOCdevkit/MyDataSet/create_list.sh：Permission denied，沒有權限，需要執行如下命令賦予執行命令：

chmod u+x data/VOCdevkit/MyDataSet/create_data.sh

出現了錯誤：ValueError: need more than 2 values to unpack，

需要將create_annoset.py中第88行的seg去掉，因為我們的Annotations只有兩個值，img_file和anno。

運行完后，會在會在自己建的VOCdevkit/MyDataSet目錄內生成lmdb文件夾：

lmdb對應訓練集和測試集的lmdb格式的文件夾：

***
好啦，今天的教程就到這裏，如有疑問，可關注公眾號【計算機視覺聯盟】私信我或留言交流！！

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

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

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

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

2020-02-14

高性能網絡通訊原理
目錄
高性能網絡通訊原理

前言

本來想對netty的源碼進行學習和探究，但是在寫netty之前許多底層的知識和原理性的東西理解清楚，那麼對學習網絡通訊框架的效果則會事半功倍。

本篇主要探討高性能網絡通訊框架的一些必要知識和底層操作系統相關的原理。在探討如何做之前，我們先討論下為什麼要做。

隨着互聯網的高速發展，用戶量呈指數形式遞增，從原來的PC普及到現在的移動設備普及。用戶量都是千萬甚至億為單位計算，尤其是實時通訊軟件，在線實時互動的應用出現，在線用戶數從原來的幾十上百到後來的上萬甚至上千萬。單台服務的性能瓶頸和網絡通訊瓶頸慢慢呈現。應用架構從單應用到應用數據分離，再到分佈式集群高可用架構。單台服務的性能不足可以通過構建服務集群的方式水平擴展，應用性能瓶頸被很好的解決。但是橫向擴展帶來了直接的經濟成本。

一個高性能的網絡通訊框架從硬件設備到操作系統內核以及用戶模式都需要精心設計。只要有任何地方有疏漏都會出現短板效應。

I/O訪問

當我們在讀取socket數據時，雖然我們在代碼僅僅是調用了一個Read操作，但是實際操作系統層面做了許多事情。首先操作系統需要從用戶模式轉換為內核模式，處理器會通過網卡驅動對網卡控制器進行操作，網卡控制器則控制網卡。

處理器不會直接操控硬件。

為了提高CPU利用率，I/O訪問方式也發生了很大變化。
1. 早期的CPU直接控制外圍設備，後來增加了控制器或I/O模塊。處理器開始將I/O操作從外部設備接口分離出來。處理器通過向I/O模塊發送命令執行I/O指令。然而當I/O操作完成時並不會通知處理器I/O，因此處理器需要定時檢查I/O模塊的狀態，它會進行忙等待，因此效率並不高。
2. 後來CPU支持了中斷方式，處理器無需等待執行I/O操作，通過中斷控制器產生中斷信號通知I/O操作完成，大大的提高了處理器利用效率。這時的I/O操作使用特定的in/out(I/O端口)指令或直接讀寫內存的方式(內存映射I/O)。但是這些方式都需要處理器使用I/O寄存器逐個內存單元進行訪問，效率並不高，在I/O操作時需要消耗的CPU時鐘周期。
3. 為了提高效率，後來增加了DMA控制器，它可以模擬處理起獲得內存總線控制權，進行I/O的讀寫。當處理器將控制權交給DMA控制器之後，DMA處理器會先讓I/O硬件設備將數據放到I/O硬件的緩衝區中，然後DMA控制器就可以開始傳輸數據了。在此過程中處理器無需消耗時鐘周期。當DMA操作完成時，會通過中斷操作通知處理器。
I/O訪問的發展趨勢是盡可能減少處理器干涉I/O操作，讓CPU從I/O任務中解脫出來，讓處理器可以去做其他事情，從而提高性能。

對於I/O訪問感興趣的同學可以看《操作系統精髓與設計原理(第5版)》第十一章I/O管理相關內容和《WINDOWS內核原理與實現》第六章I/O論述相關內容

I/O模型

在討論I/O模型之前，首先引出一個叫做C10K的問題。在早期的I/O模型使用的是同步阻塞模型，當接收到一個新的TCP連接時，就需要分配一個線程。因此隨着連接增加線程增多，頻繁的內存複製，上下文切換帶來的性能損耗導致性能不佳。因此如何使得單機網絡併發連接數達到10K成為通訊開發者熱門的討論話題。

同步阻塞

前面提到，在最原始的I/O模型中，對文件設備數據的讀寫需要同步等待操作系統內核，即使文件設備並沒有數據可讀，線程也會被阻塞住，雖然阻塞時不佔用CPU始終周期，但是若需要支持併發連接，則必須啟用大量的線程，即每個連接一個線程。這樣必不可少的會造成線程大量的上下文切換，隨着併發量的增高，性能越來越差。

select模型/poll模型

為了解決同步阻塞帶來線程過多導致的性能問題，同步非阻塞方案產生。通過一個線程不斷的判斷文件句柄數組是否有準備就緒的文件設備，這樣就不需要每個線程同步等待，減少了大量線程，降低了線程上下文切換帶來的性能損失，提高了線程利用率。這種方式也稱為I/O多路復用技術。但是由於數組是有數組長度上限的(linux默認是1024)，而且select模型需要對數組進行遍歷，因此時間複雜度是$O_{(n)}$因此當高併發量的時候，select模型性能會越來越差。

poll模型和select模型類似，但是它使用鏈表存儲而非數組存儲，解決了併發上限的限制，但是並沒有解決select模型的高併發性能底下的根本問題。

epoll模型

在linux2.6支持了epoll模型，epoll模型解決了select模型的性能瓶頸問題。它通過註冊回調事件的方式，當數據可讀寫時，將其加入到通過回調方式，將其加入到一個可讀寫事件的隊列中。這樣每次用戶獲取時不需要遍歷所有句柄，時間複雜度降低為$O_{(1)}$。因此epoll不會隨着併發量的增加而性能降低。隨着epoll模型的出現C10K的問題已經完美解決。

異步I/O模型

前面講的幾種模型都是同步I/O模型，異步I/O模型指的是發生數據讀寫時完全不同步阻塞等待，換句話來說就是數據從網卡傳輸到用戶空間的過程時完全異步的，不用阻塞CPU。為了更詳細的說明同步I/O與異步I/O的區別，接下來舉一個實際例子。

當應用程序需要從網卡讀取數據時，首先需要分配一個用戶內存空間用來保存需要讀取的數據。操作系統內核會調用網卡緩衝區讀取數據到內核空間的緩衝區，然後再複製到用戶空間。在這個過程中，同步阻塞I/O在數據讀取到用戶空間之前都會被阻塞，同步非阻塞I/O只知道數據已就緒，但是從內核空間緩衝區拷貝到用戶空間時，線程依然會被阻塞。而異步I/O模型在接收到I/O完成通知時，數據已經傳輸到用戶空間。因此整個I/O操作都是完全異步的，因此異步I/O模型的性能是最佳的。

在我的另一篇文章對windows操作系統I/O原理做了簡要的敘述，感興趣的同學可以看下。

I/O線程模型

從線程模型上常見的線程模型有Reactor模型和Proactor模型，無論是哪種線程模型都使用I/O多路復用技術，使用一個線程將I/O讀寫操作轉變為讀寫事件，我們將這個線程稱之為多路分離器。

對應上I/O模型，Reacor模型屬於同步I/O模型，Proactor模型屬於異步I/O模型。

Reactor模型

在Reactor中，需要先註冊事件就緒事件，網卡接收到數據時，DMA將數據從網卡緩衝區傳輸到內核緩衝區時，就會通知多路分離器讀事件就緒，此時我們需要從內核空間讀取到用戶空間。

同步I/O採用緩衝I/O的方式，首先內核會從申請一個內存空間用於存放輸入或輸出緩衝區，數據都會先緩存在該緩衝區。

Proactor模型

Proactor模型，需要先註冊I/O完成事件，同時申請一片用戶空間用於存儲待接收的數據。調用讀操作，當網卡接收到數據時，DMA將數據從網卡緩衝區直接傳輸到用戶緩衝區，然後產生完成通知，讀操作即完成。

異步I/O採用直接輸入I/O或直接輸出I/O，用戶緩存地址會傳遞給設備驅動程序，數據會直接從用戶緩衝區讀取或直接寫入用戶緩衝區，相比緩衝I/O減少內存複製。

總結

本文通過I/O訪問方式，I/O模型，線程模型三個方面解釋了操作系統為實現高性能I/O做了哪些事情，通過提高CPU使用效率，減少內存複製是提高性能的關鍵點。

參考文檔
7. 《操作系統精髓與設計原理(第5版)》
8. 《WINDOWS內核原理與實現》
微信掃一掃二維碼關注訂閱號傑哥技術分享
出處：
作者：傑哥很忙
本文使用「CC BY 4.0」創作共享協議。歡迎轉載，請在明顯位置給出出處及鏈接。

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

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

※網頁設計一頭霧水??該從何著手呢? 找到專業技術的網頁設計公司，幫您輕鬆架站!

※想知道最厲害的台北網頁設計公司推薦、台中網頁設計公司推薦專業設計師”嚨底家”!!
2020-02-14
氫燃料沒未來？Honda改拚電動車，明年搶攻歐陸

本田汽車（Honda）計畫於2018年一次推出兩款全電動車，且宣告未來將以電動車作為發展主軸。

根據《BusinessInsider》報導表示，本田六月就已表示將打造中國專屬電動車，8月29日發佈的聲明稿則指明另一款電動車是為歐洲客戶所設計。

本田說，新城市電動概念車（Urban EV Concept）將在九月法蘭克福車展上亮相，這也是本田在歐洲第一款電動車，肩負打開歐洲電動車市場的任務。

目前本田僅有一款純電動車在美國上市，不過本田表示，2030年旗下三分之二的車型都要電動化，達成目標的方法將在法蘭克福車展上對外說明。

本田與日本同業豐田（Toyota）此前主要開發氫燃料車與混和動力車，但由上述可知本田策略已經轉向發展電動車，可能引領其它日本車廠跟進。

（本文內容由授權使用。圖片出處：）

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

【其他文章推薦】

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

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

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

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

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

2020-02-13
馬斯克遇程咬金，第一台電動卡車不姓特斯拉

馬斯克近日才剛在推特上表示，近期將推出全電動的半掛式卡車，令市場相當的看好，然而如今第一個達成創舉的廠商，卻不是特斯拉。

特斯拉將在9 月發表載貨用的全電動半掛卡車（Tesla Semi）原型，雖然業界一直在質疑其負載及續航力，不過若成功的話也等於是對市場投下一個震撼彈。畢竟相較汽柴油車的引擎，電動車的維護成本較低，假如電池成本也持續降低，在配合環保法規的情況下，相當的有市場。

不過如今一家原本專做柴油及天然氣發動機的龍頭廠商康明斯（Cummins）推出了相當於第7 級重型卡車的電動車，其配備了先進的140 kWh 的電池組，並預計將於2019 年開始交付給客戶。這台卡車被命名為AEOS，應該是取意於希臘神話中替太陽神阿波羅拉車的火焰飛馬。

AEOS in motion: Raw footage of our first fully electric heavy duty demonstration Urban Hauler Tractor.

— Cummins Inc. (@Cummins)

且Cummins 也能僅提供動力系統總成，將可應用於電動巴士等其他車型。電動巴士製造商　Proterra 的總裁Ryan Popple 指出，這是相當令人震驚的消息，是邁向完全電氣化交通的一大步，隨著各國正積極修訂法規鼓勵電動車，目前這方面的市場相當大，巴士及校車等訂單幾乎滿載。

據Cummins 表示，這輛電動卡車動力總成與一般柴油車相當，可牽引4,4000 磅的重量，近20 噸，單次充電可行駛約100 英里，並可加掛額外的電池組將行程擴展到300 英里，還可安裝太陽能電池板，還有再生制軔（Regenerative brake）及低阻力輪胎的研發將可望進一步提昇其行駛距離。目前的充電站設計，將可在1 小時內充滿，而Cummins 的目標是希望，能夠在2020 年以前，將充電時間縮短到20 分鐘。

不過Cummins 的執行長Thomas Linebarger 也坦言，依目前電池科技的進展，第7 級重型卡車的應用已是極限，第8 級的重型拖車目前仍遙遙無期。所以也將推出同型的高效率柴油引擎及混合動力車款。儘管有所突破，但目前電動系統仍較適用於商用卡車，Cummins 一舉超越特斯拉也同樣宣示了將進軍此市場的意圖。

當然特斯拉就算被搶得頭籌，也不代表無法反擊，據之前透露的消息，其產品可能將定位於行程達200-300 英里的運輸範圍，性能可能比AEOS 還強。但Cummins 此舉或許更大意義在於強調，他們這些傳統龍頭，儘管受到挑戰，仍有相當強的底蘊，在不同動力系統的研發上，有超乎市場想像的潛力。

（合作媒體：。圖片出處：）

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

【其他文章推薦】

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

※網頁設計一頭霧水??該從何著手呢? 找到專業技術的網頁設計公司，幫您輕鬆架站!

※想要讓你的商品成為最夯、最多人討論的話題?網頁設計公司讓你強力曝光

※想知道最厲害的台北網頁設計公司推薦、台中網頁設計公司推薦專業設計師”嚨底家”!!

2020-02-13
電動車需求高、鋰價漲，澳洲Orocobre擴產

澳洲礦商Orocobre Limited聲稱鋰礦需求強、報價攀高，將擴產碳酸鋰，消息傳來激勵股價在8月31日飆高。

Barron’s.com、The Australian報導，Orocobre公布的全年度財報首度轉虧為盈，淨利來到1,940萬美元、遠優於去年的淨損2,200萬美元，而用於電池等工業產品的碳酸鋰，也將擴充產能。

Orocobre估計，2018會計年度的碳酸鋰產出，將從2017年度的11,862公噸擴充到14,000公噸，而2018年度的碳酸鋰均價，每噸則有望超過1萬美元。該公司在2017年度共計售出12,296公噸的碳酸鋰，每噸均價為9,763美元，4-6月當季的均價則是10,696美元，生產成本為3,710美元。

執行長Richard Seville說，全球鋰礦市場的基本面依舊穩健，不但需求強勁、供給吃緊，報價也相當具有吸引力。為了滿足特定電池夥伴的需求，Orocobre會分階段擴充Olaroz鋰礦廠的產能。

依據Orocobre計畫，位於阿根廷的Olaroz鋰礦廠，產能料將倍增。另外，該公司還會跟豐田通商（Toyota Tsusho Corporation）一同打造一座廠房，預計將年產10,000公噸的氫氧化鋰（lithium hydroxide）。

採用鋰電池的電動車需求日增、電池榮景全無降溫跡象，隨著鋰礦供給愈來愈難尋、未來恐有短缺之虞。

OilPrice.com 8月22日報導，特斯拉（Tesla Inc.）內華達州的Gigafactory超級電池廠，估計一年要生產50萬顆車用電池，另外還有諸多車用鋰電池廠也在加緊趕工，顯示未來鋰礦的需求將只增不減，且大部分將來自中國。全球如今有51%的鋰電池是在中國生產，僅10%產自美國。據統計，大陸業者規劃中的車用電池廠，產能預料會在2021年底前達到120億瓦小時（GW hours），是特斯拉內華達州電池廠的三倍之多。

然而，採集鋰礦在技術上其實有相當難度，成本不但高昂，且由於採礦過程牽涉到蒸餾法，產出也頗難預估。Orocobre在阿根廷北部設立的新礦場，鋰礦產出就比預期少逾20%。

假如鋰礦產能無法順利擴充、或是新開礦場產出不如預期，那麼電動車的榮景可能因而受阻。Electrek報導，福斯汽車（Volkswagen）研發部主管Ulrich Eichhorn 6月底就曾預估，業界需要多達40座規模跟特斯拉Gigafactory類似的超大電池廠，才能滿足電動車需求，假如新建礦場、工廠無法如期上線，那麼市場恐陷入短缺。

（本文內容由授權使用。圖片出處：public domain CC0）

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

【其他文章推薦】

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

※評比前十大台北網頁設計、台北網站設計公司知名案例作品心得分享

※智慧手機時代的來臨，RWD網頁設計已成為網頁設計推薦首選

2020-02-13
電動車商機大、日電池材料商瘋增產

日本鋰離子電池材料商W-Scope於9月4日發布新聞稿宣布，因來自電動車的需求急增，加上家電產品、電動工具需求持續強勁，故決議投下200億日圓擴增鋰離子電池關鍵材料分隔膜（separator）產能，計畫在南韓子公司W-SCOPE CHUNGJU PLANT CO., LTD.增設四條產線，其中第12-13號產線預計2019年下半年量產、第14-15號產線預計2020年上半年量產。

W-Scope指出，除上述四條新產線之外，該公司之前已進行增產投資，興建第8-11號分隔膜產線，其中第8-9號產線預計2017年下半年量產、第10-11號預計2018年上半年量產，而待上述八條產線全數導入量產後，2020年末整體分隔膜產能將擴增至2016年末的3.6倍水準。

在電動車市場持續擴大的背景下，日本其他鋰電池材料廠紛紛傳出增產消息。日刊工業新聞8月1日報導，Toray計畫在2020年結束前總計砸下約1,200億日圓擴增車用鋰離子電池關鍵材料分隔膜產能，目標將分隔膜年產能擴增至19.5億平方公尺、將達現行的約五倍。

日刊工業新聞6月23日報導，因車廠加快電動車的研發腳步、帶動電池材料市場成長速度超乎預期，故旭化成（Asahi Kasei）計畫上修鋰離子電池關鍵材料分隔膜的增產計畫，目標在2020年結束前將分隔膜年產能最高擴增至15億平方公尺、將達現行的2.5倍，且將遠高於原先規劃的11億平方公尺目標，期望藉由積極投資、鞏固全球龍頭位置。預估追加擴產所需的投資額約300億日圓。

鋰離子電池四大關鍵材料分別為正極材、負極材、分隔膜和電解液，而這些電池材料皆由日系廠商握有高市佔率，其中，在全球分隔膜市場上，旭化成為龍頭廠、Toray緊追在後。

根據日本市調機構富士經濟（Fuji Keizai）預估，2020年全球分隔膜市場規模將增至3,000億日圓、將達2015年的2倍水準，而電動車、混合動力車等車用用途是推動分隔膜需求急增的最大功臣，預估2020年車用分隔膜佔整體市場比重將達約45%。

富士經濟6月22日公布調查報告指出，預估2030年時電動車年銷售量將增至407萬台、超越混合動力車，且之後雙方的差距將持續擴大。富士經濟預估，在中國需求增加加持下，2035年電動車全球銷售量將擴大至630萬台、將達2016年的13.4倍。

（本文內容由授權使用。首圖為鋰電池材料分隔膜，出處：MoneyDJ）

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

【其他文章推薦】

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

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

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

2020-02-13

分類: 3C資訊

BINLOG 常見操作命令

通過 MySQL Cli 查詢 BINLOG 信息

通過 mysqlbinlog 查詢 BINLOG 信息

MySQL Cli 和 mysqlbinlog 工具之間的比較

使用 BINLOG 恢複數據

準備數據

刪除表或者數據

數據的恢復

總結

參考

1、前言

2、具體方式

2.1、執行舊錶的創建SQL來創建表

2.2、使用like創建新表（僅包含表結構）

2.3、使用as來創建新表（包含數據）

2.4、使用like+insert+select創建原表的副本（推薦）

1、發送消息軌跡流程

1.1 DefaultMQProducer構造函數

1.2 SendMessageTraceHookImpl鈎子函數

1.2.1 SendMessageTraceHookImpl類圖

1.2.2 源碼分析SendMessageTraceHookImpl

1.2.2.1 sendMessageBefore

1.2.2.2 sendMessageAfter

1.3 TraceDispatcher實現原理

1.3.1 TraceDispatcher構造函數

1.3.2 getAndCreateTraceProducer詳解

1.3.3 start

1.3.4 AsyncRunnable

1.3.5 AsyncAppenderRequest#sendTraceData

1.3.6 TraceDataEncoder#encoderFromContextBean

1.3.6.1 PUB(消息發送)

1.3.6.2 SubBefore(消息消費之前)

1.3.2.3 SubAfter（消息消費后）

2、 消息軌跡數據如何存儲

2.1 使用系統默認的主題名稱

2.2 用戶自定義消息軌跡主題

1 目標檢測簡介

2 lmdb數據製作

2.1 VOC數據製作

2.2 lmdb文件生成

高性能網絡通訊原理

前言

I/O訪問

I/O模型

同步阻塞

select模型/poll模型

epoll模型

異步I/O模型

I/O線程模型

Reactor模型

Proactor模型

總結

參考文檔

2、具體方式　

　　2.1、執行舊錶的創建SQL來創建表

　　2.2、使用like創建新表（僅包含表結構）

　　2.3、使用as來創建新表（包含數據）

　　2.4、使用like+insert+select創建原表的副本（推薦）

2、消息軌跡數據如何存儲