2. 程式人生 > >Netty實戰開發(7):Netty結合kafka實現分散式訊息佇列

Netty實戰開發(7):Netty結合kafka實現分散式訊息佇列

阿新 • • 發佈：2018-12-11

在分散式遊戲伺服器系統中，訊息處理佇列主要解決問題就是解耦系統中的業務，使得每個系統看起來功能比較單一，而且解決一些全服資料共享等問題。 通常我們知道kafka是作為訊息佇列比較火的一種方式，其實還有（Active MQ，Rabbit MQ，Zero MQ）個人覺得kafka比較好用點，哈哈，習慣吧。 同樣我們來複習kafka基礎。

kafka基礎。 要向使用kafka。首先的搭建kafka環境。博主用Windows進行開發，所以本片文章用小部分的內容來介紹kafka的基本安裝。

首先我們要去kafka官方網站上下載kafka依賴包。https://kafka.apache.org/ 下載下來之後的到安裝包解壓。 在這裡插入圖片描述

當然我編寫了一個啟動指令碼，內容很簡單

@echo off
.\bin\windows\kafka-server-start.bat .\config\server.properties

執行kafka需要啟動zookeeper，所以需要確保機器上有可用的zookeeper。關於zookeeper，可以閱讀我上篇文章所寫的內容。https://blog.csdn.net/baidu_23086307/article/details/82769234 進入conf目錄。使用文字編譯器編輯service.properties檔案，修改log.dirs熟悉，修改為你自己的路徑。 例如我的路徑為這個我就修改成這了。

在這裡插入圖片描述

然後儲存檔案。 然後我們進入bin目錄，裡面有個Windows目錄，我們執行如下命令，

kafka-topics.bat --create --zookeeper 0.0.0.0:2181 --replication-factor 1 --partitions 1 --topic defaultTopic

@echo off
.\bin\windows\kafka-server-start.bat .\config\server.properties

當然我更喜歡把它編寫成一個bat指令碼。方便與下次直接使用。然後我們啟動kafka服務。 在這裡插入圖片描述 大概看到這樣的說明kafka啟動沒有問題。否者啟動出現各種問題，需要讀者自己排查問題。

整合kafka實現分散式訊息佇列

萬事具備，我們就要將kafka使用到我們的業務場景中來了。在遊戲伺服器中，我們通常用kafka做一些全域性的東西。比如全服聊天，全服活動之類的任務。也可以用於任務系統，解耦每一部分的邏輯。 我們的系統採用netty+spring整合的遊戲伺服器系統。當然spring也對kafka進行封裝了。所以我們在工程中需要新增kafka的依賴。


        <!--kafka-->
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-kafka</artifactId>
            <version>2.0.0.RELEASE</version>
            <exclusions>
                <exclusion>
                    <groupId>log4j</groupId>
                    <artifactId>log4j</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

新增完maven依賴後我們就可以使用spring提供的kafka進行封裝了。首先我們需要一個kafka中心伺服器的配置檔案。 簡單的配置了kafka的基礎屬性 game-kafka.properties

kafka.taskThreadSize=4
kafka.corePoolSize=4
kafka.maximumPoolSize=4
kafka.keepAliveTime=5000

然後我們新增kafka的 consumer配置，application-kafka-consumer.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd">

    <!-- 定義consumer的引數 -->
    <bean id="consumerProperties" class="java.util.HashMap">
        <constructor-arg>
            <map>
                <!-- 配置kafka的broke -->
                <entry key="bootstrap.servers" value="0.0.0.0:9092"/>
                <!-- 配置組-->
                <entry key="group.id" value="0"/>
                <entry key="enable.auto.commit" value="true"/>
                <entry key="auto.commit.interval.ms" value="1000"/>
                <entry key="session.timeout.ms" value="30000"/>
                <entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
                <entry key="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
            </map>
        </constructor-arg>
    </bean>

    <!-- 建立consumerFactory bean -->
    <bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
        <constructor-arg ref="consumerProperties"/>
    </bean>

    <!-- 實際執行訊息消費的類 -->
    <bean id="messageListenerConsumerService"
          class="com.twjitm.core.common.kafka.NettyKafkaConsumerListener"/>

    <!-- 消費者容器配置資訊 -->
    <bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties">
        <!-- 重要！配置topic -->
        <constructor-arg value="defaultTopic"/>
        <property name="messageListener" ref="messageListenerConsumerService"/>
    </bean>

    <!-- 建立kafka template bean，使用的時候，只需要注入這個bean，即可使用template的send訊息方法 -->
    <bean id="messageListenerContainer" class="org.springframework.kafka.listener.KafkaMessageListenerContainer"
          init-method="doStart">
        <constructor-arg ref="consumerFactory"/>
        <constructor-arg ref="containerProperties"/>
    </bean>
