隨着中國國際影響力的日益提升，漢語學習的需求與日俱增，為此，教育部、國家語委在《國家語言文字事業“十三五”發展規劃》中明確提出要“建設適應面廣、影響力大、權威性強的全球中文學習網絡平台”。10 月 25 日，在教育部、國家語委的指導下，由科大訊飛研發的全球中文學習平台正式上線。

　　該平台上線發布儀式在京舉行。教育部副部長、國家語委主任田學軍，北京市人民政府副秘書長韓耕、科大訊飛股份有限公司董事長劉慶峰和人民教育出版社社長黃強共同為平台發布舉行了啟動儀式，200 多位中外嘉賓見證了全球中文學習平台（www.chinese-learning.cn）的正式上線。

　　針對海外學習者的“譯學中文”模塊，學習者可以通過語音或文本輸入其母語內容，實時翻譯出中文並自動分句。學習者學習每個語句的標準音並錄音跟讀，系統會實時反饋評價，指出發音問題；針對錯誤字詞，可以反覆學習，直到掌握正確中文發音。

　　這是科大訊飛承建國家語委的又一個重大項目！2004 年，科大訊飛承擔了國家語委“十五”重點科研項目“智能語音技術在普通話輔助學習中的應用研究”；2016 年，承擔國家語委“十三五”重大科研項目“智能語音及人工智能技術在語言學習中的應用研究”。目前，上述兩大項目均已成功落地，並取得了良好的社會效益。

　　全球中文學習平台，匯聚各類中文學習資源，以更好地為廣大中文學習者提供優質服務為宗旨，於 2016 年底啟動建設，是落實《國家語言文字事業“十三五”發展規劃》相關任務要求的具體舉措。在教育部、國家語委與科大訊飛的共同努力下，歷經兩年多時間的不斷完善和改進，平台建設取得积極成效，相關基礎研究取得重要進展，為平台提供了堅實技術保障。其中智能語音、智能寫作和批改等關鍵技術研究成果在中小學語言能力評價、少數民族國家通用語言學習等方面得到實際應用。平台示範功能已分別在“砥礪奮進的五年”大型成就展、第二屆語博會、第十二屆孔子學院大會等不同場合進行展示，得到了各方好評。

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

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

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

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

接上回書

書接，分享了Class文件的主要構成，同時也詳細分析了魔數、次版本號、主版本號、常量池集合、訪問標誌的構造，接下來我們就繼續學習。

歡迎關注微信公眾號：萬貓學社，每周一分享Java技術乾貨。

類索引和父類索引

類索引（this_class）和父類索引（super_class）都是一個u2類型的數據，類索引用於確定這個類的全限定名，父類索引用於確定這個類的父類全限定名。由於java語言不允許多重繼承，所以父類索引只有一個。

類索引和父類索引各自指向常量池中類型為CONSTANT_Class_info的類描述符，再通過類描述符中的索引值找到常量池中類型為CONSTANT_Utf8_info的字符串。再來看一下之前的Class文件例子：

歡迎關注微信公眾號：萬貓學社，每周一分享Java技術乾貨。

結合之前javap分析出來的常量池內容：

   #3 = Class         #17        // OneMoreStudy
   #4 = Class         #18        // java/lang/Object
  #17 = Utf8          OneMoreStudy
  #18 = Utf8          java/lang/Object

類索引為0x0003，去常量池裡找索引為3的類描述符，類描述符中的索引為17，再去找索引為17的字符串，就是“OneMoreStudy”。

父類索引為0x0004，去常量池裡找索引為4的類描述符，類描述符中的索引為18，再去常量池裡找索引為18的字符串，就是“java/lang/Object”。

歡迎關注微信公眾號：萬貓學社，每周一分享Java技術乾貨。

接口索引集合

接口索引集合（interface）是一組u2類型的數據的集合，由於java語言允許實現多個接口，所以接口索引也有多個，它們按照implements語句后的接口順序從左到右依次排列在接口索引集合中。接口索引集合的第一項數據是接口集合計數值（interfaces_count），表示有多少接口索引。如果該類沒有實現任何接口，那麼該計數值為0，後面的接口索引表不佔任何字節。之前的例子OneMoreStudy類沒有實現任何接口，所以接口集合計數值就是0，如下圖：

歡迎關注微信公眾號：萬貓學社，每周一分享Java技術乾貨。

字段表集合

字段表（field_info）是用來描述接口或類中聲明的變量。包括類級變量（靜態變量）和實例級變量（成員變量），但是不包括在方法內部聲明的局部變量。具體結構如下錶：

字段表中的access_flags，和類的access_flags是非常類似的，但是標識和含義是不一樣的。具體如下錶：

歡迎關注微信公眾號：萬貓學社，每周一分享Java技術乾貨。

這裏提到了簡單名稱、描述符，和全限定名有什麼區別呢？稍微說一下。

簡單名稱是沒有類型和參數修飾的方法或字段名稱，比如OneMoreStudy類中的number字段和plusOne()方法的簡單名稱分別是“number”和“plusOne”。

全限定名是把類全名中的“.”替換成“/”就可以了，比如java.lang.Object類的全限定名就是“java/lang/Object”。

描述符是用來描述字段的數據類型、方法的參數列表（包括數量、類型以及順序）和返回值。基礎數據類型和無返回的void類型都有一個大寫字母表示，對象類型用字符L加對象的全限定名來表示，如下錶：

