前言

上一篇【.Net Core微服務入門全紀錄（一）——項目搭建】講到要做到服務的靈活伸縮，那麼需要有一種機制來實現它，這個機制就是服務註冊與發現。當然這也並不是必要的，如果你的服務實例很少，並且很穩定，那麼就沒有必要使用服務註冊與發現。

服務註冊與發現

• 服務註冊：簡單理解，就是有一個註冊中心，我們的每個服務實例啟動時，都去註冊中心註冊一下，告訴註冊中心我的地址，端口等信息。同樣的服務實例要刪除時，去註冊中心刪除一下，註冊中心負責維護這些服務實例的信息。
• 服務發現：既然註冊中心維護了各個服務實例的信息，那麼客戶端通過註冊中心就很容易發現服務的變化了。

有了服務註冊與發現，客戶端就不用再去配置各個服務實例的地址，改為從註冊中心統一獲取。
那註冊中心又是怎麼保證每個地址的可用狀態呢，假如某個實例掛了怎麼辦呢？原則上掛掉的實例不應該被客戶端獲取到，所以就要提到：健康檢查 。

• 健康檢查：每個服務都需要提供一個用於健康檢查的接口，該接口不具備業務功能。服務註冊時把這個接口的地址也告訴註冊中心，註冊中心會定時調用這個接口來檢測服務是否正常，如果不正常，則將它移除，這樣就保證了服務的可用性。

常見註冊中心有 Consul、ZooKeeper、etcd、Eureka。

Consul

Consul官網：https://www.consul.io/
Consul的主要功能有服務註冊與發現、健康檢查、K-V存儲、多數據中心等。

• Consul安裝：很簡單，直接在官網下載解壓即可。
• Consul運行：在consul.exe目錄下打開命令行執行 consul.exe agent -dev
• 瀏覽器訪問：http://localhost:8500/

  Consul已成功運行。

服務註冊

• 首先Nuget安裝一下Consul：

  這個類庫里封裝了Consul的api操作，方便我們直接使用。當然自己去寫http調用Consul的接口也不是不行。。。接口說明：https://www.consul.io/api-docs

• 改造一下訂單服務的代碼：

ConsulHelper.cs：

    public static class ConsulHelper
    {
        /// <summary>
        /// 服務註冊到consul
        /// </summary>
        /// <param name="app"></param>
        /// <param name="lifetime"></param>
        public static IApplicationBuilder RegisterConsul(this IApplicationBuilder app, IConfiguration configuration, IHostApplicationLifetime lifetime) 
        {
            var consulClient = new ConsulClient(c =>
            {
                //consul地址
                c.Address = new Uri(configuration["ConsulSetting:ConsulAddress"]);
            });

            var registration = new AgentServiceRegistration()
            {
                ID = Guid.NewGuid().ToString(),//服務實例唯一標識
                Name = configuration["ConsulSetting:ServiceName"],//服務名
                Address = configuration["ConsulSetting:ServiceIP"], //服務IP
                Port = int.Parse(configuration["ConsulSetting:ServicePort"]),//服務端口 因為要運行多個實例，端口不能在appsettings.json里配置，在docker容器運行時傳入
                Check = new AgentServiceCheck()
                {
                    DeregisterCriticalServiceAfter = TimeSpan.FromSeconds(5),//服務啟動多久后註冊
                    Interval = TimeSpan.FromSeconds(10),//健康檢查時間間隔
                    HTTP = $"http://{configuration["ConsulSetting:ServiceIP"]}:{configuration["ConsulSetting:ServicePort"]}{configuration["ConsulSetting:ServiceHealthCheck"]}",//健康檢查地址
                    Timeout = TimeSpan.FromSeconds(5)//超時時間
                }
            };

            //服務註冊
            consulClient.Agent.ServiceRegister(registration).Wait();

            //應用程序終止時，取消註冊
            lifetime.ApplicationStopping.Register(() =>
            {
                consulClient.Agent.ServiceDeregister(registration.ID).Wait();
            });

            return app;
        }
    }

