abp(net core)+easyui+efcore實現倉儲管理系統——ABP WebAPI與EasyUI結合增刪改查之二(二十八)




 










   

       在上一篇 文章中我們學習了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價格,不怕被賤賣!

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

中油換電站今年採競標,Gogoro累計 285 站、光陽 146 站

中油從 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價格,不怕被賤賣!

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

遷移桌面程序到MS Store(12)——WPF使用UWP InkToolbar和InkCanvas

我們在提到了對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中的應用

介紹

生產者-消費者模型用於解耦生產者與消費者,平衡兩者之間的能力不平衡,該模型廣泛應用於各個系統中,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價格,不怕被賤賣!

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

Alibaba Nacos 學習(五):K8S Nacos搭建,使用nfs

 

準備環境

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價格,不怕被賤賣!

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

PHP 的 self 關鍵字用法

之前有人詢問 self 關鍵字的用法,答案是比較明顯的:靜態成員函數內不能用 this 調用非成員函數,但可以用 self 調用靜態成員函數/變量/常量;其他成員函數可以用 self 調用靜態成員函數以及非靜態成員函數。隨着討論的深入,發現 self 並沒有那麼簡單。鑒於此,本文先對幾個關鍵字做對比和區分,再總結 self 的用法。

parentstatic 以及 this 的區別

要想將徹底搞懂 self ,要與 parentstatic 以及 this 區分開。以下分別做對比。

parent

selfparent 的區分比較容易: parent 引用父類/基類被隱蓋的方法(或變量), self則引用自身方法(或變量)。例如構造函數中調用父類構造函數:

class Base {
    public function __construct() {
        echo "Base contructor!", PHP_EOL;
    }
}

class Child {
    public function __construct() {
        parent::__construct();
        echo "Child contructor!", PHP_EOL;
    }
}

new Child;
// 輸出:
// Base contructor!
// Child contructor!

 

static

static 常規用途是修飾函數或變量使其成為類函數和類變量,也可以修飾函數內變量延長其生命周期至整個應用程序的生命周期。但是其與 self 關聯上是PHP 5.3以來引入的新用途:靜態延遲綁定。

有了 static 的靜態延遲綁定功能,可以在運行時動態確定歸屬的類。例如:

class Base {
    public function __construct() {
        echo "Base constructor!", PHP_EOL;
    }

    public static function getSelf() {
        return new self();
    }

    public static function getInstance() {
        return new static();
    }

    public function selfFoo() {
        return self::foo();
    }

    public function staticFoo() {
        return static::foo();
    }

    public function thisFoo() {
        return $this->foo();
    }

    public function foo() {
        echo  "Base Foo!", PHP_EOL;
    }
}

class Child extends Base {
    public function __construct() {
        echo "Child constructor!", PHP_EOL;
    }

    public function foo() {
        echo "Child Foo!", PHP_EOL;
    }
}

$base = Child::getSelf();
$child = Child::getInstance();

$child->selfFoo();
$child->staticFoo();
$child->thisFoo();

 

程序輸出結果如下:

Base constructor!
Child constructor!
Base Foo!
Child Foo!
Child Foo!

 

在函數引用上, selfstatic 的區別是:對於靜態成員函數, self 指向代碼當前類, static 指向調用類;對於非靜態成員函數, self 抑制多態,指向當前類的成員函數, static 等同於 this ,動態指向調用類的函數。

parentselfstatic 三個關鍵字聯合在一起看挺有意思,分別指向父類、當前類、子類,有點“過去、現在、未來”的味道。

this

selfthis 是被討論最多,也是最容易引起誤用的組合。兩者的主要區別如下:

  1. this 不能用在靜態成員函數中, self 可以;
  2. 對靜態成員函數/變量的訪問, 建議 用 self ,不要用 $this::$this-> 的形式;
  3. 對非靜態成員變量的訪問,不能用 self ,只能用 this ;
  4. this 要在對象已經實例化的情況下使用, self 沒有此限制;
  5. 在非靜態成員函數內使用, self 抑制多態行為,引用當前類的函數;而 this 引用調用類的重寫(override)函數(如果有的話)。