對於數組類型，每一維度使用一個前置的“[”字符來描述，比如java.lang.Object[][]的二維數據，就是“[[Ljava/lang/Object”。在描述方法時，按照先參數列表，后返回值的順序描述，參數列表按照嚴格順序放在“()”值中，比如boolean equals(Object anObject)，就是“(Ljava/lang/Object)B”。

歡迎關注微信公眾號：萬貓學社，每周一分享Java技術乾貨。

再來看一下之前的Class文件例子：

OneMoreStudy類中只有一個字段number，所以字段計數值為0x0001。字段number只被private修飾，沒有其他修飾，所以字段的訪問標誌位為0x0002。字段的簡單名稱索引是0x0005，去常量池中找索引為5的字符串，為“number”。字段的描述符索引為0x0006，去常量池中找索引為6的字符串，為“I”，是基本類型int。以下是常量池相關內容：

   #5 = Utf8          number
   #6 = Utf8          I

字段number的屬性計數值為0x0000，也就是沒有需要額外描述的信息。

字段表集合中不會列出從父類或者父接口中繼承而來的字段，但有可能列出原版Java代碼中沒有的字段，比如在內部類中為了保持對外部類的訪問性，會自動添加指向外部類實例的字段。

歡迎關注微信公眾號：萬貓學社，每周一分享Java技術乾貨。

方法表集合

方法表的結構和字段表的是一樣的，也是依次包括了訪問標誌（access_flags）、名稱索引（name_index）、描述符索引（descriptor_index）和屬性表集合（attributes）。具體如下錶：

對於方法的訪問標誌，所有標誌位和取值如下錶：

方法中的Java代碼，經過編譯器編程成字節碼指令后，放在方法屬性表集合中一個名為“Code”的屬性里，後面會有更多分享。

歡迎關注微信公眾號：萬貓學社，每周一分享Java技術乾貨。

再來看一下之前的Class文件例子：

方法計算值為0x0003，表示集合中有兩個方法（編譯器自動添加的無參構造方法和源碼中的plusOne方法）。第一個方法的訪問標誌是0x0001，表示只有ACC_PUBLIC標誌為true。

名稱索引為0x0007，在常量池中為索引為7的字符串為“ ”，這就是編譯器自動添加的無參構造方法。描述符索引為0x0008，在常量池中為索引為7的字符串為“()V”，方法的屬性計數值為0x0001，表示該方法有1個屬性，屬性名稱索引為0x0009，在常量池中為索引為7的字符串為“Code”。以下是常量池相關內容：

   #7 = Utf8          <init>
   #8 = Utf8          ()V
   #9 = Utf8          Code

歡迎關注微信公眾號：萬貓學社，每周一分享Java技術乾貨。

屬性表集合

屬性表（attribute_info）在前面的分享中出現了幾次，在Class文件、字段表、方法表都可以有自己的屬性表集合，用來描述某些場景下特有的信息。

屬性表不在要求具有嚴格的順序，並且只要不與已有的屬性名重複，任何人實現的編譯器都可以寫入自己定義的屬性信息，Java虛擬機在運行時會忽略掉它不認識的屬性。

我總結了一些比較常見的屬性，如下錶：

對於每個屬性，它的名稱都從常量池中引用一個CONSTANT_Utf8_info類型的常量，而屬性值的結構則是完全自定義的，只需要用一個u4類型來說明屬性值所佔的位數就可以了。具體結構如下：

歡迎關注微信公眾號：萬貓學社，每周一分享Java技術乾貨。

總結

Class文件主要由魔數、次版本號、主版本號、常量池集合、訪問標誌、類索引、父類索引、接口索引集合、字段表集合、方法表集合和屬性表集合組成。隨着JDK版本的不斷升級，Class文件結構也在不斷更新，學習之路,永不止步。

歡迎關注微信公眾號：萬貓學社，每周一分享Java技術乾貨。

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

※收購3c,收購IPHONE,收購蘋果電腦-詳細收購流程一覽表

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

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

※公開收購3c價格,不怕被賤賣!

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

       在上一篇 文章中我們學習了TreeGrid的一些基礎知識，接下我們來創建我們開發組織管理功能用到的一些類。關於如何創建類我們之前的文章中已經寫了很多了，這裡會有些簡略。

四、定義應用服務接口需要用到的DTO類

      為了在進行查詢時使用, PagedOrgResultRequestDto被用來將模塊數據傳遞到基礎設施層.

       1. 在Visual Studio 2017的“解決方案資源管理器”中，右鍵單擊“ABP.TPLMS.Application”項目，在彈出菜單中選擇“添加” > “新建文件夾”，並重命名為“Orgs”

      2. 使用鼠標右鍵單擊我們剛才創建的“Orgs”文件夾，在彈出菜單中選擇“添加” > “新建文件夾”，並重命名為“Dto”。

      3．右鍵單擊“Dto”文件夾，然後選擇“添加” > “類”。 將類命名為 Paged OrgResultRequestDto，然後選擇“添加”。代碼如下。

using Abp.Application.Services.Dto;
using Abp.Application.Services.Dto;
using System;
using System.Collections.Generic;
using System.Text;
 

namespace ABP.TPLMS.Orgs.Dto
{

public class PagedOrgResultRequestDto : PagedResultRequestDto
    {
        public string Keyword { get; set; }
    }
}

      4．右鍵單擊“Dto”文件夾，然後選擇“添加” > “類”。 將類命名為 OrgDto，然後選擇“添加”。代碼如下。

using Abp.Application.Services.Dto;
using Abp.AutoMapper;
using ABP.TPLMS.Entitys;
using System;
using System.Collections.Generic;
using System.Text;
 

namespace ABP.TPLMS.Orgs.Dto
{

    [AutoMapFrom(typeof(Org))]
    public class OrgDto : EntityDto<int>
    {

        int m_parentId = 0;
        public string Name { get; set; }       

        public string HotKey { get; set; }
        public int ParentId { get { return m_parentId; } set { m_parentId = value; } }       

        public string ParentName { get; set; }
        public bool IsLeaf { get; set; }
        public bool IsAutoExpand { get; set; }       

        public string IconName { get; set; }
        public int Status { get; set; }
        public int Type { get; set; }      
        public string BizCode { get; set; }       

        public string CustomCode { get; set; }
        public DateTime CreationTime { get; set; }
        public DateTime UpdateTime { get; set; }

        public int CreateId { get; set; }
        public int SortNo { get; set; }
        public int _parentId {
            get { return m_parentId; }          

        }
    }
}

 

      5．右鍵單擊“Dto”文件夾，然後選擇“添加” > “類”。 將類命名為 CreateUpdateOrgDto，然後選擇“添加”。代碼如下。

using Abp.Application.Services.Dto;
using Abp.AutoMapper;
using ABP.TPLMS.Entitys;
using System;
using System.Collections.Generic;
using System.ComponentModel.DataAnnotations;
using System.Text;

 
namespace ABP.TPLMS.Orgs.Dto
{

    [AutoMapTo(typeof(Org))]
    public class CreateUpdateOrgDto : EntityDto<int>
    {

        public string Name { get; set; }
        [StringLength(255)]
        public string HotKey { get; set; }

        public int ParentId { get; set; }

        [Required]
        [StringLength(255)]
        public string ParentName { get; set; }

        public bool IsLeaf { get; set; }
        public bool IsAutoExpand { get; set; }

        [StringLength(255)]
        public string IconName { get; set; }

        public int Status { get; set; }
        public int Type { get; set; }

        [StringLength(255)]
        public string BizCode { get; set; }

        [StringLength(100)]
        public string CustomCode { get; set; }
        public DateTime CreationTime { get; set; }
        public DateTime UpdateTime { get; set; }

        public int CreateId { get; set; }
        public int SortNo { get; set; } 

    }
}

 

五、定義IOrgAppService接口

        6． 在Visual Studio 2017的“解決方案資源管理器”中，鼠標右鍵單擊“Org”文件夾，然後選擇“添加” > “新建項”，在彈出對話框中選擇“接口”。為應用服務定義一個名為 IOrgAppService 的接口。代碼如下。

using System;
using System.Collections.Generic;
using System.Text;
using Abp.Application.Services;
using ABP.TPLMS.Orgs.Dto; 

 

namespace ABP.TPLMS.Orgs
{
  public  interface IOrgAppService : IAsyncCrudAppService<//定義了CRUD方法

             OrgDto, //用來展示組織信息
             int, //Org實體的主鍵
             PagedOrgResultRequestDto, //獲取組織信息的時候用於分頁
             CreateUpdateOrgDto, //用於創建組織信息
             CreateUpdateOrgDto> //用於更新組織信息

    {
    }
}

      六、實現IOrgAppService

       7．在Visual Studio 2017的“解決方案資源管理器”中，右鍵單擊“Org”文件夾，然後選擇“添加” > “新建項”，在彈出對話框中選擇“類”。為應用服務定義一個名為 OrgAppService 的服務類。代碼如下。

using Abp.Application.Services;
using Abp.Domain.Repositories;
using ABP.TPLMS.Entitys;
using ABP.TPLMS.Orgs.Dto;
using System;
using System.Collections.Generic;
using System.Text;
 

namespace ABP.TPLMS.Orgs
{

    public class OrgAppService : AsyncCrudAppService<Org, OrgDto, int, PagedOrgResultRequestDto,
                            CreateUpdateOrgDto, CreateUpdateOrgDto>, IOrgAppService 

    {
        public OrgAppService(IRepository<Org, int> repository)

            : base(repository)

        { 

        }
    }
}

七 創建OrgController繼承自TPLMSControllerBase

      1. 在Visual Studio 2017的“解決方案資源管理器”中，右鍵單擊在領域層“ABP.TPLMS.Web.Mvc”項目中的Controller目錄。 選擇“添加” > “新建項…”。如下圖。

     2. 在彈出對話框“添加新項-ABP.TPLMS.Web.Mvc”中選擇“控制器類”，然後在名稱輸入框中輸入“OrgsController”，然後點擊“添加”按鈕。

     3.在OrgsController.cs文件中輸入如下代碼，通過構造函數注入對應用服務的依賴。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Abp.AspNetCore.Mvc.Authorization;
using Abp.Web.Models;
using ABP.TPLMS.Controllers;
using ABP.TPLMS.Orgs;
using ABP.TPLMS.Orgs.Dto;
using ABP.TPLMS.Web.Models.Orgs;
using Microsoft.AspNetCore.Mvc;


namespace ABP.TPLMS.Web.Controllers
{
    [AbpMvcAuthorize]
    public class OrgsController : TPLMSControllerBase
    {
        private readonly IOrgAppService _orgAppService;
        private const int MAX_COUNT= 1000;
        
        public OrgsController(IOrgAppService orgAppService)
        {
            _orgAppService = orgAppService;
        }
        [HttpGet]
        // GET: /<controller>/
        public IActionResult Index()
        {
            return View();
        }
        [DontWrapResult]
        [HttpPost]
        public string List()
        {         
            PagedOrgResultRequestDto paged = new PagedOrgResultRequestDto();
            paged.MaxResultCount = MAX_COUNT;
           var userList = _orgAppService.GetAll(paged).GetAwaiter().GetResult().Items;
            int total = userList.Count;
            var json = JsonEasyUI(userList, total);
            return json;
        }       
    }
}

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

※收購3c,收購IPHONE,收購蘋果電腦-詳細收購流程一覽表

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

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

※公開收購3c價格,不怕被賤賣!

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

中油從 107 年起，依據經濟部智慧電動機車能源補充設施普及計畫(公建計畫)於加油站建置電動機車充電、換電設備，去年僅睿能 Gogoro 參與設 144 站，今年增加光陽 Ionex，兩家採公開競標方式，由光陽 Ionex 獲得 3 標共 146 站，睿能 Gogoro 得到 1 標 48 站；其中，睿能 Gogoro 扣除合約到期站數，再加上先前自建設站，在台灣中油加油站中有 285 站可為消費者服務。

經濟部工業局委託台灣銀行辦理共同供應契約設備系統招標，其中電池交換設備系統得標廠商都需符合安全規範審核，108 年換電設備系統之得標廠商共有睿能 Gogoro 及光陽 Ionex 兩家，中油考量消費者的便利性及設備使用公平性兩大原則，於採購系統設備前即採取公開方式辦理合作經營招標。

中油表示，北部地區有 52 站與睿能 Gogoro 合作的換電站，今年合約陸續到期，睿能 Gogoro 依據契約與決標結果遷移設備，為執行公建計畫，中油將此52站列入今年建置194站換電站的地點，公建換電站依各縣市站點打散，再平均分4案開標，讓每一標案站點均涵蓋各縣市，投標結果為光陽 Ionex 獲得 3 標（146 站）、睿能 Gogoro 得 1 標（48 站），原先台灣中油與睿能 Gogoro 合約到期的52站中，有38站改由光陽Ionex得標。

中油指出，上述政府電動機車充、換電站公建計畫，於 107 年已執行 161 站，其中 144 座換電站均為睿能 Gogoro 所有，另有充電 17 站，今年預計建置 216 座充換電站，其中 194 座為換電站，光陽 Ionex 有 146 站、睿能 Gogoro 48 站，另有充電 22 站，以總數來說，再加上睿能 Gogoro 之前的自建站 79 站，其仍在台灣中油有 285 站，光陽 Ionex 有 146 站為消費者服務，明年亦將配合政府預算持續建置更多站點，協助政府綠能政策推動。

（本文內容由 授權使用。首圖來源：Gogoro）

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

【其他文章推薦】

※收購3c,收購IPHONE,收購蘋果電腦-詳細收購流程一覽表

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

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

※公開收購3c價格,不怕被賤賣!

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

我們在提到了對Win10 API的調用，但仍存在無法在WPF中使用UWP控件的問題，雖然都是XAML控件，但卻是兩套命名空間下的同名類型，無法混用。
人總會被現實打敗，強大如某軟也得向生活低頭，UWP一直沒有起色，某軟的老大又一心去搞Azure。Windows平台的重振，似乎想走回頭路，從1903版本開始，支持在.NET Framwork的WPF和WinForm工程中，直接使用部分的UWP控件了。首當其沖的，就是有點騷包的InkToolbar和InkCanvas。

接下來我們就來試試如何在WPF工程中，使用UWP的InkToolbar和InkCanvas。
首先創建一個空的WPF工程，完成后，在Nuget的搜索界面填入 Microsoft.Toolkit.Wpf.UI.Controls ，選中第一個進行安裝。

完成安裝后，打開MainWindow.xaml，添加對命名空間的引用xmlns:Controls=”clr-namespace:Microsoft.Toolkit.Wpf.UI.Controls;assembly=Microsoft.Toolkit.Wpf.UI.Controls”。接着就可以在<Grid>節點中添加UWP版本的InkToolbar和InkCanvas控件了。

<Window x:Class="WPFInkSample.MainWindow"
        xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
        xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
        xmlns:d="http://schemas.microsoft.com/expression/blend/2008"
        xmlns:mc="http://schemas.openxmlformats.org/markup-compatibility/2006"
        xmlns:local="clr-namespace:WPFInkSample"
        xmlns:Controls="clr-namespace:Microsoft.Toolkit.Wpf.UI.Controls;assembly=Microsoft.Toolkit.Wpf.UI.Controls"
        mc:Ignorable="d"
        Title="MainWindow" Height="450" Width="800">
    <Grid >
        <Grid.RowDefinitions>
            <RowDefinition Height="Auto"/>
            <RowDefinition Height="*"/>
        </Grid.RowDefinitions>
        <Controls:InkToolbar TargetInkCanvas="{x:Reference myInkCanvas}" Grid.Row="0" />
        <Controls:InkCanvas x:Name="myInkCanvas" Grid.Row="1" />
    </Grid>
</Window>

同時我們還需要在MainWindow.xaml.cs中設置InputDeviceTypes。

    public partial class MainWindow : Window
    {
        public MainWindow()
        {
            InitializeComponent();
            this.myInkCanvas.InkPresenter.InputDeviceTypes = CoreInputDeviceTypes.Mouse | CoreInputDeviceTypes.Pen;
        }
    }

然後按下F5運行，某軟的騷操作來了……因為僅在1903以後的版本才支持這種騷操作（10.0.18226是稍早的preview版），所以需要做額外的處理才可以。

我們這裡有兩種選擇，一是通過來打包這個WPF程序，然後在Packaging工程的屬性里，將Target version和Minimum version同時設置為Windows 10, version 1903 (10.0.18362) 。這是MSDN上推薦的標準做法，這樣做的好處在於，打包好的程序可以直接上傳MS Store。
如果我們想保持exe的可執行文件形式，還有另一種做法，在Project文件上右鍵點擊Add->New Item，添加一個manifest文件。
在這個文件中，找到<!–Windows 10–>，然後做如下編輯：

  <compatibility xmlns="urn:schemas-microsoft-com:compatibility.v1">
    <application>
      <!-- A list of the Windows versions that this application has been tested on
           and is designed to work with. Uncomment the appropriate elements
           and Windows will automatically select the most compatible environment. -->
  
      <!-- Windows Vista -->
      <!--<supportedOS Id="{e2011457-1546-43c5-a5fe-008deee3d3f0}" />-->
  
      <!-- Windows 7 -->
      <!--<supportedOS Id="{35138b9a-5d96-4fbd-8e2d-a2440225f93a}" />-->
  
      <!-- Windows 8 -->
      <!--<supportedOS Id="{4a2f28e3-53b9-4441-ba9c-d69d4a4a6e38}" />-->
  
      <!-- Windows 8.1 -->
      <!--<supportedOS Id="{1f676c76-80e1-4239-95bb-83d0f6d0da78}" />-->
  
      <!-- Windows 10 -->
      <maxversiontested Id="10.0.18362.0"/>
      <supportedOS Id="{8e0f7a12-bfb3-4fe8-b9a5-48fd50a15a9a}" />
  
    </application>
  </compatibility>

保存后，再通過F5運行，即發現一切正常，不在出現之前的運行時錯誤了。
本篇我們介紹了如何在WPF工程中使用UWP InkToolbar和InkCavas。因為這個功能僅在1903后的版本支持，所以下一篇我們會介紹如何簡單地判斷Win10 API 版本，在運行時判斷是否執行對應版本的代碼。
Github：

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

※收購3c,收購IPHONE,收購蘋果電腦-詳細收購流程一覽表

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

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

※公開收購3c價格,不怕被賤賣!

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

介紹

生產者-消費者模型用於解耦生產者與消費者，平衡兩者之間的能力不平衡，該模型廣泛應用於各個系統中，Hudi也使用了該模型控制對記錄的處理，即記錄會被生產者生產至隊列中，然後由消費者從隊列中消費，更具體一點，對於更新操作，生產者會將文件中老的記錄放入隊列中等待消費者消費，消費后交由HoodieMergeHandle處理；對於插入操作，生產者會將新記錄放入隊列中等待消費者消費，消費后交由HandleCreateHandle處理。

入口

前面的文章中提到過無論是HoodieCopyOnWriteTable#handleUpdate處理更新時直接生成了一個SparkBoundedInMemoryExecutor對象，還是HoodieCopyOnWriteTable#handleInsert處理插入時生成了一個CopyOnWriteLazyInsertIterable對象，再迭代時調用該對象的CopyOnWriteLazyInsertIterable#computeNext方法生成SparkBoundedInMemoryExecutor對象。最後兩者均會調用SparkBoundedInMemoryExecutor#execute開始記錄的處理，該方法核心代碼如下

  public E execute() {
    try {
      ExecutorCompletionService<Boolean> producerService = startProducers();
      Future<E> future = startConsumer();
      // Wait for consumer to be done
      return future.get();
    } catch (Exception e) {
      throw new HoodieException(e);
    }
  }

該方法會啟動所有生產者和單個消費者進行處理。

Hudi定義了BoundedInMemoryQueueProducer接口表示生產者，其子類實現如下

• FunctionBasedQueueProducer，基於Function來生產記錄，在合併日誌log文件和數據parquet文件時使用，以便提供RealTimeView。
• IteratorBasedQueueProducer，基於迭代器來生產記錄，在插入更新時使用。

定義了BoundedInMemoryQueueConsumer類表示消費者，其主要子類實現如下

• CopyOnWriteLazyInsertIterable$CopyOnWriteInsertHandler，主要處理CopyOnWrite表類型時的插入。
  • MergeOnReadLazyInsertIterable$MergeOnReadInsertHandler，主要處理MergeOnRead

表類型時的插入，其為CopyOnWriteInsertHandler的子類。

• CopyOnWriteLazyInsertIterable$UpdateHandler，主要處理CopyOnWrite表類型時的更新。

整個生產消費相關的類繼承結構非常清晰。

對於生產者的啟動，startProducers方法核心代碼如下

  public ExecutorCompletionService<Boolean> startProducers() {
    // Latch to control when and which producer thread will close the queue
    final CountDownLatch latch = new CountDownLatch(producers.size());
    final ExecutorCompletionService<Boolean> completionService =
        new ExecutorCompletionService<Boolean>(executorService);
    producers.stream().map(producer -> {
      return completionService.submit(() -> {
        try {
          preExecute();
          producer.produce(queue);
        } catch (Exception e) {
          logger.error("error producing records", e);
          queue.markAsFailed(e);
          throw e;
        } finally {
          synchronized (latch) {
            latch.countDown();
            if (latch.getCount() == 0) {
              // Mark production as done so that consumer will be able to exit
              queue.close();
            }
          }
        }
        return true;
      });
    }).collect(Collectors.toList());
    return completionService;
  }

該方法使用CountDownLatch來協調生產者線程與消費者線程的退出動作，然後調用produce方法開始生產，對於插入更新時的IteratorBasedQueueProducer而言，其核心代碼如下

  public void produce(BoundedInMemoryQueue<I, ?> queue) throws Exception {
    ...
    while (inputIterator.hasNext()) {
      queue.insertRecord(inputIterator.next());
    }
    ...
  }

可以看到只要迭代器還有記錄（可能為插入時的新記錄或者更新時的舊記錄），就會往隊列中不斷寫入。

對於消費者的啟動，startConsumer方法的核心代碼如下

  private Future<E> startConsumer() {
    return consumer.map(consumer -> {
      return executorService.submit(() -> {
        ...
        preExecute();
        try {
          E result = consumer.consume(queue);
          return result;
        } catch (Exception e) {
          queue.markAsFailed(e);
          throw e;
        }
      });
    }).orElse(CompletableFuture.completedFuture(null));
  }

消費時會先進行執行前的準備，然後開始消費，其中consume方法的核心代碼如下

  public O consume(BoundedInMemoryQueue<?, I> queue) throws Exception {
    Iterator<I> iterator = queue.iterator();

    while (iterator.hasNext()) {
      consumeOneRecord(iterator.next());
    }

    // Notifies done
    finish();

    return getResult();
  }

可以看到只要隊列中還有記錄，就可以獲取該記錄，然後調用不同BoundedInMemoryQueueConsumer子類的consumeOneRecord進行更新插入處理。

值得一提的是Hudi對隊列進行了流控，生產者不能無限制地將記錄寫入隊列中，隊列緩存的大小由用戶配置，隊列能放入記錄的條數由採樣的記錄大小和隊列緩存大小控制。

在生產時，會調用BoundedInMemoryQueue#insertRecord將記錄寫入隊列，其核心代碼如下

  public void insertRecord(I t) throws Exception {
    ...
    rateLimiter.acquire();
    // We are retrieving insert value in the record queueing thread to offload computation
    // around schema validation
    // and record creation to it.
    final O payload = transformFunction.apply(t);
    adjustBufferSizeIfNeeded(payload);
    queue.put(Option.of(payload));
  }

首先獲取一個許可(Semaphore)，未成功獲取會被阻塞直至成功獲取，然後獲取記錄的負載以便調整隊列，然後放入內部隊列（LinkedBlockingQueue）中，其中adjustBufferSizeIfNeeded方法的核心代碼如下

  private void adjustBufferSizeIfNeeded(final O payload) throws InterruptedException {
    if (this.samplingRecordCounter.incrementAndGet() % RECORD_SAMPLING_RATE != 0) {
      return;
    }

    final long recordSizeInBytes = payloadSizeEstimator.sizeEstimate(payload);
    final long newAvgRecordSizeInBytes =
        Math.max(1, (avgRecordSizeInBytes * numSamples + recordSizeInBytes) / (numSamples + 1));
    final int newRateLimit =
        (int) Math.min(RECORD_CACHING_LIMIT, Math.max(1, this.memoryLimit / newAvgRecordSizeInBytes));

    // If there is any change in number of records to cache then we will either release (if it increased) or acquire
    // (if it decreased) to adjust rate limiting to newly computed value.
    if (newRateLimit > currentRateLimit) {
      rateLimiter.release(newRateLimit - currentRateLimit);
    } else if (newRateLimit < currentRateLimit) {
      rateLimiter.acquire(currentRateLimit - newRateLimit);
    }
    currentRateLimit = newRateLimit;
    avgRecordSizeInBytes = newAvgRecordSizeInBytes;
    numSamples++;
  }

首先看是否已經達到採樣頻率，然後計算新的記錄平均大小和限流速率，如果新的限流速率大於當前速率，則可釋放一些許可（供阻塞的生產者獲取後繼續生產），否則需要獲取（回收）一些許可（許可變少後生產速率自然就降低了）。該操作可根據採樣的記錄大小動態調節速率，不至於在記錄負載太大和記錄負載太小時，放入同等個數，從而起到動態調節作用。

在消費時，會調用BoundedInMemoryQueue#readNextRecord讀取記錄，其核心代碼如下

  private Option<O> readNextRecord() {
    ...
    rateLimiter.release();
    Option<O> newRecord = Option.empty();
    while (expectMoreRecords()) {
      try {
        throwExceptionIfFailed();
        newRecord = queue.poll(RECORD_POLL_INTERVAL_SEC, TimeUnit.SECONDS);
        if (newRecord != null) {
          break;
        }
      } catch (InterruptedException e) {
        throw new HoodieException(e);
      }
    }
    ...

    if (newRecord != null && newRecord.isPresent()) {
      return newRecord;
    } else {
      // We are done reading all the records from internal iterator.
      this.isReadDone.set(true);
      return Option.empty();
    }
  }

可以看到首先會釋放一個許可，然後判斷是否還可以讀取記錄（還在生產或者停止生產但隊列不為空都可讀取），然後從內部隊列獲取記錄或返回。

上述便是生產者-消費者在Hudi中應用的分析。

總結

Hudi採用了生產者-消費者模型來控制記錄的處理，與傳統多生產者-多消費者模型不同的是，Hudi現在只支持多生產者-單消費者模型，單消費者意味着Hudi暫時不支持文件的併發寫入。而對於生產消費的隊列的實現，Hudi並未僅僅只是基於LinkedBlockingQueue，而是採用了更精細化的速率控制，保證速率會隨着記錄負載大小的變化和配置的隊列緩存大小而動態變化，這也降低了系統發生OOM的概率。

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

※收購3c,收購IPHONE,收購蘋果電腦-詳細收購流程一覽表

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

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

※公開收購3c價格,不怕被賤賣!

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

準備環境

Centos7  192.168.50.21 k8s-master 2G
Centos7  192.168.50.22 k8s-node01 2G
Centos7  192.168.50.23 k8s-node02 2G

K8S集群搭建參考 

master安裝好Git ,yum install git

master,node01,node02  安裝 nfs-utils

yum install nfs-utils

master,node01,node02添加nfs exports配置，為了解決後續的nfs報錯異常

/data/mysql-slave *(insecure,fsid=0,rw,async,no_root_squash)
/data/mysql-master *(insecure,fsid=0,rw,async,no_root_squash)
/data/nfs-share *(rw,fsid=0,sync,no_root_squash)

mysql-slave 數據庫從庫 

mysql-master 數據庫主庫

nfs-share nocas文件掛在目錄

後面的yml中會提到
master,node01,node02創建目錄

mkdir /data/mysql-slave
mkdir /data/mysql-master
mkdir /data/nfs-share 

 master 克隆代碼

   git clone https://github.com/nacos-group/nacos-k8s.git

克隆完成進入以下目錄

 cd /opt/nacos-k8s/deploy/

1.nfs安裝

kubectl create -f nfs/rbac.yaml 

kubectl create -f nfs/class.yaml 

修改nfs/deployment.yaml IP配置

kubectl create -f nfs/deployment.yaml

查看安裝狀態

kubectl get pod -l app=nfs-client-provisioner

2.mysql部署

cd /opt/nacos-k8s/deploy/mysql/

修改數據配置文件ip

vi mysql-master-nfs.yaml

 部署主庫

kubectl create -f mysql-master-nfs.yaml 

修改存庫ip

vi mysql-slave-nfs.yaml

kubectl create -f mysql-slave-nfs.yaml 

主從部署非常慢 耐心等待，如果報nfs相關的錯，重啟nfs即可

service nfs restart

3. 部署nacos

cd /opt/nacos-k8s/deploy/nacos/

kubectl create -f nacos-pvc-nfs.yaml 

 查看訪問端口

kubectl get svc|grep nacos

 查看K8S集群狀態

 Failed to pull image “nacos/nacos-server:latest”: rpc error: code = Unknown desc = context canceled

進去對應節點機器 ，拉取鏡像后，重新應用即可

kubectl apply -f

 4. 部署問題

部署過程中大部分都是NFS問題

可以參考

mount.nfs: No route to host
Warning FailedMount 100s (x5 over 10m) kubelet, node2 Unable to mount volumes for pod “nfs-client-provisioner-594f778474-whhb5_default(56aef93a-9d31-11e9-a4c4-00163e069f44)”: timeout expired waiting for volumes to attach or mount for pod “default”/”nfs-client-provisioner-594f778474-whhb5”. list of unmounted volumes=[nfs-client-root]. list of unattached volumes=[nfs-client-root nfs-client-provisioner-token-8dcrx]

修改deployment.yaml中server的IP地址為某個node節點的內網IP地址，圖1已標註

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

※收購3c,收購IPHONE,收購蘋果電腦-詳細收購流程一覽表

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

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

※公開收購3c價格,不怕被賤賣!

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

之前我們介紹過了 Kafka 整體架構，Kafka 生產者，Kafka 生產的消息最終流向哪裡呢？當然是需要消費了，要不只產生一系列數據沒有任何作用啊，如果把 Kafka 比作餐廳的話，那麼生產者就是廚師的角色，消費者就是客人，只有廚師的話，那麼炒出來的菜沒有人吃也沒有意義，如果只有客人沒有廚師的話，誰會去這個店吃飯呢？！所以如果你看完前面的文章意猶未盡的話，可以繼續讓你爽一爽。如果你沒看過前面的文章，那就從現在開始讓你爽。

Kafka 消費者概念

應用程序使用 KafkaConsumer 從 Kafka 中訂閱主題並接收來自這些主題的消息，然後再把他們保存起來。應用程序首先需要創建一個 KafkaConsumer 對象，訂閱主題並開始接受消息，驗證消息並保存結果。一段時間后，生產者往主題寫入的速度超過了應用程序驗證數據的速度，這時候該如何處理？如果只使用單個消費者的話，應用程序會跟不上消息生成的速度，就像多個生產者像相同的主題寫入消息一樣，這時候就需要多個消費者共同參与消費主題中的消息，對消息進行分流處理。

Kafka 消費者從屬於消費者群組。一個群組中的消費者訂閱的都是相同的主題，每個消費者接收主題一部分分區的消息。下面是一個 Kafka 分區消費示意圖

上圖中的主題 T1 有四個分區，分別是分區0、分區1、分區2、分區3，我們創建一個消費者群組1，消費者群組中只有一個消費者，它訂閱主題T1，接收到 T1 中的全部消息。由於一個消費者處理四個生產者發送到分區的消息，壓力有些大，需要幫手來幫忙分擔任務，於是就演變為下圖

這樣一來，消費者的消費能力就大大提高了，但是在某些環境下比如用戶產生消息特別多的時候，生產者產生的消息仍舊讓消費者吃不消，那就繼續增加消費者。

如上圖所示，每個分區所產生的消息能夠被每個消費者群組中的消費者消費，如果向消費者群組中增加更多的消費者，那麼多餘的消費者將會閑置，如下圖所示

向群組中增加消費者是橫向伸縮消費能力的主要方式。總而言之，我們可以通過增加消費組的消費者來進行水平擴展提升消費能力。這也是為什麼建議創建主題時使用比較多的分區數，這樣可以在消費負載高的情況下增加消費者來提升性能。另外，消費者的數量不應該比分區數多，因為多出來的消費者是空閑的，沒有任何幫助。

Kafka 一個很重要的特性就是，只需寫入一次消息，可以支持任意多的應用讀取這個消息。換句話說，每個應用都可以讀到全量的消息。為了使得每個應用都能讀到全量消息，應用需要有不同的消費組。對於上面的例子，假如我們新增了一個新的消費組 G2，而這個消費組有兩個消費者，那麼就演變為下圖這樣

在這個場景中，消費組 G1 和消費組 G2 都能收到 T1 主題的全量消息，在邏輯意義上來說它們屬於不同的應用。

總結起來就是如果應用需要讀取全量消息，那麼請為該應用設置一個消費組；如果該應用消費能力不足，那麼可以考慮在這個消費組裡增加消費者。

消費者組和分區重平衡

消費者組是什麼

消費者組（Consumer Group）是由一個或多個消費者實例（Consumer Instance）組成的群組，具有可擴展性和可容錯性的一種機制。消費者組內的消費者共享一個消費者組ID，這個ID 也叫做 Group ID，組內的消費者共同對一個主題進行訂閱和消費，同一個組中的消費者只能消費一個分區的消息，多餘的消費者會閑置，派不上用場。

我們在上面提到了兩種消費方式

• 一個消費者群組消費一個主題中的消息，這種消費模式又稱為點對點的消費方式，點對點的消費方式又被稱為消息隊列
• 一個主題中的消息被多個消費者群組共同消費，這種消費模式又稱為發布-訂閱模式

消費者重平衡

我們從上面的消費者演變圖中可以知道這麼一個過程：最初是一個消費者訂閱一個主題並消費其全部分區的消息，後來有一個消費者加入群組，隨後又有更多的消費者加入群組，而新加入的消費者實例分攤了最初消費者的部分消息，這種把分區的所有權通過一個消費者轉到其他消費者的行為稱為重平衡，英文名也叫做 Rebalance 。如下圖所示

重平衡非常重要，它為消費者群組帶來了高可用性 和 伸縮性，我們可以放心的添加消費者或移除消費者，不過在正常情況下我們並不希望發生這樣的行為。在重平衡期間，消費者無法讀取消息，造成整個消費者組在重平衡的期間都不可用。另外，當分區被重新分配給另一個消費者時，消息當前的讀取狀態會丟失，它有可能還需要去刷新緩存，在它重新恢復狀態之前會拖慢應用程序。

消費者通過向組織協調者（Kafka Broker）發送心跳來維護自己是消費者組的一員並確認其擁有的分區。對於不同不的消費群體來說，其組織協調者可以是不同的。只要消費者定期發送心跳，就會認為消費者是存活的並處理其分區中的消息。當消費者檢索記錄或者提交它所消費的記錄時就會發送心跳。

如果過了一段時間 Kafka 停止發送心跳了，會話（Session）就會過期，組織協調者就會認為這個 Consumer 已經死亡，就會觸發一次重平衡。如果消費者宕機並且停止發送消息，組織協調者會等待幾秒鐘，確認它死亡了才會觸發重平衡。在這段時間里，死亡的消費者將不處理任何消息。在清理消費者時，消費者將通知協調者它要離開群組，組織協調者會觸發一次重平衡，盡量降低處理停頓。

重平衡是一把雙刃劍，它為消費者群組帶來高可用性和伸縮性的同時，還有有一些明顯的缺點(bug)，而這些 bug 到現在社區還無法修改。

重平衡的過程對消費者組有極大的影響。因為每次重平衡過程中都會導致萬物靜止，參考 JVM 中的垃圾回收機制，也就是 Stop The World ，STW，(引用自《深入理解 Java 虛擬機》中 p76 關於 Serial 收集器的描述)：

更重要的是它在進行垃圾收集時，必須暫停其他所有的工作線程。直到它收集結束。Stop The World 這個名字聽起來很帥，但這項工作實際上是由虛擬機在後台自動發起並完成的，在用戶不可見的情況下把用戶正常工作的線程全部停掉，這對很多應用來說都是難以接受的。

也就是說，在重平衡期間，消費者組中的消費者實例都會停止消費，等待重平衡的完成。而且重平衡這個過程很慢……

創建消費者

上面的理論說的有點多，下面就通過代碼來講解一下消費者是如何消費的

在讀取消息之前，需要先創建一個 KafkaConsumer 對象。創建 KafkaConsumer 對象與創建 KafkaProducer 對象十分相似 — 把需要傳遞給消費者的屬性放在 properties 對象中，後面我們會着重討論 Kafka 的一些配置，這裏我們先簡單的創建一下，使用3個屬性就足矣，分別是 bootstrap.server，key.deserializer，value.deserializer 。

這三個屬性我們已經用過很多次了，如果你還不是很清楚的話，可以參考

還有一個屬性是 group.id 這個屬性不是必須的，它指定了 KafkaConsumer 是屬於哪個消費者群組。創建不屬於任何一個群組的消費者也是可以的

Properties properties = new Properties();
        properties.put("bootstrap.server","192.168.1.9:9092");     properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");   properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
KafkaConsumer<String,String> consumer = new KafkaConsumer<>(properties);

主題訂閱

創建好消費者之後，下一步就開始訂閱主題了。subscribe() 方法接受一個主題列表作為參數，使用起來比較簡單

consumer.subscribe(Collections.singletonList("customerTopic"));

為了簡單我們只訂閱了一個主題 customerTopic，參數傳入的是一個正則表達式，正則表達式可以匹配多個主題，如果有人創建了新的主題，並且主題的名字與正則表達式相匹配，那麼會立即觸發一次重平衡，消費者就可以讀取新的主題。

要訂閱所有與 test 相關的主題，可以這樣做

consumer.subscribe("test.*");

輪詢

我們知道，Kafka 是支持訂閱/發布模式的，生產者發送數據給 Kafka Broker，那麼消費者是如何知道生產者發送了數據呢？其實生產者產生的數據消費者是不知道的，KafkaConsumer 採用輪詢的方式定期去 Kafka Broker 中進行數據的檢索，如果有數據就用來消費，如果沒有就再繼續輪詢等待，下面是輪詢等待的具體實現

try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(100));
    for (ConsumerRecord<String, String> record : records) {
      int updateCount = 1;
      if (map.containsKey(record.value())) {
        updateCount = (int) map.get(record.value() + 1);
      }
      map.put(record.value(), updateCount);
    }
  }
}finally {
  consumer.close();
}