</beans>

然後在新增一個kafka的producer。application-kafka-producer.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd">

    <!--bootstrap.servers 消費者提供伺服器:example:0.0.0.1:9092,10.0.1.73:9092,10.0.1.74:9092,10.0.1.75:9092-->
    <!-- 定義producer的引數 -->
    <bean id="producerProperties" class="java.util.HashMap">
        <constructor-arg>
            <map>

                <entry key="bootstrap.servers"
                       value="0.0.0.0:9092"/>
                <entry key="group.id" value="0"/>
                <entry key="retries" value="1"/>
                <entry key="batch.size" value="16384"/>
                <entry key="linger.ms" value="1"/>
                <entry key="buffer.memory" value="33554432"/>
                <!--序列化方式-->
                <entry key="key.serializer"
                       value="org.apache.kafka.common.serialization.StringSerializer"/>
                <entry key="value.serializer"
                       value="org.apache.kafka.common.serialization.StringSerializer"/>
            </map>

        </constructor-arg>
    </bean>

    <!-- 建立kafka template需要使用的producer factory bean -->
    <bean id="producerFactory"
          class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
        <constructor-arg ref="producerProperties"/>
    </bean>

    <!-- 建立kafka template bean，使用的時候，只需要注入這個bean，即可使用template的send訊息方法 -->
    <bean id="KafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">
        <constructor-arg ref="producerFactory"/>
        <constructor-arg name="autoFlush" value="true"/>
        <property name="defaultTopic" value="defaultTopic"/>
        <!--<property name="producerListener" ref="producerListener"/>-->
    </bean>


</beans>

具體詳細配置正如程式碼中的註釋一樣，配置好kafka的基礎屬性，整合到spring 的bean物件中，使用spring容器來管理這些bean物件。省去了物件管理的工作。

程式實現： 我們需要定義一個kafka抽象任務，每個任務其實就是想kafka裡面傳送訊息。屬於producer範疇。AbstractKafkaPushTask .java

package com.twjitm.core.common.kafka;

/**
 * @author twjitm - [Created on 2018-09-04 21:31]
 */
public abstract class AbstractKafkaPushTask {
    private KafkaTaskType taskType;
    public Object value;

    public AbstractKafkaPushTask(KafkaTaskType taskType) {
        this.taskType = taskType;
    }

    public KafkaTaskType getTaskType() {
        return taskType;
    }


    public Object getValue() {
        return value;
    }

    /**
     * 需要將訊息值儲存到這裡面，值如何獲得由子類自己實心
     *
     * @param value
     */
    public abstract void setValue(Object value);
}

kafka訊息監聽器：訊息消費者.

package com.twjitm.core.common.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.MessageListener;

/**
 * kafka 訊息消費者
 *
 * @author twjitm- [Created on 2018-09-04 16:12]
 */
public class NettyKafkaConsumerListener implements MessageListener<String, String> {
    @Override
    public void onMessage(ConsumerRecord<String, String> consumerRecord) {

        System.out.println(consumerRecord);
        consumerRecord.value();
    }
}

在這我們對訊息做簡單的處理，但是在實際專案開發中我們還得更具訊息的具體型別做不同的業務處理，因此我們可以將訊息的具體應用做出自己業務邏輯處理。