self 的用途

看完與上述三個關鍵字的區別, self 的用途是不是呼之即出?一句話總結,那就是: self總是指向“當前類(及類實例)”。詳細說則是:

  1. 替代類名,引用當前類的靜態成員變量和靜態函數;
  2. 抑制多態行為,引用當前類的函數而非子類中覆蓋的實現;

槽點

  1. 這幾個關鍵字中,只有 this 要加 $ 符號且必須加,強迫症表示很難受;
  2. 靜態成員函數中不能通過 $this-> 調用非靜態成員函數,但是可以通過 self:: 調用,且在調用函數中未使用 $this-> 的情況下還能順暢運行。此行為貌似在不同PHP版本中表現不同,在當前的7.3中ok;
  3. 在靜態函數和非靜態函數中輸出 self ,猜猜結果是什麼?都是 string(4) "self" ,迷之輸出;
  4. return $this instanceof static::class; 會有語法錯誤,但是以下兩種寫法就正常:
    $class = static::class;
    return $this instanceof $class;
    // 或者這樣:
    return $this instanceof static;

 

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

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

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

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

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

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

帶你漲姿勢的認識一下 Kafka 消費者

之前我們介紹過了 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.serverkey.deserializervalue.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 有兩個默認的分配策略RangeRoundRobin

  • 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價格,不怕被賤賣!

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

特斯拉發表電動卡車 Cybertruck,一台從科幻電影走出來的鋼鐵車

今天的 Elon Musk 看起來不像鋼鐵人,更像蝙蝠俠,因為他們的新車 Cybertruck,不但外型酷似蝙蝠車,同時還能防彈防撞,並且擁有超越保時捷的加速度,跟勝過市面上卡車的拖吊能力,更驚人的是,售價只要 39,900 美元起。

眾所期待的特斯拉新車 Cybertruck 今日正式發表,和之前流出的影像不同,Cybertruck 酷似隱形戰機 F-117 的設計,讓人聯想到蝙蝠車,甚至懷疑這是不是一台防雷達偵測的戰車?

Tesla Cybertruck 全車採用冷鑄鋼板,能夠抵擋 9mm 口徑手槍的射擊,現場展示用重鎚敲擊也毫髮無傷;車窗玻璃同樣採用防彈設計,然而有趣的是,現場展示時,被大鐵球砸出了一片雪花。「至少,它沒被打穿,你坐在裡面很安全。」Elon Musk 笑著說。

Cybertruck 為了因應負重,搭載了適應性氣壓懸吊系統,針對高速公路,或是越野泥巴路,能夠自動調整懸吊高度,同時也順便使用這個氣壓系統,做了一個高壓出力裝置,使用者可以自行加裝不同氣壓工具,像是高壓水槍或是電鑽等。

當重裝電動機車開上後廂時,懸吊系統會自動調整車尾高度,讓車身保持平衡。

車尾與其他皮卡車開放式貨斗不同,Cybertruck 採用封閉式貨斗,並有升降式尾門,現場展示時,將這台電動機車 ATV 直接騎上貨斗後,還能直接充電,顯然是在致敬蝙蝠車跟蝙蝠機車。

Cybertruck 如同其他皮卡車,車尾裝有釣鉤,能夠充當拖車使用,而歸功於它的強力馬達,拖車能力屌打了皮卡車霸主 Ford F-150,在現場展示的影片中,特斯拉讓 Cybertruck 跟 F-150 互相拖住對方,進行拔河測試,結果 F-150 整台被 Cybertruck 拖走。

F-150 慘遭 Cybertruck 拖走。

馬斯克強調,一般皮卡車需要另外裝載發電機才能使用電動工具,Cybertruck 直接提供了電源,因此省下不少空間,同時還提供強大的拖力。

此外,做為一台卡車,Cybertruck 莫名其妙地擁有超越保時捷的加速度,根據現場公布數據,最頂級版的 0-100 公里加速時間不到 3 秒。現場展示了 Cybertruck 與 Porsche 911 賽跑的影片,起步雖然小輸一點,但隨後就超越了 911。

