與Apache Kafka在Java中以事件為導(dǎo)向的體系結(jié)構(gòu)可以通過(guò)使用異步通信的事件進(jìn)行可擴(kuò)展的，鬆散的耦合系統(tǒng)。 1。生產(chǎn)者將事件發(fā)布給主題，消費(fèi)者無(wú)直接依賴性做出反應(yīng)。 2。 Kafka提供具有可重複性和高吞吐量的耐用，劃分的日誌。 3。使用kafkaproducer和kafkaconsumer使用Java的Kafka-Clients庫(kù)進(jìn)行手動(dòng)控制。 4。生產(chǎn)商將結(jié)構(gòu)化事件（最好是JSON或AVRO）使用帶回調(diào)處理的ProductErrecord進(jìn)行主題。 5。消費(fèi)者在循環(huán)中進(jìn)行調(diào)查，對(duì)其進(jìn)行處理並提交偏移，可選地使用消費(fèi)者組進(jìn)行水平縮放。 6。最佳實(shí)踐包括有意義的主題命名，事件版本，錯(cuò)誤處理，避免阻止操作，監(jiān)視消費(fèi)者滯後和確保持續(xù)性。 7?？蛇x的是，將Spring Kafka與@Kafkalistener和Kafkatemplate一起使用，以簡(jiǎn)化註釋驅(qū)動(dòng)的開發(fā)。通過(guò)從單個(gè)工作流程（例如訂單處理和利用Kafka的可靠性）開始，基於Java的分佈式系統(tǒng)可以通過(guò)事件驅(qū)動(dòng)的設(shè)計(jì)實(shí)現(xiàn)實(shí)時(shí)響應(yīng)能力，彈性和可維護(hù)性。

與Apache Kafka的Java中的事件驅(qū)動(dòng)的體系結(jié)構(gòu)（EDA）是一種強(qiáng)大的組合，用於構(gòu)建可擴(kuò)展，彈性和鬆散耦合的系統(tǒng)。它們沒(méi)有直接互相呼喚，而是通過(guò)事件進(jìn)行交流，而其他人發(fā)生了反應(yīng)。 Apache Kafka由於其高吞吐量，耐用性和實(shí)時(shí)處理功能而脫穎而出，成為此類體系結(jié)構(gòu)的骨幹。

這是您可以有效地在Java中實(shí)現(xiàn)EDA的方法。

1。事件驅(qū)動(dòng)體系結(jié)構(gòu)的核心概念

在EDA中：

• 生產(chǎn)者將事件（消息）發(fā)佈到主題。
• 消費(fèi)者訂閱主題並對(duì)事件做出反應(yīng)。
• 溝通是異步和脫鉤的。
• 組件不需要彼此了解，而只需要了解這些事件。

與kafka：

• 主題耐用，分區(qū)日誌。
• 事件存儲(chǔ)在可配置的時(shí)間中，允許重播。
• 多重消費(fèi)者可以獨(dú)立閱讀相同的事件流。

這使Kafka非常適合微服務(wù)，實(shí)時(shí)分析，審計(jì)記錄等。

2.在Java項(xiàng)目中設(shè)置Kafka

在您的Maven或Gradle項(xiàng)目中使用Apache Kafka客戶庫(kù)庫(kù)。

Maven依賴性：

 <依賴項(xiàng)>
    <groupId> org.apache.kafka </groupid>
    <Artifactid> kafka-clients </artifactid>
    <版本> 3.7.0 </version>
</dependency>

您將主要使用：

• KafkaProducer - 發(fā)送事件
• KafkaConsumer - 接收和處理活動(dòng)

還要考慮使用Spring Kafka ，如果您使用的是Spring Boot，它簡(jiǎn)化了配置和註釋驅(qū)動(dòng)的開發(fā)。

3。在Java舉行活動(dòng)

要發(fā)布事件，請(qǐng)創(chuàng)建KafkaProducer並將消息發(fā)送到主題。

導(dǎo)入org.apache.kafka.clients.producer。 *;
導(dǎo)入java.util.properties;

公共類訂單生產(chǎn)商{
    私人最終生產(chǎn)者<string，string> producer;

    public Order orderproducer（）{
        屬性props = new Properties（）;
        props.put（“ bootstrap.servers”，“ localhost：9092”）;
        props.put（“鍵。
        props.put（“ value.serializer”，org.apache.kafka.common.serialization.stringserializer&#39;）;

        this.producer =新的kafkaproducer <>（props）;
    }

    public void sendorderevent（字符串順序，字符串狀態(tài)）{
        字符串主題=“訂單事件”;
        ProducerRecord <String，String> record = 
            new producterRecord <>（主題，orderID，“ order” orderid&#39;is is status status）;

        producer.send（記錄，（元數(shù)據(jù)，異常） - > {
            如果（異常！= null）{
                system.err.println（“未能發(fā)送消息：” except.getMessage（））;
            } 別的 {
                system.out.println（“發(fā)送到分區(qū)” Metadata.partition（）   
                                   “帶有偏移” Metadata.offset（））;
            }
        }）;
    }

    public void close（）{
        producer.close（）;
    }
}