• 這是一個無限循環。消費者實際上是一個長期運行的應用程序，它通過輪詢的方式向 Kafka 請求數據。
• 第三行代碼非常重要，Kafka 必須定期循環請求數據，否則就會認為該 Consumer 已經掛了，會觸發重平衡，它的分區會移交給群組中的其它消費者。傳給 poll() 方法的是一個超市時間，用 java.time.Duration 類來表示，如果該參數被設置為 0 ，poll() 方法會立刻返回，否則就會在指定的毫秒數內一直等待 broker 返回數據。
• poll() 方法會返回一個記錄列表。每條記錄都包含了記錄所屬主題的信息，記錄所在分區的信息、記錄在分區中的偏移量，以及記錄的鍵值對。我們一般會遍歷這個列表，逐條處理每條記錄。
• 在退出應用程序之前使用 close() 方法關閉消費者。網絡連接和 socket 也會隨之關閉，並立即觸發一次重平衡，而不是等待群組協調器發現它不再發送心跳並認定它已經死亡。

線程安全性

在同一個群組中，我們無法讓一個線程運行多個消費者，也無法讓多個線程安全的共享一個消費者。按照規則，一個消費者使用一個線程，如果一個消費者群組中多個消費者都想要運行的話，那麼必須讓每個消費者在自己的線程中運行，可以使用 Java 中的 ExecutorService 啟動多個消費者進行進行處理。

