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，可以根據實際情況調整

    public async Task SendLargeMessageAsync(string largeMessage)
    {
        var producerConfig = new ProducerConfig { BootstrapServers = "localhost:9092" }; // 配置Kafka服務器地址
        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服務器地址
            GroupId = GroupId,
            AutoOffsetReset = AutoOffsetReset.Earliest // 從最早的消息開始消費
        };
        using var consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
        consumer.Subscribe(Topic);

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

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

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

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

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