appsettings.json：

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft": "Warning",
      "Microsoft.Hosting.Lifetime": "Information"
    }
  },
  "AllowedHosts": "*",
  "ConsulSetting": {
    "ServiceName": "OrderService",
    "ServiceIP": "localhost",
    "ServiceHealthCheck": "/healthcheck",
    "ConsulAddress": "http://host.docker.internal:8500"//注意，docker容器內部無法使用localhost訪問宿主機器，如果是控制台啟動的話就用localhost
  }
}

Startup.cs：

    public class Startup
    {
        public Startup(IConfiguration configuration)
        {
            Configuration = configuration;
        }

        public IConfiguration Configuration { get; }

        // This method gets called by the runtime. Use this method to add services to the container.
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddControllers();
        }

        // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
        public void Configure(IApplicationBuilder app, IWebHostEnvironment env, IHostApplicationLifetime lifetime)
        {
            if (env.IsDevelopment())
            {
                app.UseDeveloperExceptionPage();
            }

            app.UseRouting();

            app.UseAuthorization();

            app.UseEndpoints(endpoints =>
            {
                endpoints.MapControllers();
            });

            //服務註冊
            app.RegisterConsul(Configuration, lifetime);
        }
    }

OrdersController.cs：

    [Route("[controller]")]
    [ApiController]
    public class OrdersController : ControllerBase
    {
        private readonly ILogger<OrdersController> _logger;
        private readonly IConfiguration _configuration;

        public OrdersController(ILogger<OrdersController> logger, IConfiguration configuration)
        {
            _logger = logger;
            _configuration = configuration;
        }

        [HttpGet]
        public IActionResult Get()
        {
            string result = $"【訂單服務】{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}——" +
                $"{Request.HttpContext.Connection.LocalIpAddress}:{_configuration["ConsulSetting:ServicePort"]}";
            return Ok(result);
        }
    }

HealthCheckController.cs：

    [Route("[controller]")]
    [ApiController]
    public class HealthCheckController : ControllerBase
    {
        /// <summary>
        /// 健康檢查接口
        /// </summary>
        /// <returns></returns>
        [HttpGet]
        public IActionResult Get()
        {
            return Ok();
        }
    }

至此就完成了服務註冊，取消註冊，健康檢查等功能的代碼編寫。

• 同樣的改造一下產品服務，代碼差不多一樣，就不貼了。

運行服務

繼續在docker中運行服務實例，不習慣docker的話用控制台啟動也行。–ConsulSetting:ServicePort參數就是傳入容器的端口信息。

docker build -t orderapi:1.0 -f ./Order.API/Dockerfile .
docker run -d -p 9060:80 --name orderservice orderapi:1.0 --ConsulSetting:ServicePort="9060"
docker run -d -p 9061:80 --name orderservice1 orderapi:1.0 --ConsulSetting:ServicePort="9061"
docker run -d -p 9062:80 --name orderservice2 orderapi:1.0 --ConsulSetting:ServicePort="9062"

docker build -t productapi:1.0 -f ./Product.API/Dockerfile .
docker run -d -p 9050:80 --name productservice productapi:1.0 --ConsulSetting:ServicePort="9050"
docker run -d -p 9051:80 --name productservice1 productapi:1.0 --ConsulSetting:ServicePort="9051"
docker run -d -p 9052:80 --name productservice2 productapi:1.0 --ConsulSetting:ServicePort="9052"

至此，6個服務器實例都已運行，並且成功註冊到Consul。

隨便停止2個服務：

可以看到停止的服務已經在Consul中被移除。注意，這個是我們停止程序時主動調用Consul移除的。

//應用程序終止時，取消註冊
lifetime.ApplicationStopping.Register(() =>
{
    consulClient.Agent.ServiceDeregister(registration.ID).Wait();
});