現場展示競速影片,大約 1 秒後,Cybertruck 就超過了 911。

Tesla Cybertruck 共有 3 種版本,依照馬達數量來分別,最低價 39,900 美元起,最高 69,900 美元。Cybertruck 從今天起在美國開放預購,實際交車時間預計要等到 2021 年底。頂級的三馬達款,更預計要等到 2022 年底才會開始生產。

如同馬斯克開場所說,卡車在過去幾十年來都長得差不多,特斯拉要打造一台完全不一樣的卡車,同時還要保持零排放,跟超高性能,從今天的現場展示來看,特斯拉再次完成一個不可能的任務。在興奮之餘也別忘了,這一切都是現場展示,實際上如何,就有待實際交車後驗證了!

(合作媒體:。圖片來源:)

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

【其他文章推薦】

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

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

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

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

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

【其他文章推薦】

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

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

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

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

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

[apue] 神奇的 Solaris pipe

說到 pipe 大家可能都不陌生,經典的pipe調用配合fork進行父子進程通訊,簡直就是Unix程序的標配。

然而Solaris上的pipe卻和Solaris一樣是個奇葩(雖然Solaris前途黯淡,但是不妨礙我們從它裏面挖掘一些有價值的東西),

有着和一般pipe諸多的不同之處,本文就來說說Solaris上神奇的pipe和一般pipe之間的異同。

 

1.solaris pipe 是全雙工的

一般系統上的pipe調用是半雙工的,只能單向傳遞數據,如果需要雙向通訊,我們一般是建兩個pipe分別讀寫。像下面這樣:

 1     int n, fd1[2], fd2[2]; 
 2     if (pipe (fd1) < 0 || pipe(fd2) < 0)
 3         err_sys ("pipe error"); 
 4 
 5     char line[MAXLINE]; 
 6     pid_t pid = fork (); 
 7     if (pid < 0) 
 8         err_sys ("fork error"); 
 9     else if (pid > 0)
10     {
11         close (fd1[0]);  // write on pipe1 as stdin for co-process
12         close (fd2[1]);  // read on pipe2 as stdout for co-process
13         while (fgets (line, MAXLINE, stdin) != NULL) { 
14             n = strlen (line); 
15             if (write (fd1[1], line, n) != n)
16                 err_sys ("write error to pipe"); 
17             if ((n = read (fd2[0], line, MAXLINE)) < 0)
18                 err_sys ("read error from pipe"); 
19 
20             if (n == 0) { 
21                 err_msg ("child closed pipe"); 
22                 break;
23             }
24             line[n] = 0; 
25             if (fputs (line, stdout) == EOF)
26                 err_sys ("fputs error"); 
27         }
28 
29         if (ferror (stdin))
30             err_sys ("fputs error"); 
31 
32         return 0; 
33     }
34     else { 
35         close (fd1[1]); 
36         close (fd2[0]); 
37         if (fd1[0] != STDIN_FILENO) { 
38             if (dup2 (fd1[0], STDIN_FILENO) != STDIN_FILENO)
39                 err_sys ("dup2 error to stdin"); 
40             close (fd1[0]); 
41         }
42 
43         if (fd2[1] != STDOUT_FILENO) { 
44             if (dup2 (fd2[1], STDOUT_FILENO) != STDOUT_FILENO)
45                 err_sys ("dup2 error to stdout"); 
46             close (fd2[1]); 
47         }
48 
49         if (execl (argv[1], "add2", (char *)0) < 0)
50             err_sys ("execl error"); 
51     }

這個程序創建兩個管道,fd1用來寫請求,fd2用來讀應答;對子進程而言,fd1重定向到標準輸入,fd2重定向到標準輸出,讀取stdin中的數據相加然後寫入stdout完成工作。父進程在取得應答後向標準輸出寫入結果。

如果在Solaris上,可以直接用一個pipe同時讀寫,代碼可以重寫成這樣:

 1 int fd[2];
 2 if (pipe(fd) < 0) 
 3     err_sys("pipe error\n");
 4 
 5 char line[MAXLINE];
 6 pid_t pid = fork();
 7 if (pid < 0)
 8     err_sys("fork error\n");
 9 else if (pid > 0)