消費者配置

到目前為止，我們學習了如何使用消費者 API，不過只介紹了幾個最基本的屬性，Kafka 文檔列出了所有與消費者相關的配置說明。大部分參數都有合理的默認值，一般不需要修改它們，下面我們就來介紹一下這些參數。

• fetch.min.bytes

該屬性指定了消費者從服務器獲取記錄的最小字節數。broker 在收到消費者的數據請求時，如果可用的數據量小於 fetch.min.bytes 指定的大小，那麼它會等到有足夠的可用數據時才把它返回給消費者。這樣可以降低消費者和 broker 的工作負載，因為它們在主題使用頻率不是很高的時候就不用來回處理消息。如果沒有很多可用數據，但消費者的 CPU 使用率很高，那麼就需要把該屬性的值設得比默認值大。如果消費者的數量比較多，把該屬性的值調大可以降低 broker 的工作負載。

• fetch.max.wait.ms

我們通過上面的 fetch.min.bytes 告訴 Kafka，等到有足夠的數據時才會把它返回給消費者。而 fetch.max.wait.ms 則用於指定 broker 的等待時間，默認是 500 毫秒。如果沒有足夠的數據流入 kafka 的話，消費者獲取的最小數據量要求就得不到滿足，最終導致 500 毫秒的延遲。如果要降低潛在的延遲，就可以把參數值設置的小一些。如果 fetch.max.wait.ms 被設置為 100 毫秒的延遲，而 fetch.min.bytes 的值設置為 1MB，那麼 Kafka 在收到消費者請求后，要麼返回 1MB 的數據，要麼在 100 ms 后返回所有可用的數據。就看哪個條件首先被滿足。