當然程序發生異常，健康檢查不能正確響應的話，Consul也會移除，有一點區別。

那麼註冊，發現，健康檢查功能都完成了，下一步就該考慮客戶端如何拿到這些服務實例的地址了。

代碼放在：https://github.com/xiajingren/NetCoreMicroserviceDemo

未完待續…

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

【其他文章推薦】

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

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

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

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

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

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

目錄

• 簡介
• Selector介紹
• 創建Selector
• 註冊Selector到Channel中
• SelectionKey
• selector 和 SelectionKey
• 總的例子
• 總結

簡介

NIO有三寶:Buffer,Channel，Selector少不了。本文將會介紹NIO三件套中的最後一套Selector，並在理解Selector的基礎上，協助小師妹發一張好人卡。我們開始吧。

Selector介紹

小師妹：F師兄，最近我的桃花有點旺，好幾個師兄莫名其妙的跟我打招呼，可是我一心向著工作，不想談論這些事情。畢竟先有事業才有家嘛。我又不好直接拒絕，有沒有什麼比較隱晦的方法來讓他們放棄這個想法？

更多內容請訪問www.flydean.com

這個問題，我沉思了大約0.001秒，於是給出了答案：給他們發張好人卡吧，應該就不會再來糾纏你了。

小師妹：F師兄，如果給他們發完好人卡還沒有用呢？

那就只能切斷跟他們的聯繫了，來個一刀兩斷。哈哈。

這樣吧，小師妹你最近不是在學NIO嗎？剛好我們可以用Selector來模擬一下發好人卡的過程。

假如你的志偉師兄和子丹師兄想跟你建立聯繫，每個人都想跟你建立一個溝通通道，那麼你就需要創建兩個channel。

兩個channel其實還好，如果有多個人都想同時跟你建立聯繫通道，那麼要維持這些通道就需要保持連接，從而浪費了資源。

但是建立的這些連接並不是時時刻刻都有消息在傳輸，所以其實大多數時間這些建立聯繫的通道其實是浪費的。

如果使用Selector就可以只啟用一個線程來監聽通道的消息變動，這就是Selector。

從上面的圖可以看出，Selector監聽三個不同的channel，然後交給一個processor來處理，從而節約了資源。

創建Selector

先看下selector的定義：

public abstract class Selector implements Closeable

Selector是一個abstract類，並且實現了Closeable，表示Selector是可以被關閉的。

雖然Selector是一個abstract類，但是可以通過open來簡單的創建：

Selector selector = Selector.open();

如果細看open的實現可以發現一個很有趣的現象：

public static Selector open() throws IOException {
        return SelectorProvider.provider().openSelector();
    }

open方法調用的是SelectorProvider中的openSelector方法。

再看下provider的實現：

 public SelectorProvider run() {
   if (loadProviderFromProperty())
        return provider;
    if (loadProviderAsService())
        return provider;
      provider = sun.nio.ch.DefaultSelectorProvider.create();
      return provider;
    }
 });

有三種情況可以加載一個SelectorProvider，如果系統屬性指定了java.nio.channels.spi.SelectorProvider，那麼從指定的屬性加載。

如果沒有直接指定屬性，則從ServiceLoader來加載。

最後如果都找不到的情況下，使用默認的DefaultSelectorProvider。

關於ServiceLoader的用法，我們後面會有專門的文章來講述。這裏先不做多的解釋。

註冊Selector到Channel中

ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress("localhost", 9527));
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

如果是在服務器端，我們需要先創建一個ServerSocketChannel，綁定Server的地址和端口，然後將Blocking設置為false。因為我們使用了Selector，它實際上是一個非阻塞的IO。

注意FileChannels是不能使用Selector的，因為它是一個阻塞型IO。

小師妹：F師兄，為啥FileChannel是阻塞型的呀？做成非阻塞型的不是更快？