10 {
11     close(fd[1]);
12     while (fgets(line, MAXLINE, stdin) != NULL) {
13         n = strlen(line);
14         if (write(fd[0], line, n) != n)
15             err_sys("write error to pipe\n")
16         if ((n = read(fd[0], line, MAXLINE)) < 0) 
17             err_sys("read error from pipe\n");
18 
19         if (n == 0) 
20             err_sys("child closed pipe\n");
21         line[n] = 0;
22         if (fputs(line, stdout) == EOF) 
23             err_sys("fputs error\n");
24     }
25 
26     if (ferror(stdin))
27         err_sys("fputs error\n");
28 
29     return 0;
30 }
31 else {
32     close(fd[0]);
33     if (fd[1] != STDIN_FILENO)
34         if (dup2(fd[1], STDIN_FILENO) != STDIN_FILENO)
35             err_sys("dup2 error to stdin\n");
36 
37     if (fd[1] != STDOUT_FILENO) {
38         if (dup2(fd[1], STDOUT_FILENO) != STDOUT_FILENO)
39             err_sys("dup2 error to stdout\n");
40         close(fd[1]);
41     }
42 
43     if (execl(argv[1], argv[2], (char *)0) < 0)
44         err_sys("execl error\n");
45 
46 }

代碼清爽多了,不用去考慮fd1[0]和fd2[1]是啥意思是一件很養腦的事。

不過這樣的代碼只能在Solaris上運行(聽說BSD也支持?),如果考慮到可移植性,還是寫上面的比較穩妥。

 

測試程序

 

 

2. solaris pipe 可以脫離父子關係建立

pipe 好用但是沒法脫離fork使用,一般的pipe如果想讓任意兩個進程通訊,得藉助它的變身fifo來實現。

關於FIFO,詳情可參考我之前寫的一篇文章:

 

而Solaris上的pipe沒這麼多事,加入兩個調用:fattach / fdetach,你就可以像使用FIFO一樣使用pipe了:

 1 int fd[2];
 2 if (pipe(fd) < 0)
 3     err_sys("pipe error\n");
 4 
 5 if (fattach(fd[1], "./pipe") < 0)
 6     err_sys("fattach error\n");
 7 
 8 printf("attach to file pipe ok\n");
 9 
10 close(fd[1]);
11 char line[MAXLINE];
12 while (fgets(line, MAXLINE, stdin) != NULL) {
13     n = strlen(line);
14     if (write(fd[0], line, n) != n)
15         err_sys("write error to pipe\n");
16     if ((n = read(fd[0], line, MAXLINE)) < 0)
17         err_sys("read error from pipe\n");
18 
19     if (n == 0) 
20         err_sys("child closed pipe\n");
21 
22     line[n] = 0;
23     if (fputs(line, stdout) == EOF)
24         err_sys("fputs error\n");
25 }
26 
27 if (ferror(stdin))
28     err_sys("fputs error\n");
29 
30 if (fdetach("./pipe") < 0)
31     err_sys("fdetach error\n");
32 
33 printf("detach from file pipe ok\n");

在pipe調用之後立即加入fattach調用,可以將管道關聯到文件系統的一個文件名上,該文件必需事先存在,且可讀可寫。

在fattach調用之前這個文件(./pipe)是個普通文件,打開讀寫都是磁盤IO;

在fattach調用之後,這個文件就變身成為一個管道了,打開讀寫都是內存流操作,且管道的另一端就是attach的那個進程。

子進程也需要改造一下,以便使用pipe通訊:

 1 int fd, n, int1, int2;
 2 char line[MAXLINE];
 3 fd = open("./pipe", O_RDWR);
 4 if (fd < 0)
 5     err_sys("open file pipe failed\n");
 6 
 7 printf("open file pipe ok, fd = %d\n", fd);
 8 while ((n = read(fd, line, MAXLINE)) > 0) {
 9     line[n] = 0;
10     if (sscanf(line, "%d%d", &int1, &int2) == 2) {
11         sprintf(line, "%d\n", int1 + int2);
12         n = strlen(line);
13         if (write(fd, line, n) != n)
14             err_sys("write error\n");
15 
16         printf("i am working on %s\n", line);
17     }
18     else {
19         if (write(fd, "invalid args\n", 13) != 13)
20             err_sys("write msg error\n");
21     }
22 }
23 
24 close(fd);

