using System;
using System.Text;
using System.Threading.Tasks;
using Confluent.Kafka;

public class KafkaProducer
{
    private const string Topic = "large-messages";
    private const int MaxMessageSize = 1024 * 1024; // 1MB，可以根據(jù)實(shí)際情況調(diào)整

    public async Task SendLargeMessageAsync(string largeMessage)
    {
        var producerConfig = new ProducerConfig { BootstrapServers = "localhost:9092" }; // 配置Kafka服務(wù)器地址
        using var producer = new ProducerBuilder<string, string>(producerConfig).Build();

        int chunkSize = MaxMessageSize - 100; // 留出一些空間用于消息頭和分塊信息
        byte[] largeMessageBytes = Encoding.UTF8.GetBytes(largeMessage);
        int totalChunks = (int)Math.Ceiling((double)largeMessageBytes.Length / chunkSize);

        for (int i = 0; i < totalChunks; i++)
        {
            int startIndex = i * chunkSize;
            int endIndex = Math.Min(startIndex + chunkSize, largeMessageBytes.Length);
            byte[] chunk = new byte[endIndex - startIndex];
            Array.Copy(largeMessageBytes, startIndex, chunk, 0, chunk.Length);
            string chunkMessage = Encoding.UTF8.GetString(chunk);
            string key = $"Chunk-{i+1}-{totalChunks}"; // 用于在接收端重組消息

            await producer.ProduceAsync(Topic, new Message<string, string> { Key = key, Value = chunkMessage });
        }
    }
}

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Confluent.Kafka;

public class KafkaConsumer
{
    private const string Topic = "large-messages";
    private const string GroupId = "large-message-consumer-group";

    public async Task ConsumeLargeMessagesAsync()
    {
        var consumerConfig = new ConsumerConfig
        {
            BootstrapServers = "localhost:9092", // 配置Kafka服務(wù)器地址
            GroupId = GroupId,
            AutoOffsetReset = AutoOffsetReset.Earliest // 從最早的消息開始消費(fèi)
        };
        using var consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
        consumer.Subscribe(Topic);

        var chunks = new Dictionary<string, StringBuilder>(); // 用于存儲(chǔ)和組裝消息塊

        while (true) // 持續(xù)消費(fèi)消息，直到程序被終止或遇到錯(cuò)誤
        {
            try
            {
                var result = consumer.Consume(); // 消費(fèi)下一條消息
                string key = result.Key; // 獲取消息塊的關(guān)鍵信息（如：Chunk-1-3）
                string chunk = result.Value; // 獲取消息塊內(nèi)容

                if (!chunks.ContainsKey(key.Split('-')[1])) // 如果這是新消息的第一個(gè)塊，則創(chuàng)建一個(gè)新的StringBuilder來(lái)存儲(chǔ)它
                {
                    chunks[key.Split('-')[1]] = new StringBuilder(chunk);
                }
                else // 否則，將塊追加到現(xiàn)有的StringBuilder中
                {
                    chunks[key.Split('-')[1]].Append(chunk);
                }

                // 檢查是否已接收完整個(gè)大消息的所有塊
                if (IsCompleteMessage(key, chunks))
                {
                    string largeMessage = chunks[key.Split('-')[1]].ToString(); // 組裝完整的大消息
                    Console.WriteLine($"Received large message: {largeMessage}"); // 處理大消息（此處僅為打印輸出）
                    chunks.Remove(key.Split('-')[1]); // 清理已處理完的消息塊數(shù)據(jù)，以節(jié)省內(nèi)存空間
                }
            }
            catch (ConsumeException e) // 處理消費(fèi)過(guò)程中可能發(fā)生的異常（如網(wǎng)絡(luò)問題、Kafka服務(wù)器故障等）
            {
                Console.WriteLine($"Error occurred: {e.Error.Reason}");
            }
        }
    }