小師妹，我們使用FileChannel的目的是什麼？就是為了讀文件呀，讀取文件肯定是一直讀一直讀，沒有可能讀一會這個channel再讀另外一個channel吧，因為對於每個channel自己來講，在文件沒讀取完之前，都是繁忙狀態，沒有必要在channel中切換。

最後我們將創建好的Selector註冊到channel中去。

SelectionKey

SelectionKey表示的是我們希望監聽到的事件。

總的來說，有4種Event：

• SelectionKey.OP_READ 表示服務器準備好，可以從channel中讀取數據。
• SelectionKey.OP_WRITE 表示服務器準備好，可以向channel中寫入數據。
• SelectionKey.OP_CONNECT 表示客戶端嘗試去連接服務端
• SelectionKey.OP_ACCEPT 表示服務器accept一個客戶端的請求

public static final int OP_READ = 1 << 0;
public static final int OP_WRITE = 1 << 2;
public static final int OP_CONNECT = 1 << 3;
public static final int OP_ACCEPT = 1 << 4;

我們可以看到上面的4個Event是用位運算來定義的，如果將這個四個event使用或運算合併起來，就得到了SelectionKey中的interestOps。

和interestOps類似，SelectionKey還有一個readyOps。

一個表示感興趣的操作，一個表示ready的操作。

最後，SelectionKey在註冊的時候，還可以attach一個Object，比如我們可以在這個對象中保存這個channel的id：

SelectionKey key = channel.register(
  selector, SelectionKey.OP_ACCEPT, object);
key.attach(Object);
Object object = key.attachment();

object可以在register的時候傳入，也可以調用attach方法。

最後，我們可以通過key的attachment方法，獲得該對象。

selector 和 SelectionKey

我們通過selector.select()這個一個blocking操作，來獲取一個ready的channel。

然後我們通過調用selector.selectedKeys()來獲取到SelectionKey對象。

在SelectionKey對象中，我們通過判斷ready的event來處理相應的消息。

總的例子

接下來，我們把之前將的串聯起來，先建立一個小師妹的ChatServer：

public class ChatServer {

    private static String BYE_BYE="再見";

    public static void main(String[] args) throws IOException, InterruptedException {
        Selector selector = Selector.open();
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress("localhost", 9527));
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        ByteBuffer byteBuffer = ByteBuffer.allocate(512);

        while (true) {
            selector.select();
            Set<SelectionKey> selectedKeys = selector.selectedKeys();
            Iterator<SelectionKey> iter = selectedKeys.iterator();
            while (iter.hasNext()) {
                SelectionKey selectionKey = iter.next();
                if (selectionKey.isAcceptable()) {
                    register(selector, serverSocketChannel);
                }
                if (selectionKey.isReadable()) {
                    serverResonse(byteBuffer, selectionKey);
                }
                iter.remove();
            }
            Thread.sleep(1000);
        }
    }

    private static void serverResonse(ByteBuffer byteBuffer, SelectionKey selectionKey)
            throws IOException {
        SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
        socketChannel.read(byteBuffer);
        byteBuffer.flip();
        byte[] bytes= new byte[byteBuffer.limit()];
        byteBuffer.get(bytes);
        log.info(new String(bytes).trim());
        if(new String(bytes).trim().equals(BYE_BYE)){
            log.info("說再見不如不見！");
            socketChannel.write(ByteBuffer.wrap("再見".getBytes()));
            socketChannel.close();
        }else {
            socketChannel.write(ByteBuffer.wrap("你是個好人".getBytes()));
        }
        byteBuffer.clear();
    }

    private static void register(Selector selector, ServerSocketChannel serverSocketChannel)
            throws IOException {
        SocketChannel socketChannel = serverSocketChannel.accept();
        socketChannel.configureBlocking(false);
        socketChannel.register(selector, SelectionKey.OP_READ);
    }
}

上面例子有兩點需要注意，我們在循環遍歷中，當selectionKey.isAcceptable時，表示服務器收到了一個新的客戶端連接，這個時候我們需要調用register方法，再註冊一個OP_READ事件到這個新的SocketChannel中，然後繼續遍歷。