打開pipe就如同打開普通文件一樣,open直接搞定。當然前提是attach進程必需已經在運行。

當attach進程detach后,管道文件又將恢復它的本來面目。

 

脫離了父子關係的pipe其實可以建立多對一關係(多對多關係不可以,因為只能有一個進程attach)。

例如開4個cmd窗口,分別執行以下命令:

./padd2 abc
./add2
./add2
./add2

 向attach進程(padd2)發送9個計算請求后,可以看到輸出結果如下:

-bash-3.2$ ./padd2 abc
attach to file pipe ok
1 1
2
2 2
4
3 3 
6
4 4
8
5 5
10
6 6 
12
7 7 
14
8 8
16
9 9
18

 再回來看各個open管道的進程,輸出分別如下:

-bash-3.2$ ./add2
open file pipe ok, fd = 3
source: 1 1
i am working on 2
source: 4 4
i am working on 8
source: 7 7 
i am working on 14 

 

-bash-3.2$ ./add2
open file pipe ok, fd = 3
source: 2 2
i am working on 4
source: 5 5
i am working on 10
source: 9 9
i am working on 18 

 

-bash-3.2$ ./add2
open file pipe ok, fd = 3
source: 2 2
i am working on 4
source: 5 5
i am working on 10
source: 9 9
i am working on 18 

 

-bash-3.2$ ./add2
open file pipe ok, fd = 3
source: 3 3
i am working on 6
source: 6 6
i am working on 12
source: 8 8 
i am working on 16

 

可以發現一個很有趣的現象,就是各個add2進程基本是輪着來獲取請求的,可以猜想底層的pipe可能有一個進程排隊機制。

但是反過來使用pipe就不行了。就是說當啟動一個add3(區別於上例的add2與padd2)作為fattach端打開pipe,啟動多個padd3作為open端使用pipe,

然後通過命令行給padd3傳遞要相加的值,可以寫一個腳本同時啟動多個padd3,來查看效果:

#! /bin/sh
./padd3 1 1 &
./padd3 2 2 &
./padd3 3 3 &
./padd3 4 4 &

 這個腳本中啟動了4個加法進程,同時向add3發送4個加法請求,腳本中四個進程輸出如下:

-bash-3.2$ ./padd3.sh
-bash-3.2$ open file pipe ok, fd = 3
1 1 = 2
open file pipe ok, fd = 3
2 2 = 4
open file pipe ok, fd = 3
open file pipe ok, fd = 3
4 4 = 37

 可以看到3+3的請求被忽略了,轉到add3查看輸出:

-bash-3.2$ ./add3
attach to file pipe ok
source: 1 1
i am working on 1 + 1 = 2
source: 2 2
i am working on 2 + 2 = 4
source: 3 34 4
i am working on 3 + 34 = 37

 原來是3+3與4+4兩個請求粘連了,導致add3識別成一個3+34的請求,所以出錯了。

多運行幾遍腳本后,發現還有這樣的輸出:

-bash-3.2$ ./padd3.sh
-bash-3.2$ open file pipe ok, fd = 3
4 4 = 2
open file pipe ok, fd = 3
2 2 = 4
open file pipe ok, fd = 3
3 3 = 6
open file pipe ok, fd = 3
1 1 = 8

  4+4=2?1+1=8?再看add3這頭的輸出:

-bash-3.2$ ./add3
attach to file pipe ok
source: 1 1
i am working on 1 + 1 = 2
source: 2 2
i am working on 2 + 2 = 4
source: 3 3
i am working on 3 + 3 = 6
source: 4 4
i am working on 4 + 4 = 8

 完全正常呢。

經過一番推理,發現是4+4的請求取得了1+1請求的應答;1+1的請求取得了4+4的應答。