• max.partition.fetch.bytes

該屬性指定了服務器從每個分區里返回給消費者的最大字節數。它的默認值時 1MB，也就是說，KafkaConsumer.poll() 方法從每個分區里返回的記錄最多不超過 max.partition.fetch.bytes 指定的字節。如果一個主題有20個分區和5個消費者，那麼每個消費者需要至少4 MB的可用內存來接收記錄。在為消費者分配內存時，可以給它們多分配一些，因為如果群組裡有消費者發生崩潰，剩下的消費者需要處理更多的分區。max.partition.fetch.bytes 的值必須比 broker 能夠接收的最大消息的字節數(通過 max.message.size 屬性配置大)，否則消費者可能無法讀取這些消息，導致消費者一直掛起重試。 在設置該屬性時，另外一個考量的因素是消費者處理數據的時間。消費者需要頻繁的調用 poll() 方法來避免會話過期和發生分區再平衡，如果單次調用poll() 返回的數據太多，消費者需要更多的時間進行處理，可能無法及時進行下一個輪詢來避免會話過期。如果出現這種情況，可以把 max.partition.fetch.bytes 值改小，或者延長會話過期時間。

• session.timeout.ms

這個屬性指定了消費者在被認為死亡之前可以與服務器斷開連接的時間，默認是 3s。如果消費者沒有在 session.timeout.ms 指定的時間內發送心跳給群組協調器，就會被認定為死亡，協調器就會觸發重平衡。把它的分區分配給消費者群組中的其它消費者，此屬性與 heartbeat.interval.ms 緊密相關。heartbeat.interval.ms 指定了 poll() 方法向群組協調器發送心跳的頻率，session.timeout.ms 則指定了消費者可以多久不發送心跳。所以，這兩個屬性一般需要同時修改，heartbeat.interval.ms 必須比 session.timeout.ms 小，一般是 session.timeout.ms 的三分之一。如果 session.timeout.ms 是 3s，那麼 heartbeat.interval.ms 應該是 1s。把 session.timeout.ms 值設置的比默認值小，可以更快地檢測和恢復崩憤的節點，不過長時間的輪詢或垃圾收集可能導致非預期的重平衡。把該屬性的值設置得大一些，可以減少意外的重平衡，不過檢測節點崩潰需要更長的時間。