提示：使用JSON或AVRO進(jìn)行結(jié)構(gòu)化事件，而不是生產(chǎn)中的普通字符串。

4。與卡夫卡消費(fèi)者一起消費(fèi)活動(dòng)

創(chuàng)建一個(gè)消費(fèi)者，從主題中進(jìn)行調(diào)查並處理它們。

導(dǎo)入org.apache.kafka.clients.consumer。 *;
導(dǎo)入Java.Time.Duration;
導(dǎo)入java.util.collections;
導(dǎo)入java.util.properties;

公共班級(jí)訂購(gòu){
    私有最終的kafkaconsumer <字符串，字符串>消費(fèi)者；
    私有最終字符串主題=“訂單事件”;

    Public Ordorconsumer（）{
        屬性props = new Properties（）;
        props.put（“ bootstrap.servers”，“ localhost：9092”）;
        props.put（“ group.id”，“訂單處理組”）;
        props.put（“鍵。
        props.put（“ value.deserializer”，“ org.apache.kafka.common.serialization.stringDeserializer”）;
        props.put（“ auto.offset.reset”，“最早”）; //或“最新”

        this.consumer = new kafkaconsumer <>（props）;
    }

    public void start（）{
        consumer.subscribe（collections.singletonList（toble））;

        嘗試 {
            而（true）{
                commuterRecords <string，string> records = computer.poll（listation.ofmillis（1000））;
                for（commuterRecord <string，string> record：record）{
                    system.out.printf（“接收：key =％s，value =％s，分區(qū)=％d，offset =％d％n”，
                            record.key（），record.value（），record.partition（），record.offset（））;

                    //處理事件
                    processOrderevent（record.value（））;
                }
                //如果需要，請(qǐng)手動(dòng)承諾偏移
                commuter.commitsync（）;
            }
        } 最後 {
            consumer.close（）;
        }
    }

    私有void processorderevent（字符串值）{
        //業(yè)務(wù)邏輯：更新DB，通知用戶，觸發(fā)下一步，等。
        system.out.println（“處理事件：”值）;
    }
}

使用消費(fèi)者組水平擴(kuò)展 - 在組中讀取不同分區(qū)中的實(shí)例。

5。與Java的Kafka一起EDA的最佳實(shí)踐

• 使用有意義的主題名稱：例如， user-signed-up ， payment-failed ， inventory-updated 。
• 版本您的事件：在事件有效載荷中包含架構(gòu)版本或使用Confluent模式註冊(cè)表。
• 優(yōu)雅地處理錯(cuò)誤：不要崩潰不好的消息 - log，重試或發(fā)送到一個(gè)書信主題。
• 避免消費(fèi)者中的長(zhǎng)時(shí)間阻止操作：如果需要，將工作卸載到線程或消息隊(duì)列。
• 監(jiān)視滯後：跟蹤消費(fèi)者組滯後以檢測(cè)處理延遲。
• 使用IDEMTOTENT消費(fèi)者：確保處理事件兩次不會(huì)引起副作用。

6。可選：為簡(jiǎn)單起見使用Spring Kafka

使用Spring Boot，您可以使用@KafkaListener和@SendTo之類的註釋。

 @服務(wù)
公共類KafkaconsumerService {

    @kafkalistener（topics =“ order-events”，groupId =“ order-group”）
    public void listing（字符串消息）{
        System.out.println（“通過(guò)Spring Kafka收到：”消息）;
        //處理事件
    }
}

並容易產(chǎn)生：

 @Autowired
私有kafkatemplate <字符串，字符串> kafkatemplate;

public void sendevent（字符串主題，字符串鍵，字符串有效載荷）{
    kafkatemplate.send（主題，鍵，有效載荷）;
}

Spring Kafka處理序列化，線程和錯(cuò)誤處理的序列化處理。

概括

使用Apache Kafka在Java中構(gòu)建事件驅(qū)動(dòng)的系統(tǒng)，讓您：

• 切換服務(wù)
• 獨(dú)立擴(kuò)展
• 實(shí)時(shí)反應(yīng)
• 重播恢復(fù)或分析的事件

開始?。哼x擇一個(gè)工作流程（例如，訂單處理），將其建模為事件，然後將生產(chǎn)者和消費(fèi)者連接。然後根據(jù)需要擴(kuò)展。

憑藉Kafka的可靠性和Java的生態(tài)系統(tǒng)，EDA不僅可以可行，而且可以在大規(guī)模上保持且績(jī)效。

基本上，如果您在Java建造分佈式系統(tǒng)，Kafka是進(jìn)行事件驅(qū)動(dòng)的最佳工具之一。

以上是與Apache Kafka的Java事件驅(qū)動(dòng)的架構(gòu)的詳細(xì)內(nèi)容。更多資訊請(qǐng)關(guān)注PHP中文網(wǎng)其他相關(guān)文章！