第二，我們定義了一個stop word，當收到這個stop word的時候，會直接關閉這個client channel。

再看看客戶端的代碼：

public class ChatClient {

    private static SocketChannel socketChannel;
    private static ByteBuffer byteBuffer;

    public static void main(String[] args) throws IOException {

        ChatClient chatClient = new ChatClient();
        String response = chatClient.sendMessage("hello 小師妹！");
        log.info("response is {}", response);
        response = chatClient.sendMessage("能不能？");
        log.info("response is {}", response);
        chatClient.stop();

    }

    public void stop() throws IOException {
        socketChannel.close();
        byteBuffer = null;
    }

    public ChatClient() throws IOException {
        socketChannel = SocketChannel.open(new InetSocketAddress("localhost", 9527));
        byteBuffer = ByteBuffer.allocate(512);
    }

    public String sendMessage(String msg) throws IOException {
        byteBuffer = ByteBuffer.wrap(msg.getBytes());
        String response = null;
        socketChannel.write(byteBuffer);
        byteBuffer.clear();
        socketChannel.read(byteBuffer);
        byteBuffer.flip();
        byte[] bytes= new byte[byteBuffer.limit()];
        byteBuffer.get(bytes);
        response =new String(bytes).trim();
        byteBuffer.clear();
        return response;

    }
}

客戶端代碼沒什麼特別的，需要注意的是Buffer的讀取。

最後輸出結果：

server收到： INFO com.flydean.ChatServer - hello 小師妹！
client收到: INFO com.flydean.ChatClient - response is 你是個好人
server收到： INFO com.flydean.ChatServer - 能不能？
client收到： INFO com.flydean.ChatClient - response is 再見

解釋一下整個流程：志偉跟小師妹建立了一個連接，志偉向小師妹打了一個招呼，小師妹給志偉發了一張好人卡。志偉不死心，想繼續糾纏，小師妹回復再見，然後自己關閉了通道。

總結

本文介紹了Selector和channel在發好人卡的過程中的作用。

• 區塊鏈從入門到放棄系列教程-涵蓋密碼學,超級賬本,以太坊,Libra,比特幣等持續更新
• Spring Boot 2.X系列教程:七天從無到有掌握Spring Boot-持續更新
• Spring 5.X系列教程:滿足你對Spring5的一切想象-持續更新
• java程序員從小工到專家成神之路（2020版）-持續更新中,附詳細文章教程

本文作者：flydean程序那些事

本文鏈接：http://www.flydean.com/java-io-nio-selector/

本文來源：flydean的博客

歡迎關注我的公眾號:程序那些事，更多精彩等着您！

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

【其他文章推薦】

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

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

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

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

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

Kakfa在大數據消息引擎領域，絕對是沒有爭議的國民老公。

這是kafka系列的第一篇文章。預計共出20篇系列文章，全部原創，從0到1，跟你一起死磕kafka。

本文盤點了 Kafka 的各種術語並且進行解讀，術語可能比較枯燥，但真的是精髓中的精髓！

了解Kafka之前我們必須先掌握它的相關概念和術語，這對於後面深入學習 Kafka 各種功能將大有裨益。所以，枯燥你也得給我看完！

大概是有這麼些東西要掌握，不多不多，預計20分鐘可以吃透：

主題層

主題層有三個兒子，分別叫做:Topic、Partition、Replica。既然我說是三個兒子，那你懂了，是不可分割的整體。

Topic（主題）

Kafka 是分佈式的消息引擎系統，它的主要功能是提供一套完備的消息（Message）發布與訂閱解決方案。

在 Kafka 中，發布訂閱的對象是主題（Topic），你可以為每個業務、每個應用甚至是每類數據都創建專屬的主題。

一個Topic是對一組消息的歸納。也可以理解成傳統數據庫里的表，或者文件系統里的一個目錄。

Partition（分區）