• auto.offset.reset

該屬性指定了消費者在讀取一個沒有偏移量的分區或者偏移量無效的情況下的該如何處理。它的默認值是 latest，意思指的是，在偏移量無效的情況下，消費者將從最新的記錄開始讀取數據。另一個值是 earliest，意思指的是在偏移量無效的情況下，消費者將從起始位置處開始讀取分區的記錄。

• enable.auto.commit

我們稍後將介紹幾種不同的提交偏移量的方式。該屬性指定了消費者是否自動提交偏移量，默認值是 true，為了盡量避免出現重複數據和數據丟失，可以把它設置為 false，由自己控制何時提交偏移量。如果把它設置為 true，還可以通過 auto.commit.interval.ms 屬性來控制提交的頻率

• partition.assignment.strategy

我們知道，分區會分配給群組中的消費者。PartitionAssignor 會根據給定的消費者和主題，決定哪些分區應該被分配給哪個消費者，Kafka 有兩個默認的分配策略Range 和 RoundRobin

• client.id

該屬性可以是任意字符串，broker 用他來標識從客戶端發送過來的消息，通常被用在日誌、度量指標和配額中

• max.poll.records

該屬性用於控制單次調用 call() 方法能夠返回的記錄數量，可以幫你控制在輪詢中需要處理的數據量。