可見這樣的結構還有一個弊端,同時請求的進程可能無法得到自己的應答,應答與請求之間相互錯位。

所以想用fattach來實現多路請求的人還是洗洗睡吧,畢竟它就是一個pipe不是,還能給它整成tcp么?

而之前的例子可以,是因為請求是順序發送的,上個請求得到應答后才發送下個請求,所以不存在這個例子的問題(但是實用性也不高)。

 

測試程序

 

 

3. solaris pipe 可以通過connld模塊實現類似tcp的多路連接

第2條剛說不能實現多路連接,第3條就接着來打臉了,這是由於Solaris上的pipe都是基於STREAMS技術構建,

而STREAMS是支持靈活的PUSH、POP流處理模塊的,再加上STREAMS傳遞文件fd的能力,就可以支持類似tcp中accept的能力。

即每個open pipe文件的進程,得到的不是原來管道的fd,而是新創建管道的fd,而管道的另一側fd則通過已有的管道發送到attach進程,

後者使用這個新的fd與客戶進程通訊。為了支持多路連接,我們的代碼需要重新整理一下,首先看客戶端:

1 int fd;
2 char line[MAXLINE];
3 fd = cli_conn("./pipe");
4 if (fd < 0)
5     return 0;

這裏將open相關邏輯封裝到了cli_conn函數中,以便之後復用:

 1 int cli_conn(const char *name)
 2 {
 3     int fd;
 4     if ((fd = open(name, O_RDWR)) < 0) {
 5         printf("open pipe file failed\n");
 6         return -1;
 7     }
 8 
 9     if (isastream(fd) == 0) {
10         close(fd);
11         return -2;
12     }
13 
14     return fd;
15 }

可以看到與之前幾乎沒有變化,只是增加了isastream調用防止attach進程沒有啟動。

再來看下服務端:

 1 int listenfd = serv_listen("./pipe");
 2 if (listenfd < 0)
 3     return 0;
 4 
 5 int acceptfd = 0;
 6 int n = 0, int1 = 0, int2 = 0;
 7 char line[MAXLINE];
 8 uid_t uid = 0;
 9 while ((acceptfd = serv_accept(listenfd, &uid)) >= 0)
10 {
11     printf("accept a client, fd = %d, uid = %ld\n", acceptfd, uid);
12     while ((n = read(acceptfd, line, MAXLINE)) > 0) {
13         line[n] = 0;
14         printf("source: %s\n", line);
15         if (sscanf(line, "%d%d", &int1, &int2) == 2) {
16             sprintf(line, "%d\n", int1 + int2);
17             n = strlen(line);
18             if (write(acceptfd, line, n) != n) {
19                 printf("write error\n");
20                 return 0;
21             }
22             printf("i am working on %d + %d = %s\n", int1, int2, line);
23         }
24         else {
25             if (write(acceptfd, "invalid args\n", 13) != 13) {
26                 printf("write msg error\n");
27                 return 0;
28             }
29         }
30     }
31 
32     close(acceptfd);
33 }
34 
35 if (fdetach("./pipe") < 0) {
36     printf("fdetach error\n");
37     return 0;
38 }
39 
40 printf("detach from file pipe ok\n");
41 close(listenfd);

首先調用serv_listen建立基本pipe,然後不斷在該pipe上調用serv_accept來獲取獨立的客戶端連接。之後的邏輯與以前一樣。

現在重點看下封裝的這兩個方法:

 1 int serv_listen(const char *name)
 2 {
 3     int tempfd;
 4     int fd[2];
 5     unlink(name);
 6     tempfd = creat(name, FIFO_MODE);
 7     if (tempfd < 0) {
 8         printf("creat failed\n");
 9         return -1;
10     }
11 
12     if (close(tempfd) < 0) {
13         printf("close temp fd failed\n");
14         return -2;
15     }
16 
17     if (pipe(fd) < 0) {
18         printf("pipe error\n");
19         return -3;
20     }
21 
22     if (ioctl(fd[1], I_PUSH, "connld") < 0) {
23         printf("I_PUSH connld failed\n");
24         close(fd[0]);
25         close(fd[1]);
26         return -4;
27     }
28 
29     printf("push connld ok\n");
30     if (fattach(fd[1], name) < 0) {
31         printf("fattach error\n");
32         close(fd[0]);
33         close(fd[1]);
34         return -5;
35     }
36 
37     printf("attach to file pipe ok\n");
38     close(fd[1]);
39     return fd[0];
40 }

