Kafka 分散式流處理平台的訊息處理與擴充套件技術

Topic test is deleted
Topic test is created
2021-08-23 17:15:12,353 INFO services.py:1264 -- View the Ray dashboard at http://...
(pid=20100) Message delivered to topic test partition 8 offset 0
(pid=20100) Message delivered to topic test partition 2 offset 0
(pid=20103) Consumer 9e2773d4-f006-4d4d-aac3-fe75ed27f44b
(pid=20103) Assignment: [TopicPartition{topic=test,partition=0,offset=-1001,error=...
(pid=20107) Consumer bdedddd9-db16-4c24-a7ef-338e91b4e100
(pid=20107) Assignment: [TopicPartition{topic=test,partition=4,offset=-1001,error=...
(pid=20101) Consumer d76b7fad-0b98-4e03-92e3-510aac2fcb11
(pid=20101) Assignment: [TopicPartition{topic=test,partition=6,offset=-1001,error=...
(pid=20106) Consumer e3d181af-d095-4b7f-b3d6-830299c207a8

def process_message(message):
    return message.upper()

# Kafka consumer start method
def start_consumer():
    for message in kafka_consumer:
        processed_message = process_message(message)
        kafka_producer.send('uppercase_topic', processed_message)

user_visits = {}

def process_user_visit(user_id, visit):
    if user_id not in user_visits:
        user_visits[user_id] = []
    user_visits[user_id].append(visit)
    if len(user_visits[user_id]) > 100:
        user_visits[user_id].pop(0)

# Kafka consumer start method
def start_consumer():
    for message in kafka_consumer:
        user_id = message['user_id']
        visit = message['visit']
        process_user_visit(user_id, visit)

from enum import Enum
class Action(Enum):
    NONE = -1
    OFF = 0
    ON = 1

class BaseTemperatureController:
    def __init__(self, id: str):
        self.current_setting = None
        self.previous_command = -1
        self.id = id

    # 處理新訊息
    def process_new_message(self, message: dict):
        if 'measurement' in message: # 測量請求
            self.process_sensor_data(message)
        else: # 溫度設定請求
            self.set_temperature(message)

    # 設定新溫度
    def set_temperature(self, setting: dict):
        desired = setting['temperature']
        updelta = setting['up_delta']
        downdelta = setting['down_delta']
        print(f'Controller {self.id} 新溫度設定 {desired} 上限 {updelta} 下限 {downdelta}')
        self.current_setting = desired
        self.up_delta = updelta
        self.down_delta = downdelta

    # 處理新測量資料
    def process_sensor_data(self, sensor: dict) -> bool:
        # 必須設定期望溫度，否則忽略
        if self.current_setting is not None:
            # 計算期望操作
            measurement = sensor['measurement']
            action = Action.NONE
            if measurement > (self.current_setting + self.up_delta):
                action = Action.ON
            if measurement < (self.current_setting - self.down_delta):
                action = Action.OFF

            # 新操作
            if action != Action.NONE and self.previous_command != action:
                self.previous_command = action
                # 在這裡發布新操作到 Kafka
                return True
            else:
                return False
        else:
            return False

#### 內容解密：
玄貓在此範例中建立了一個 Python 基本溫控器類別 `BaseTemperatureController`，它具有幾種方法來處理不同型別的訊息。以下是每種方法的詳細說明：

- **`__init__`**：初始化方法接受一個 ID 作為引數，這個 ID 用來識別每個溫控器。
- **`process_new_message`**：根據接收到的訊息內容呼叫不同的方法。如果訊息包含 `measurement` 鍵，則呼叫 `process_sensor_data`；否則呼叫 `set_temperature`。
- **`set_temperature`**：處理來自溫控設定的新溫度設定。該設定包含目標溫度以及允許的上下限。
- **`process_sensor_data`**：處理從感測器接收到的新測量資料。如果已經設定了目標溫度，則比較當前測量值與目標溫度，並計算出需要執行的操作（開啟或關閉加熱器）。這個方法還會檢查操作是否已經改變，以避免重複傳送相同的控制命令。

### 溫控管理器實作示例

為了管理多個溫控器，我們引入了一個 `TemperatureControllerManager` 類別來處理這些個體溫控器。

```python
class TemperatureControllerManager:
    def __init__(self, producer):
        self.controllers = {}
        self.producer = producer

    def process_controller_message(self, key: str, request: dict):
        if key not in self.controllers: # 建立新的溫控器
            self.controllers[key] = BaseTemperatureController(key)
        controller = self.controllers[key]
        controller.process_new_message(request)

#### 內容解密：
玄貓在此範例中建立了一個 `TemperatureControllerManager` 類別來管理多個 `BaseTemperatureController` 個體。這個類別維護了一個字典來儲存每個溫控器例項，並根據接收到的訊息鍵來呼叫相應溫控器的方法。以下是詳細說明：

- **`__init__`**：初始化方法接受一個 Kafka 生產者物件作為引數，並初始化一個空字典來儲存溫控器例項。
- **`process_controller_message`**：根據接收到的訊息鍵和請求來呼叫相應溫控器的 `process_new_message` 方法。如果沒有找到相應溫控器例項，則建立一個新例項並儲存在字典中。

### Ray 中使用 threading 確保 Kafka 消費者持續執行

為了確保 Kafka 消費者能夠持續執行而不受測量計算幹擾，我們使用了執行緒（threading）。雖然這是我們為了簡單起見做出的一項簡化措施；在實際實作中，每個對溫度控制器發出的請求都應該包含一個 replyTo 主題名來確保回覆能夠到達正確例項。

#### 推薦進階步驟

雖然本範例簡化了許多內容以便理解與實作較為容易，但在真實情況下仍需考慮以下進階步驟：

1. **故障還原機制**：確保在節點失效時能夠還原狀態。
2. **高效平行處理**：最佳化資料流處理速度和吞吐量。
3. **安全性考量**：確保所有資料傳輸和儲存都符合安全標準。

透過不斷地最佳化和改進系統設計與功能特性將能夠提升整體系統之穩定性與可靠性。