• receive.buffer.bytes 和 send.buffer.bytes

socket 在讀寫數據時用到的 TCP 緩衝區也可以設置大小。如果它們被設置為 -1，就使用操作系統默認值。如果生產者或消費者與 broker 處於不同的數據中心內，可以適當增大這些值，因為跨數據中心的網絡一般都有比較高的延遲和比較低的帶寬。

提交和偏移量的概念

特殊偏移

我們上面提到，消費者在每次調用poll() 方法進行定時輪詢的時候，會返回由生產者寫入 Kafka 但是還沒有被消費者消費的記錄，因此我們可以追蹤到哪些記錄是被群組裡的哪個消費者讀取的。消費者可以使用 Kafka 來追蹤消息在分區中的位置（偏移量）

消費者會向一個叫做 _consumer_offset 的特殊主題中發送消息，這個主題會保存每次所發送消息中的分區偏移量，這個主題的主要作用就是消費者觸發重平衡後記錄偏移使用的，消費者每次向這個主題發送消息，正常情況下不觸發重平衡，這個主題是不起作用的，當觸發重平衡后，消費者停止工作，每個消費者可能會分到對應的分區，這個主題就是讓消費者能夠繼續處理消息所設置的。

如果提交的偏移量小於客戶端最後一次處理的偏移量，那麼位於兩個偏移量之間的消息就會被重複處理