訊息監聽器繼承MessageListener，而spring對這個介面做了處理，所以能夠監聽到訊息的到來。 在這裡插入圖片描述

有了訊息消費者，我們需要編寫一個訊息提供者，訊息提供者負責將訊息釋出到kafka訊息佇列中。實現程式碼如下：

package com.twjitm.core.common.kafka;

import com.alibaba.fastjson.JSON;
import com.twjitm.core.common.config.global.GlobalConstants;
import com.twjitm.core.common.config.global.KafkaConfig;
import com.twjitm.core.common.service.IService;
import com.twjitm.core.spring.SpringServiceManager;
import com.twjitm.threads.thread.NettyThreadNameFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.concurrent.*;

/**
 * kafka 訊息提供者
 *
 * @author twjitm- [Created on 2018-09-04 16:17]
 */
@Service
public class NettyKafkaProducerListener implements IService {
    private Logger logger = LoggerFactory.getLogger(NettyKafkaProducerListener.class);
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 訊息處理執行緒池
     */
    private volatile ExecutorService executorService;

    private boolean run = true;
    /**
     * 訊息佇列
     */
    private BlockingQueue<AbstractKafkaPushTask> queue;

    /**
     * 將訊息傳送給kafka中心
     *
     * @param abstractKafkaPushTask
     */
    private void sendMessage(AbstractKafkaPushTask abstractKafkaPushTask) {
        String type = abstractKafkaPushTask.getTaskType().getTypeName();
        String value = JSON.toJSONString(abstractKafkaPushTask.getValue());
        //本身send方法就是一個非同步執行方法
        ListenableFuture<SendResult<String, String>> result =
                kafkaTemplate.sendDefault(type, value);
        /**
         * 添加回調監聽
         */
        result.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                logger.error("KAFKA MESSAGE SEND FAIL", throwable);
            }