serv_listen封裝了與建立基本pipe相關的代碼,首先確保pipe文件存在且可讀寫,然後創建普通的pipe,在fattach調用之前必需先PUSH一個connld模塊到該pipe STREAM中。這樣就大功告成!

 1 int serv_accept(int listenfd, uid_t *uidptr)
 2 {
 3     struct strrecvfd recvfd;
 4     if (ioctl(listenfd, I_RECVFD, &recvfd) < 0) {
 5         printf("I_RECVFD from listen fd failed\n");
 6         return -1;
 7     }
 8 
 9     if (uidptr)
10         *uidptr = recvfd.uid;
11 
12     return recvfd.fd;
13 }

當有客戶端連接上來的時候,使用I_RECVFD接收connld返回的另一個pipe的fd。之後的數據將在該pipe進行。

看了看,感覺和tcp的listen與accept別無二致,看來天下武功,至精深處都是英雄所見略同。

之前的多個客戶端同時運行的例子再跑一遍,觀察attach端輸出:

-bash-3.2$ ./add4
push connld ok
attach to file pipe ok
accept a client, fd = 4, uid = 101
source: 1 1
i am working on 1 + 1 = 2
accept a client, fd = 4, uid = 101
source: 2 2
i am working on 2 + 2 = 4
accept a client, fd = 4, uid = 101
source: 3 3
i am working on 3 + 3 = 6
accept a client, fd = 4, uid = 101
source: 4 4
i am working on 4 + 4 = 8

 一切正常。再看下腳本中四個進程的輸出:

-bash-3.2$ ./padd4.sh
-bash-3.2$ open file pipe ok, fd = 3
1 1 = 2
open file pipe ok, fd = 3
2 2 = 4
open file pipe ok, fd = 3
3 3 = 6
open file pipe ok, fd = 3
4 4 = 8

 也是沒問題的,既沒有出現多個請求粘連的情況,也沒有出現請求與應答錯位的情況。

 

測試程序

 

 

4.結論

Solaris 上的pipe不僅可以全雙工通訊、不依賴父子進程關係,還可以實現類似tcp那樣分離多個客戶端通訊連接的能力。

雖然Solaris前途未卜,但是希望一些好的東西還是能流傳下來,就比如這個神奇的pipe。

 

看完今天的文章,你是不是對特立獨行的Solaris又加深了一層了解?歡迎留言區說說你認識的Solaris。

 

 

 

 

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

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

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

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

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

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

[ch02-03] 梯度下降

系列博客,原文在筆者所維護的github上:,
點擊star加星不要吝嗇,星越多筆者越努力。

2.3 梯度下降

2.3.1 從自然現象中理解梯度下降

在大多數文章中,都以“一個人被困在山上,需要迅速下到谷底”來舉例,這個人會“尋找當前所處位置最陡峭的地方向下走”。這個例子中忽略了安全因素,這個人不可能沿着最陡峭的方向走,要考慮坡度。

在自然界中,梯度下降的最好例子,就是泉水下山的過程:

  1. 水受重力影響,會在當前位置,沿着最陡峭的方向流動,有時會形成瀑布(梯度下降);
  2. 水流下山的路徑不是唯一的,在同一個地點,有可能有多個位置具有同樣的陡峭程度,而造成了分流(可以得到多個解);
  3. 遇到坑窪地區,有可能形成湖泊,而終止下山過程(不能得到全局最優解,而是局部最優解)。

2.3.2 梯度下降的數學理解

梯度下降的數學公式:

\[\theta_{n+1} = \theta_{n} – \eta \cdot \nabla J(\theta) \tag{1}\]