如果提交的偏移量大於最後一次消費時的偏移量，那麼處於兩個偏移量中間的消息將會丟失

既然_consumer_offset 如此重要，那麼它的提交方式是怎樣的呢？下面我們就來說一下

提交方式

KafkaConsumer API 提供了多種方式來提交偏移量

自動提交

最簡單的方式就是讓消費者自動提交偏移量。如果 enable.auto.commit 被設置為true，那麼每過 5s，消費者會自動把從 poll() 方法輪詢到的最大偏移量提交上去。提交時間間隔由 auto.commit.interval.ms 控制，默認是 5s。與消費者里的其他東西一樣，自動提交也是在輪詢中進行的。消費者在每次輪詢中會檢查是否提交該偏移量了，如果是，那麼就會提交從上一次輪詢中返回的偏移量。

提交當前偏移量

把 auto.commit.offset 設置為 false，可以讓應用程序決定何時提交偏移量。使用 commitSync() 提交偏移量。這個 API 會提交由 poll() 方法返回的最新偏移量，提交成功后馬上返回，如果提交失敗就拋出異常。

commitSync() 將會提交由 poll() 返回的最新偏移量，如果處理完所有記錄后要確保調用了 commitSync()，否則還是會有丟失消息的風險，如果發生了在均衡，從最近一批消息到發生在均衡之間的所有消息都將被重複處理。

異步提交

異步提交 commitAsync() 與同步提交 commitSync() 最大的區別在於異步提交不會進行重試，同步提交會一致進行重試。

同步和異步組合提交

一般情況下，針對偶爾出現的提交失敗，不進行重試不會有太大的問題，因為如果提交失敗是因為臨時問題導致的，那麼後續的提交總會有成功的。但是如果在關閉消費者或再均衡前的最後一次提交，就要確保提交成功。

因此，在消費者關閉之前一般會組合使用commitAsync和commitSync提交偏移量。

提交特定的偏移量

消費者API允許調用 commitSync() 和 commitAsync() 方法時傳入希望提交的 partition 和 offset 的 map，即提交特定的偏移量。

文章參考：

《極客時間-Kafka核心技術與實戰》

《Kafka 權威指南》

關注公眾號獲取更多優質电子書，關注一下你就知道資源是有多好了

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

※收購3c,收購IPHONE,收購蘋果電腦-詳細收購流程一覽表

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

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

※公開收購3c價格,不怕被賤賣!

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