一個Topic通常都是由多個partition組成的，創建topic時候可以指定partition數量。

分區優勢

為什麼需要將Topic分區呢？如果你了解其他分佈式系統，你可能聽說過分片、分區域等說法，比如 MongoDB 和 Elasticsearch 中的 Sharding、HBase 中的 Region，其實它們都是相同的原理。

試想，如果一個Topic積累了太多的數據以至於單台 Broker 機器都無法容納了，此時應該怎麼辦呢？

一個很自然的想法就是，能否把數據分割成多份保存在不同的機器上？這不就是分區的作用嗎？其實就是解決伸縮性的問題，每個partition都可以放在獨立的服務器上。

當然優勢不僅於此，也可以提高吞吐量。kafka只允許單個partition的數據被一個consumer線程消費。因此，在consumer端，consumer并行度完全依賴於被消費的分區數量。綜上所述，通常情況下，在一個Kafka集群中，partition的數量越多，意味着可以到達的吞吐量越大。

partition結構

每個partition對應於一個文件夾，該文件夾下存儲該partition的數據和索引文件。

如圖所示,可以看到兩個文件夾，都對應着一個叫做asd的topic，在該台服務器上有兩個分區，0和2，那麼1呢？在其他服務器上啦！畢竟是分佈式分佈的！

我們進去asd-0目錄中看看是什麼？有後綴為.index和.log的文件，他們就是該partition的數據和索引文件：

現在先不管它們是何方神聖，因為我會在【分區機制原理】這篇文章中詳細描述。

partition順序性

現在，我需要你睜大眼睛看看關於分區非常重要的一點：

【每個partition內部保證消息的順序。但是分區之間是不保證順序的】

這一點很重要，例如kafka中的消息是某個業務庫的數據，mysql binlog是有先後順序的，10:01分我沒有付款，所以pay_date為null，而10:02分我付款了，pay_date被更新了。

但到了kafka那，由於是分佈式的，多分區的，可就不一定能保證順序了，也許10:02分那條先來，這樣可就會引發嚴重生產問題了。因此，一般我們需要按表+主鍵來分區。保證同一主鍵的數據發送到同一個分區中。

如果你想要 kafka 中的所有數據都按照時間的先後順序進行存儲，那麼可以設置分區數為 1。

Replica （副本）

每個partition可以配置若干個副本。Kafka 定義了兩類副本：領導者副本（Leader Replica）和追隨者副本（Follower Replica）。只能有 1 個領導者副本和 N-1 個追隨者副本。

為啥要用副本？也很好理解，反問下自己為什麼重要的文件需要備份多份呢？備份機制（Replication）是實現高可用的一個手段。

需要注意的是：僅Leader Replica對外提供服務，與客戶端程序進行交互，生產者總是向領導者副本寫消息，而消費者總是從領導者副本讀消息。而Follower Replica不能與外界進行交互，它只做一件事：向領導者副本發送請求，請求領導者把最新生產的消息發給它，保持與領導者的同步。

如果對於剛剛所說的主題、分區、副本還有疑惑，那麼結合下面這張圖再思考一下，我相信你就可以玩轉它了：

下圖所示，TopicA，具有三個partition，每個partion都有1 個leader副本和 1 個follower者副本。為了保證高可用性，一台機器宕機不會有影響，因此leader副本和follower副本必然分佈在不同的機器上。

消息層

Kafka的官方定義是message system，由此我們可以知道Kafka 中最基本的數據單元無疑是消息message，它可理解成數據庫里的一條行或者一條記錄。消息是由字符數組組成。關於消息你必須知道這幾件事：

消息key

發送消息的時候指定 key，這個 key 也是個字符數組。key 用來確定消息寫入分區時，進入哪一個分區。你可以用有明確業務含義的字段作為key，比如用戶號，這樣就可以保證同一個用戶號進入同一個分區。

批量寫入

為了提高效率， Kafka 以批量batch的方式寫入。