    private bool IsCompleteMessage(string key, Dictionary<string, StringBuilder> chunks) // 檢查是否已接收完整個(gè)大消息的所有塊
    {
        string[] keyParts = key.Split('-'); // 解析關(guān)鍵信息（如：Chunk-1-3）以獲取總塊數(shù)（如：3）和當(dāng)前塊號(hào)（如：1）等信息。這里假設(shè)關(guān)鍵信息的格式為“Chunk-<當(dāng)前塊號(hào)>-<總塊數(shù)>”。在實(shí)際應(yīng)用中，你可能需要根據(jù)實(shí)際情況調(diào)整此解析邏輯。同時(shí)，為了簡(jiǎn)化示例代碼，這里省略了對(duì)解析結(jié)果的有效性檢查（如確保當(dāng)前塊號(hào)在有效范圍內(nèi)等）。在實(shí)際應(yīng)用中，你應(yīng)該添加這些檢查以確保代碼的健壯性。另外，“<”和“>”符號(hào)僅用于說(shuō)明格式，并非實(shí)際出現(xiàn)在關(guān)鍵信息中。在實(shí)際應(yīng)用中，你應(yīng)該使用合適的分隔符（如“-”）來(lái)分割關(guān)鍵信息中的各個(gè)部分。最后，請(qǐng)注意在實(shí)際應(yīng)用中處理可能出現(xiàn)的異常情況（如關(guān)鍵信息格式不正確等）。如果關(guān)鍵信息的格式與示例中的不同，請(qǐng)相應(yīng)地調(diào)整解析邏輯。同時(shí)也要注意處理可能出現(xiàn)的異常情況以確保代碼的健壯性。 
        int totalChunks = int.Parse(keyParts[2]); // 獲取總塊數(shù)（假設(shè)關(guān)鍵信息的最后一個(gè)部分是總塊數(shù)）在實(shí)際應(yīng)用中，請(qǐng)確保關(guān)鍵信息的格式與你的解析邏輯相匹配，并處理可能出現(xiàn)的異常情況（如解析失敗等）。另外，“<”和“>”符號(hào)并非實(shí)際出現(xiàn)在關(guān)鍵信息中，而是用于說(shuō)明格式。你應(yīng)該使用合適的分隔符來(lái)分割關(guān)鍵信息中的各個(gè)部分。如果關(guān)鍵信息的格式與示例中的不同，請(qǐng)相應(yīng)地調(diào)整解析邏輯。同時(shí)也要注意在實(shí)際應(yīng)用中處理可能出現(xiàn)的異常情況以確保代碼的健壯性。此外，在解析完關(guān)鍵信息后，你可以通過(guò)比較已接收的消息塊數(shù)量與總塊數(shù)來(lái)判斷是否已接收完整個(gè)大消息的所有塊。具體實(shí)現(xiàn)方式可能因你的應(yīng)用場(chǎng)景和需求而有所不同。例如，你可以使用一個(gè)字典來(lái)存儲(chǔ)每個(gè)大消息的已接收塊，并在每次接收到新塊時(shí)更新字典中的信息。當(dāng)某個(gè)大消息的所有塊都已接收完畢時(shí)，你可以從字典中移除該消息的相關(guān)數(shù)據(jù)，并進(jìn)行后續(xù)處理（如重新組裝消息、觸發(fā)回調(diào)函數(shù)等）。在實(shí)現(xiàn)這一功能時(shí)，請(qǐng)注意線程安全和內(nèi)存管理方面的問題以確保程序的穩(wěn)定性和性能。 
        return chunks.Count == totalChunks; // 如果已接收的消息塊數(shù)量等于總塊數(shù)，則表示已接收完整個(gè)大消息的所有塊。注意，這里假設(shè)每個(gè)塊都會(huì)被正確接收且不會(huì)重復(fù)接收。在實(shí)際應(yīng)用中，你可能需要添加額外的邏輯來(lái)處理丟包、重傳等情況以確保數(shù)據(jù)的完整性和一致性。同時(shí)，也要注意優(yōu)化內(nèi)存使用以避免內(nèi)存泄漏或溢出等問題。另外，“==”運(yùn)算符用于比較兩個(gè)值是否相等。在這里，它用于比較已接收的消息塊數(shù)量（即字典中的鍵值對(duì)數(shù)量）與總塊數(shù)是否相等。如果相等，則表示已接收完整個(gè)大消息的所有塊；否則，表示還有未接收的塊需要繼續(xù)等待。 
    }
}