其中:

  • \(\theta_{n+1}\):下一個值;
  • \(\theta_n\):當前值;
  • \(-\):減號,梯度的反向;
  • \(\eta\):學習率或步長,控制每一步走的距離,不要太快以免錯過了最佳景點,不要太慢以免時間太長;
  • \(\nabla\):梯度,函數當前位置的最快上升點;
  • \(J(\theta)\):函數。

梯度下降的三要素

  1. 當前點;
  2. 方向;
  3. 步長。

為什麼說是“梯度下降”?

“梯度下降”包含了兩層含義:

  1. 梯度:函數當前位置的最快上升點;
  2. 下降:與導數相反的方向,用數學語言描述就是那個減號。

亦即與上升相反的方向運動,就是下降。

圖2-9 梯度下降的步驟

圖2-9解釋了在函數極值點的兩側做梯度下降的計算過程,梯度下降的目的就是使得x值向極值點逼近。

2.3.3 單變量函數的梯度下降

假設一個單變量函數:

\[J(x) = x ^2\]

我們的目的是找到該函數的最小值,於是計算其微分:

\[J'(x) = 2x\]

假設初始位置為:

\[x_0=1.2\]

假設學習率:

\[\eta = 0.3\]

根據公式(1),迭代公式:

\[x_{n+1} = x_{n} – \eta \cdot \nabla J(x)= x_{n} – \eta \cdot 2x\tag{1}\]

假設終止條件為J(x)<1e-2,迭代過程是:

x=0.480000, y=0.230400
x=0.192000, y=0.036864
x=0.076800, y=0.005898
x=0.030720, y=0.000944

上面的過程如圖2-10所示。

圖2-10 使用梯度下降法迭代的過程

2.3.4 雙變量的梯度下降

假設一個雙變量函數:

\[J(x,y) = x^2 + \sin^2(y)\]

我們的目的是找到該函數的最小值,於是計算其微分:

\[{\partial{J(x,y)} \over \partial{x}} = 2x\]
\[{\partial{J(x,y)} \over \partial{y}} = 2 \sin y \cos y\]

假設初始位置為:

\[(x_0,y_0)=(3,1)\]

假設學習率:

\[\eta = 0.1\]

根據公式(1),迭代過程是的計算公式:
\[(x_{n+1},y_{n+1}) = (x_n,y_n) – \eta \cdot \nabla J(x,y)\]
\[ = (x_n,y_n) – \eta \cdot (2x,2 \cdot \sin y \cdot \cos y) \tag{1}\]

根據公式(1),假設終止條件為\(J(x,y)<1e-2\),迭代過程如表2-3所示。

表2-3 雙變量梯度下降的迭代過程

迭代次數 x y J(x,y)
1 3 1 9.708073
2 2.4 0.909070 6.382415
15 0.105553 0.063481 0.015166
16 0.084442 0.050819 0.009711

迭代16次后,J(x,y)的值為0.009711,滿足小於1e-2的條件,停止迭代。

上面的過程如表2-4所示,由於是雙變量,所以需要用三維圖來解釋。請注意看兩張圖中間那條隱隱的黑色線,表示梯度下降的過程,從紅色的高地一直沿着坡度向下走,直到藍色的窪地。

表2-4 在三維空間內的梯度下降過程

觀察角度1 觀察角度2

2.3.5 學習率η的選擇

在公式表達時,學習率被表示為\(\eta\)。在代碼里,我們把學習率定義為learning_rate,或者eta。針對上面的例子,試驗不同的學習率對迭代情況的影響,如表2-5所示。

表2-5 不同學習率對迭代情況的影響

學習率 迭代路線圖 說明
1.0 學習率太大,迭代的情況很糟糕,在一條水平線上跳來跳去,永遠也不能下降。
0.8 學習率大,會有這種左右跳躍的情況發生,這不利於神經網絡的訓練。
0.4 學習率合適,損失值會從單側下降,4步以後基本接近了理想值。
0.1 學習率較小,損失值會從單側下降,但下降速度非常慢,10步了還沒有到達理想狀態。

代碼位置

ch02, Level3, Level4, Level5

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

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

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

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

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

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