一個 batch 就是一組消息的集合， 這一組的數據都會進入同一個 topic 和 partition（這個是根據 producer 的配置來定的） 。

每一個消息都進行一次網絡傳輸會很消耗性能，因此把消息收集到一起再同時處理就高效的多。

當然，這樣會引入更高的延遲以及吞吐量：batch 越大，同一時間處理的消息就越多。batch 通常都會進行壓縮，這樣在傳輸以及存儲的時候效率都更高一些。

位移
生產者向分區寫入消息，每條消息在分區中的位置信息由一個叫位移（Offset）的數據來表徵。分區位移總是從 0 開始，假設一個生產者向一個空分區寫入了 10 條消息，那麼這 10 條消息的位移依次是 0、1、2、…、9。

服務端

Kafka 的服務器端由被稱為 Broker 的服務進程構成，即一個 Kafka 集群由多個 Broker 組成，Kafka支持水平擴展，broker數量越多，集群吞吐量越高。在集群中每個broker都有一個唯一brokerid，不得重複。Broker 負責接收和處理客戶端發送過來的請求，以及對消息進行持久化。

一般會將不同的 Broker 分散運行在不同的機器上，這樣如果集群中某一台機器宕機，kafka可以自動選舉出其他機器上的 Broker 繼續對外提供服務。這其實就是 Kafka 提供高可用的手段之一。

controller

Kafka集群中會有一個或者多個broker，其中有且僅有一個broker會被選舉為控制器（Kafka Controller），它負責管理整個集群中所有分區和副本的狀態。

當某個分區的leader副本出現故障時，由控制器負責為該分區選舉新的leader副本。當檢測到某個分區的ISR集合發生變化時，由控制器負責通知所有broker更新其元數據信息。當為某個topic增加分區數量時，同樣還是由控制器負責分區的重新分配。

這幾句話可能會讓你覺得困惑_不要方 只是突出下控制器的職能很多，而這些功能的具體細節會在後面的文章中做具體的介紹。

Kafka中的控制器選舉的工作依賴於Zookeeper，成功競選為控制器的broker會在Zookeeper中創建/controller這個臨時（EPHEMERAL）節點，此臨時節點的內容參考如下：

其中version在目前版本中固定為1，brokerid表示稱為控制器的broker的id編號，timestamp表示競選稱為控制器時的時間戳。

兩種客戶端

Kafka有兩種客戶端。生產者和消費者。我們把生產者和消費者統稱為客戶端（Clients）。

向主題Topic發布消息Message的客戶端應用程序稱為生產者（Producer），生產者程序通常持續不斷地向一個或多個主題發送消息。

而訂閱這些主題消息的客戶端應用程序就被稱為消費者（Consumer）。和生產者類似，消費者也能夠同時訂閱多個主題的消息。

Producer

Producer 用來創建Message。在發布訂閱系統中，他們也被叫做 Publisher 發布者或 writer 寫作者。

通常情況下，會發布到特定的Topic，並負責決定發布到哪個分區（通常簡單的由負載均衡機制隨機選擇，或者通過key，或者通過特定的分區函數選擇分區。）
Producer分為Sync Producer 和 Aync Producer。

Sync Producer同步的生產者，即一定要某條消息成功才會發送下一條。所以它是低吞吐率、一般不會出現數據丟失。

Aync Producer異步的生產者，有個隊列的概念，是直接發送到隊列裏面，批量發送。高吞吐率、可能有數據丟失的。

Consumer 和 Consumer Group

消費者

Consumer 讀取消息。在發布訂閱系統中，也叫做 subscriber 訂閱者或者 reader 閱讀者。消費者訂閱一個或者多個主題，然後按照順序讀取主題中的數據。

消費位移

消費者需要記錄消費進度，即消費到了哪個分區的哪個位置上，這是消費者位移（Consumer Offset）。注意，這和上面所說的消息在分區上的位移完全不是一個概念。上面的“位移”表徵的是分區內的消息位置，它是不變的，即一旦消息被成功寫入到一個分區上，它的位移值就是固定的了。