            @Override
            public void onSuccess(SendResult<String, String> kafkaTaskTypeObjectSendResult) {
                logger.info("KAFKA MESSAGE SEND SUCCESS");
            }
        });


    }

    /**
     * 將任務存放到佇列中
     *
     * @param task
     */
    public void put(AbstractKafkaPushTask task) {
        try {
            queue.put(task);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String getId() {
        return NettyKafkaProducerListener.class.getSimpleName();
    }

    @Override
    public void startup() throws Exception {
        run = true;
        queue = new LinkedBlockingQueue<>();
        NettyThreadNameFactory factory = new NettyThreadNameFactory(GlobalConstants.Thread.GAME_KAFKA_TASK_EXECUTOR);
        //開啟執行緒數量
        KafkaConfig kafkaConfig = SpringServiceManager.getSpringLoadService().getNettyGameServiceConfigService().getKafkaConfig();
        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(kafkaConfig.getCorePoolSize(), kafkaConfig.getMaximumPoolSize(), kafkaConfig.getKeepAliveTime(), TimeUnit.SECONDS, new LinkedBlockingQueue<>(), factory);
        executorService = poolExecutor;
        for (int i = 0; i < 4; i++) {
            executorService.execute(new Worker());
        }
    }

    @Override
    public void shutdown() throws Exception {
        logger.info("STOP KAFKA EXECUTOR");
        run = false;
        executorService.shutdown();
    }

    private class Worker implements Runnable {

        @Override
        public void run() {
            try {
                while (run) {
                    sendMessage(queue.take());
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


}

封裝看一個訊息本地訊息佇列，將上層呼叫過來的訊息存放到佇列中，再有佇列tack到kafka中。這樣保證每個程序進來的訊息都是有順序的。

測試：

使用spring整合kafka使用起來說是特別方便的。說以我們編寫一個測試類

package com.twjitm.kafka;

import com.twjitm.TestSpring;
import com.twjitm.core.common.kafka.KafkaTaskType;
import com.twjitm.core.common.kafka.NettyKafkaProducerListener;

import javax.annotation.Resource;

/**
 * @author twjitm- [Created on 2018-09-05 12:25]
 */

public class TestKafka {
    @Resource
    static NettyKafkaProducerListener nettyKafkaProducerListener;

    public static void main(String[] args) {
        TestSpring.initSpring();
        test();
    }

    public static void test() {
        WordChatTask task = new WordChatTask(KafkaTaskType.WORLD_CHAT);
        task.setValue("hello，world");
        nettyKafkaProducerListener.put(task);
    }

}

建立執行緒成功： 在這裡插入圖片描述

傳送訊息成功：

在這裡插入圖片描述

接收訊息成功：

在這裡插入圖片描述

相關程式碼已經提交到github，歡迎各位大大star

相關推薦

Netty實戰開發(7):Netty結合kafka實現分散式訊息佇列

在分散式遊戲伺服器系統中，訊息處理佇列主要解決問題就是解耦系統中的業務，使得每個系統看起來功能比較單一，而且解決一些全服資料共享等問題。 通常我們知道kafka是作為訊息佇列比較火的一種方式，其實還有（Active MQ，Rabbit MQ，Zero MQ）個人

docker：zookeeper與kafka實現分散式訊息佇列

一、安裝 下載映象 docker pull wurstmeister/zookeeper docker pull wurstmeister/kafka 通過docker-compose啟動 docker-compose.yml指令碼（zk+kafka版） vers

基於Redis實現分散式訊息佇列（彙總目錄）

基於Redis實現分散式訊息佇列（1）– 緣起 基於Redis實現分散式訊息佇列（2）– 分散式訊息佇列功能設計 基於Redis實現分散式訊息佇列（3）– Redis功能分析 基於Redis實現分散式訊息佇列（4）– 程式碼實現

基於Redis實現分散式訊息佇列（1）

1、為什麼需要訊息佇列？ 當系統中出現“生產“和“消費“的速度或穩定性等因素不一致的時候，就需要訊息佇列，作為抽象層，彌合雙方的差異。 舉個例子：業務系統觸發簡訊傳送申請，但簡訊傳送模組速度跟不上，需要將來不及處理的訊息暫存一下，緩衝壓力。 再舉個例子：調

基於Redis實現分散式訊息佇列（4）

1、訪問Redis的工具類 public class RedisManager { private static Pool<Jedis> pool; protected final static Logger logger

基於Redis實現分散式訊息佇列（3）

1、Redis是什麼鬼？ Redis是一個簡單的，高效的，分散式的，基於記憶體的快取工具。 假設好伺服器後，通過網路連線（類似資料庫），提供Key－Value式快取服務。 簡單，是Redis突出的特色。 簡單可以保證核心功能的穩定和優異。 2、效能

「Netty實戰 02」手把手教你實現自己的第一個 Netty 應用!新手也能搞懂！

[大家好，我是 **「後端技術進階」** 作者，一個熱愛技術的少年。](https://www.yuque.com/docs/share/aaa50642-c892-4c41-8c0c-9d2fc2b0d93c?#%20%E3%80%8A%E8%B5%B0%E8%BF%9B%E5%90%8E%E7%AB%AF

SpringBoot開發案例之整合Kafka實現訊息佇列

　　前言 　　最近在做一款秒殺的案例，涉及到了同步鎖、資料庫鎖、分散式鎖、程序內佇列以及分散式訊息佇列，這裡對SpringBoot整合Kafka實現訊息佇列做一個簡單的記錄。 　　Kafka簡介 　　Kafka是由Apache軟體基金會開發的一個開源流處理平臺，由Scala和Java編寫

Netty實戰-第一個Netty

      本部分將簡單介紹Netty的核心概念，核心概念就是學習Netty是如何攔截和處理異常。1 設定開發環境   設定開發環境的步驟包括如下三個部分：安裝jdk下載netty包安裝Ecplise2 Netty客戶端和伺服器概述    下面將構建一個完整的Netty伺服器

SpringBoot實戰開發（一）——HelloWord實現

1、建立Maven project專案點next，然後選擇quickstart模板以上完成Maven專案的建立2、引入Pom依賴<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://ww

Netty構建分散式訊息佇列實現原理淺析（十六）

　在本人的上一篇部落格文章：Netty構建分散式訊息佇列（AvatarMQ）設計指南之架構篇 中，重點向大家介紹了AvatarMQ主要構成模組以及目前存在的優缺點。最後以一個生產者、消費者傳遞訊息的例子，具體演示了AvatarMQ所具備的基本訊息路由功能。而本文的寫作

ASP.NET結合Redis實現分散式快取

  最近一個專案ASP.NET+MySQL 有的網頁開啟初始化的查詢需要10秒甚至更久，使用者體驗極差，而且併發量變大的時候網站容易崩潰 後來想了兩種解決方案都不是太滿意 1、資料庫裡建一張快取表，後臺作業定時去更新這張表，每次網頁開啟就可以直接從快取表裡查詢 2、使用者第一次開啟網站將資

SpringBoot專案開發(十五)：redisson實現分散式鎖

1.為什麼要使用分散式鎖?       在分散式場景下為了保證資料最終一致性。在單程序的系統中，存在多個執行緒可以同時改變某個變數（可變共享變數）時，就需要對變數或程式碼塊做同步(lock—synchronized)，使其在修改這種變數時能夠線性執行消除併發修

java結合redis實現分散式鎖

今天工作之餘，檢視一下利用redis來實現分散式鎖，因此，在檢視別人文章之餘，自己也來手動模擬實現Java的lock介面，來自己手動實現一個分散式鎖。擁有簡單的加鎖，解鎖，鎖中斷等操作。 利用redis的分散式鎖，主要還是利用redis的setnx命令，檢視redis文件，

Shiro許可權管理框架（二）：Shiro結合Redis實現分散式環境下的Session共享

首發地址：https://www.guitu18.com/post/2019/07/28/44.html 本篇是Shiro系列第二篇，使用Shiro基於Redis實現分散式環境下的Session共享。在講Session共享之前先說一下為什麼要做Session共享。 為什麼要做Session共享 什麼是Ses

kafka實現無訊息丟失與精確一次語義（exactly once）處理

在很多的流處理框架的介紹中，都會說kafka是一個可靠的資料來源，並且推薦使用Kafka當作資料來源來進行使用。這是因為與其他訊息引擎系統相比，kafka提供了可靠的資料儲存及備份機制。並且通過消費者位移這一概念，可以讓消費者在因某些原因宕機而重啟後，可以輕易得回到宕機前的位置。 但其實kafka的可靠性也

Kafka分散式訊息佇列

基本架構 Kafka分散式訊息佇列的作用： 解耦：將訊息生產階段和處理階段拆分開，兩個階段互相獨立各自實現自己的處理邏輯，通過Kafka提供的訊息寫入和消費介面實現對訊息的連線處理。降低開發複雜度，提高系統穩定性。 高吞吐率：kafka通過順序讀寫磁碟提供可以和記憶體隨機讀寫相匹敵的讀寫速度，靈活的客戶

基於Docker搭建分散式訊息佇列Kafka

本文基於Docker搭建一套單節點的Kafka訊息佇列，Kafka依賴Zookeeper為其管理叢集資訊，雖然本例不涉及叢集，但是該有的元件都還是會有，典型的kafka分散式架構如下圖所示。本例搭建的示例包含Zookeeper + Kafka + Kafka-manger mark &

php實現的訊息佇列類

<?php /** * Created by PhpStorm. * User: lin * Date: 2017/6/9 * Time: 11:19 * 實現php共享記憶體訊息佇列 */ class ShmQueue { private $maxQSize = 0;/

Spark Streaming實時流處理筆記（4）—— 分散式訊息佇列Kafka

1 Kafka概述 和訊息系統類似 1.1 訊息中介軟體 生產者和消費者 1.2 Kafka 架構和概念 producer:生產者（生產饅頭） consumer：消費者（吃饅頭） broker:籃子 topic : 主題，給饅頭帶一個標籤，（