而消費者位移則不同，它可能是隨時變化的，畢竟它是消費者消費進度的指示器嘛。通過存儲最後消費的 Offset，消費者應用在重啟或者停止之後，還可以繼續從之前的位置讀取。保存的機制可以是 zookeeper,或者 kafka 自己。

消費者組

ConsumerGroup：消費者組，指的是多個消費者實例組成一個組來消費一組主題，分區只能被消費者組中的其中一個消費者去消費，組員之間不能重複消費。

為什麼要引入消費者組呢？主要是為了提升消費者端的吞吐量。多個消費者實例同時消費，加速整個消費端的吞吐量（TPS）。

當然它的作用不僅僅是瓜分訂閱主題的數據，加速消費。它們還能彼此協助。假設組內某個實例掛掉了，Kafka 能夠自動檢測到，然後把這個 Failed 實例之前負責的分區轉移給其他活着的消費者，這個過程稱之為重平衡(Rebalance)。

你務必先把這個詞記住，它是kafka大名鼎鼎的重平衡機制，生產出現的異常問題很多都是由於它導致的。後續我會在【kafka大名鼎鼎又臭名昭著的重平衡】文章中詳細分析。

Zookeeper

zookeeper目前在kafka中扮演着舉重輕重的角色和作用~是kafka不可缺少的一個組件。

目前，Apache Kafka 使用 Apache ZooKeeper 來存儲它的元數據，比如brokers信息、分區的位置和主題的配置等數據就是存儲在 ZooKeeper 集群中。

注意我的用詞，我只說是目前。why？在 2019 年社區提出了一個計劃，以打破這種依賴關係，並將元數據管理引入 Kafka 本身。因為擁有兩個系統會導致大量的重複。

在之前的設計中，我們至少需要運行三個額外的 Java 進程，有時甚至更多。事實上，我們經常看到具有與 Kafka 節點一樣多的 ZooKeeper 節點的 Kafka 集群！此外，ZooKeeper 中的數據還需要緩存在 Kafka 控制器上，這導致了雙重緩存。

更糟糕的是，在外部存儲元數據限制了 Kafka 的可伸縮性。當 Kafka 集群啟動時，或者一個新的控制器被選中時，控制器必須從 ZooKeeper 加載集群的完整狀態。隨着元數據數量的增加，加載過程需要的時間也會增加，這限制了 Kafka 可以存儲的分區數量。

最後，將元數據存儲在外部會增加控制器的內存狀態與外部狀態不同步的可能性。

因此，未來，Kafka 的元數據將存儲在 Kafka 本身中，而不是存儲在 ZooKeeper 之類的外部系統中。可以持續關注kafka社區動態哦！

總結

一個典型的kafka集群包含若干個producer（向主題發布新消息），若干consumer（從主題訂閱新消息，用Consumer Offset表徵消費者消費進度），cousumergroup（多個消費者實例共同組成的一個組，共同消費多個分區），若干broker（服務器端進程）。還有zookeeper。

kafka發布訂閱的對象叫主題，每個Topic下可以有多個Partition，Partition中每條消息的位置信息又叫做消息位移(Offset)，Partition有副本機制，使得同一條消息能夠被拷貝到多個地方以提供數據冗餘，副本分為領導者副本和追隨者副本。

可以用下面這張圖來形象表達kafka的組成：

另外，再po一張思維導圖助你回顧本文所述的術語。

重要！！關注【胖滾豬學編程】公眾號發送”kafka”。獲取本文所有架構圖以及Kafka全系列思維導圖！

本文來源於公眾號：【胖滾豬學編程】。一枚集顏值與才華於一身，不算聰明卻足夠努力的女程序媛。用漫畫形式讓編程so easy and interesting！求關注！

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

【其他文章推薦】

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

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

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

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

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

※超省錢租車方案