构建基于CDC与Python的Meilisearch实时索引服务以支持iOS端

后端架构

文章字数: 3.7k

阅读时长: 16 分

我们的iOS应用最初的搜索功能是直接查询生产环境的PostgreSQL数据库，使用了几个LIKE '%query%'。这个方案在用户量过百后迅速崩溃，不仅查询缓慢，还给主库带来了不必要的压力。第一个迭代是引入Meilisearch，我们编写了一个Python脚本，每小时通过cron全量同步一次商品数据。这解决了数据库压力问题，但一小时的延迟对于一个期望即时性的电商应用来说，几乎是不可接受的。用户上架一件商品，却要等一个小时才能搜到，投诉随之而来。我们需要一个能将延迟降低到秒级的方案。

技术选型决策：从应用层触发到数据库日志捕获

最初的构想是在应用的业务逻辑中，当商品信息发生变更（创建、更新、删除）时，同步调用一个服务来更新Meilisearch索引。这个方案很快被否决。它将索引维护的逻辑与核心业务逻辑紧密耦合，增加了API请求的延迟，并且如果Meilisearch服务短暂不可用，整个商品更新操作就会失败。这是一个脆弱且不具备弹性的设计。

我们需要的是一个解耦的、异步的、可靠的方案。数据库触发器是另一个选项，但它将复杂的逻辑侵入到数据库层面，难以调试、版本控制和扩展，在真实项目中，我们倾向于让数据库尽可能“干净”。

最终，我们选择了变更数据捕获（Change Data Capture, CDC）的方案。具体技术栈定为：

PostgreSQL: 我们现有的主数据库，其Write-Ahead Log (WAL) 是CDC的数据源。
Debezium: 一个顶级的开源CDC平台，作为Kafka Connect的连接器运行，能够稳定地从PostgreSQL的WAL中捕获行级变更，并将它们作为结构化事件推送到Kafka。
Apache Kafka: 作为事件流的中间件。它的持久性和高吞吐量为我们提供了一个可靠的缓冲区，即使下游的索引消费服务宕机，数据变更事件也不会丢失。
Python: 用于编写消费Kafka中CDC事件的独立服务。Python的生态系统（confluent-kafka客户端、meilisearch-python-sdk）和开发效率非常适合这类中间件服务的开发。
Meilisearch: 目标搜索引擎，提供给iOS客户端进行毫秒级搜索。

这个架构的优势是显而易见的：索引服务与主应用完全解耦，主应用甚至不知道Meilisearch的存在。整个过程是异步的，对用户操作的延迟影响为零，并且具备极高的可用性和弹性。

graph TD
    subgraph "生产环境"
        A[iOS App] --> B{业务后端 API}
        B --> C[(PostgreSQL)]
    end

    subgraph "实时索引管道"
        C -- WAL --> D[Debezium Connector]
        D -- CDC Events --> E[(Apache Kafka)]
        E --> F[Python索引消费服务]
        F -- Indexing --> G[(Meilisearch)]
    end

    subgraph "搜索查询"
        A -- Search Queries --> H{API Gateway}
        H --> G
    end

    style C fill:#d8e8c2,stroke:#333,stroke-width:2px
    style E fill:#f9f2d0,stroke:#333,stroke-width:2px
    style G fill:#f9d0d0,stroke:#333,stroke-width:2px

注意: 在这个实现中，为了简化，iOS App将直接查询Meilisearch。在生产环境中，如上图所示，应当通过一个API Gateway来管理认证和安全。

基础设施搭建：Docker Compose下的模拟环境

在真实项目中，这些组件会部署在Kubernetes或专有服务器上。但在开发和演示阶段，docker-compose是管理这一切最有效的方式。

docker-compose.yml:

version: '3.8'

services:
  postgres:
    image: debezium/postgres:13
    container_name: postgres_db
    ports:
      - "5432:5432"
    environment:
      - POSTGRES_DB=app_db
      - POSTGRES_USER=app_user
      - POSTGRES_PASSWORD=app_password
    volumes:
      - ./pg_data:/var/lib/postgresql/data
    # Debezium需要修改postgresql.conf，必须设置wal_level=logical
    command: >
      postgres
      -c wal_level=logical

  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.3.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1

  connect:
    image: confluentinc/cp-kafka-connect:7.3.0
    container_name: kafka_connect
    depends_on:
      - kafka
    ports:
      - "8083:8083"
    environment:
      CONNECT_BOOTSTRAP_SERVERS: 'kafka:29092'
      CONNECT_REST_ADVERTISED_HOST_NAME: connect
      CONNECT_REST_PORT: 8083
      CONNECT_GROUP_ID: compose-connect-group
      CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_INTERNAL_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_INTERNAL_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_PLUGIN_PATH: /usr/share/java,/usr/share/confluent-hub-components
    volumes:
      - ./debezium-connector-postgres:/usr/share/confluent-hub-components/debezium-connector-postgres

  meilisearch:
    image: getmeili/meilisearch:v1.3
    container_name: meilisearch_engine
    ports:
      - "7700:7700"
    environment:
      - MEILI_MASTER_KEY=aMasterKey # 不要在生产中使用默认key
      - MEILI_ENV=development
    volumes:
      - ./meili_data:/meili_data

volumes:
  pg_data:
  meili_data:

这里有几个关键点：

PostgreSQL wal_level: 必须设置为 logical，Debezium才能工作。
Kafka Advertised Listeners: 这是docker-compose环境中常见的坑。我们需要配置PLAINTEXT://kafka:29092供容器内部通信，以及PLAINTEXT_HOST://localhost:9092供宿主机上的Python应用连接。
Debezium Connector Volume: 我们需要手动下载Debezium的PostgreSQL连接器，并将其挂载到kafka-connect容器的插件路径下。

启动环境后，需要向Kafka Connect注册我们的PostgreSQL连接器。

register-postgres-connector.sh:

#!/bin/bash
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
localhost:8083/connectors/ \
-d '{
  "name": "products-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "database.hostname": "postgres_db",
    "database.port": "5432",
    "database.user": "app_user",
    "database.password": "app_password",
    "database.dbname": "app_db",
    "database.server.name": "pgserver1",
    "table.include.list": "public.products",
    "publication.autocreate.mode": "all_tables",
    "plugin.name": "pgoutput"
  }
}'

执行此脚本后，Debezium会开始监控public.products表的变更，并将事件发送到名为pgserver1.public.products的Kafka Topic中。

核心实现：健壮的Python索引消费服务

消费服务是这个管道的核心。它必须是健壮的、可容错的，并且能正确地处理Debezium事件的复杂结构。

项目结构:

cdc-indexer/
├── main.py             # 服务主入口
├── consumer.py         # Kafka消费者与处理逻辑
├── meilisearch_client.py # Meilisearch交互客户端
├── config.py           # 配置管理
├── utils.py            # 工具函数与日志配置
└── requirements.txt

config.py - 配置管理
我们使用pydantic来管理环境变量，确保配置的类型安全。

# config.py
import os
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    KAFKA_BOOTSTRAP_SERVERS: str = "localhost:9092"
    KAFKA_TOPIC: str = "pgserver1.public.products"
    KAFKA_CONSUMER_GROUP_ID: str = "meilisearch-indexer-group"
    
    MEILISEARCH_HOST: str = "http://localhost:7700"
    MEILISEARCH_API_KEY: str = "aMasterKey"
    MEILISEARCH_INDEX_NAME: str = "products"

    # 批量处理配置
    BATCH_SIZE: int = 100
    BATCH_TIMEOUT_MS: int = 5000 # 5 seconds

    LOG_LEVEL: str = "INFO"

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = Settings()

meilisearch_client.py - 封装Meilisearch交互

# meilisearch_client.py
import meilisearch
from meilisearch.errors import MeiliSearchApiError
import logging

from config import settings

logger = logging.getLogger(__name__)

class MeiliSearchClient:
    def __init__(self):
        try:
            self.client = meilisearch.Client(
                settings.MEILISEARCH_HOST, 
                settings.MEILISEARCH_API_KEY
            )
            self.index = self.client.index(settings.MEILISEARCH_INDEX_NAME)
            # 确保主键已设置, 这对于add_or_update_documents至关重要
            self.index.update_settings({'primaryKey': 'id'})
            logger.info(f"成功连接到Meilisearch并选择了索引: {settings.MEILISEARCH_INDEX_NAME}")
        except Exception as e:
            logger.critical(f"无法连接到Meilisearch: {e}")
            raise

    def add_or_update_documents(self, documents: list[dict]):
        if not documents:
            return
        try:
            task = self.index.add_documents(documents, primary_key='id')
            logger.info(f"发送了 {len(documents)} 个文档到Meilisearch进行添加/更新。任务ID: {task.task_uid}")
        except MeiliSearchApiError as e:
            logger.error(f"向Meilisearch添加文档时出错: {e}")
            # 在真实项目中，这里可能需要将失败的批次推送到死信队列(DLQ)
            
    def delete_documents(self, document_ids: list[int]):
        if not document_ids:
            return
        try:
            task = self.index.delete_documents(document_ids)
            logger.info(f"发送了 {len(document_ids)} 个文档ID到Meilisearch进行删除。任务ID: {task.task_uid}")
        except MeiliSearchApiError as e:
            logger.error(f"从Meilisearch删除文档时出错: {e}")

meili_client = MeiliSearchClient()

consumer.py - 核心处理逻辑
这是最关键的部分。它需要处理Debezium消息格式、批处理以及错误。

# consumer.py
import json
import logging
import time
from threading import Thread, Event
from confluent_kafka import Consumer, KafkaError, KafkaException

from config import settings
from meilisearch_client import meili_client

logger = logging.getLogger(__name__)

class CDCProcessor:
    def __init__(self):
        self._documents_to_update = []
        self._document_ids_to_delete = []
        self._last_flush_time = time.time()
        self._shutdown_event = Event()

    def process_message(self, msg_value: dict):
        """解析单个Debezium消息并将其分类到更新或删除缓冲区"""
        payload = msg_value.get('payload', {})
        if not payload:
            logger.warning("收到了没有'payload'的空消息")
            return

        op = payload.get('op')
        
        # 'c' for create, 'u' for update, 'r' for initial snapshot read
        if op in ('c', 'u', 'r'):
            document = payload.get('after')
            if document:
                # 真实项目中可能需要数据清洗或转换
                # 例如，将某些字段类型转换为Meilisearch期望的类型
                self._documents_to_update.append(document)
        # 'd' for delete
        elif op == 'd':
            # 对于删除事件，Debezium在'before'字段中提供已删除行的数据
            document_before = payload.get('before')
            if document_before:
                doc_id = document_before.get('id')
                if doc_id:
                    self._document_ids_to_delete.append(doc_id)
        else:
            logger.warning(f"接收到未知的操作类型: {op}")

    def flush(self):
        """将缓冲区的数据批量推送到Meilisearch"""
        if not self._documents_to_update and not self._document_ids_to_delete:
            return

        logger.info(f"准备刷新... 更新: {len(self._documents_to_update)}, 删除: {len(self._document_ids_to_delete)}")
        
        try:
            meili_client.add_or_update_documents(self._documents_to_update)
            meili_client.delete_documents(self._document_ids_to_delete)
        finally:
            # 无论成功与否都清空缓冲区，防止重复处理。
            # 失败的批次应由外部机制（如DLQ）处理。
            self._documents_to_update.clear()
            self._document_ids_to_delete.clear()
            self._last_flush_time = time.time()
    
    def maybe_flush(self):
        """根据批次大小或超时时间决定是否刷新"""
        batch_full = len(self._documents_to_update) + len(self._document_ids_to_delete) >= settings.BATCH_SIZE
        timeout_reached = (time.time() - self._last_flush_time) * 1000 > settings.BATCH_TIMEOUT_MS
        
        if batch_full or timeout_reached:
            self.flush()

    def run(self):
        """主消费循环"""
        conf = {
            'bootstrap.servers': settings.KAFKA_BOOTSTRAP_SERVERS,
            'group.id': settings.KAFKA_CONSUMER_GROUP_ID,
            'auto.offset.reset': 'earliest',
            'enable.auto.commit': False # 我们将手动提交偏移量
        }
        consumer = Consumer(conf)
        
        try:
            consumer.subscribe([settings.KAFKA_TOPIC])
            logger.info(f"已订阅Kafka Topic: {settings.KAFKA_TOPIC}")

            while not self._shutdown_event.is_set():
                msg = consumer.poll(timeout=1.0)

                if msg is None:
                    # 没有新消息，检查是否需要因超时而刷新
                    self.maybe_flush()
                    continue
                
                if msg.error():
                    if msg.error().code() == KafkaError._PARTITION_EOF:
                        continue
                    else:
                        raise KafkaException(msg.error())

                # 核心逻辑：处理消息、刷新批次、提交偏移量
                msg_value = json.loads(msg.value().decode('utf-8'))
                self.process_message(msg_value)
                self.maybe_flush()
                consumer.commit(asynchronous=True)

        except KeyboardInterrupt:
            logger.info("接收到中断信号，正在关闭...")
        finally:
            self.flush() # 关闭前最后刷新一次
            consumer.close()
            logger.info("Kafka消费者已关闭")
            
    def shutdown(self):
        self._shutdown_event.set()

def run_consumer():
    processor = CDCProcessor()
    # 可以在这里添加信号处理来优雅地调用processor.shutdown()
    processor.run()

这里的关键设计：

手动提交偏移量: enable.auto.commit设为False。我们在处理完一批消息并成功（或决定放弃）将其发送到Meilisearch后，才提交偏移量。这提供了至少一次（at-least-once）的处理语义。
批处理与超时: maybe_flush方法实现了批处理的核心逻辑。它避免了对每条消息都进行一次网络请求，极大地提高了吞吐量。超时机制确保了即使在消息流量较低时，数据也能在可接受的延迟内被索引。
优雅关闭: 捕获KeyboardInterrupt并在退出前执行最后一次flush，确保缓冲区内的数据不会丢失。在生产环境中，这应该由SIGTERM信号处理器来完成。
错误处理: 当前的实现是在日志中记录Meilisearch的API错误。一个更健壮的系统会实现重试逻辑（例如使用tenacity库），并在多次重试失败后将该批次消息发送到死信队列（Dead-Letter Queue），以便后续进行人工排查，而不是阻塞整个消费流程。

iOS客户端集成：消费实时搜索结果

在iOS端，集成Meilisearch非常直接。由于Meilisearch提供了RESTful API，我们可以使用任何HTTP客户端库。

SearchViewModel.swift - 一个简单的SwiftUI示例

// SearchViewModel.swift
import Foundation
import Combine

struct Product: Codable, Identifiable {
    let id: Int
    let name: String
    let description: String
    let price: Double
}

struct SearchResult: Codable {
    let hits: [Product]
    let query: String
    let processingTimeMs: Int
}

class SearchViewModel: ObservableObject {
    @Published var searchText = ""
    @Published var searchResults: [Product] = []
    @Published var isLoading = false
    
    private var searchCancellable: AnyCancellable?
    
    // 在真实应用中，这些配置应来自安全的地方，而不是硬编码
    private let meiliHost = "http://localhost:7700"
    private let meiliApiKey = "aMasterKey"
    private let meiliIndex = "products"

    init() {
        // 使用Combine来处理搜索节流，避免用户每输入一个字符就发起一次请求
        searchCancellable = $searchText
            .debounce(for: .milliseconds(300), scheduler: RunLoop.main) // 300ms防抖
            .removeDuplicates()
            .filter { !$0.isEmpty }
            .handleEvents(receiveOutput: { _ in self.isLoading = true })
            .flatMap { query -> AnyPublisher<SearchResult, Never> in
                self.performSearch(query: query)
            }
            .map { $0.hits }
            .handleEvents(receiveOutput: { _ in self.isLoading = false })
            .receive(on: DispatchQueue.main)
            .assign(to: \.searchResults, on: self)
    }
    
    private func performSearch(query: String) -> AnyPublisher<SearchResult, Never> {
        guard let url = URL(string: "\(meiliHost)/indexes/\(meiliIndex)/search") else {
            return Just(SearchResult(hits: [], query: "", processingTimeMs: 0)).eraseToAnyPublisher()
        }
        
        var request = URLRequest(url: url)
        request.httpMethod = "POST"
        request.addValue("application/json", forHTTPHeaderField: "Content-Type")
        request.addValue("Bearer \(meiliApiKey)", forHTTPHeaderField: "Authorization")
        
        let body: [String: Any] = ["q": query, "limit": 20]
        request.httpBody = try? JSONSerialization.data(withJSONObject: body)
        
        return URLSession.shared.dataTaskPublisher(for: request)
            .map { $0.data }
            .decode(type: SearchResult.self, decoder: JSONDecoder())
            .replaceError(with: SearchResult(hits: [], query: query, processingTimeMs: 0))
            .eraseToAnyPublisher()
    }
}

// 在SwiftUI View中使用
/*
 struct ContentView: View {
     @StateObject private var viewModel = SearchViewModel()
 
     var body: some View {
         NavigationView {
             VStack {
                 SearchBar(text: $viewModel.searchText)
                 if viewModel.isLoading {
                     ProgressView()
                 }
                 List(viewModel.searchResults) { product in
                     VStack(alignment: .leading) {
                         Text(product.name).font(.headline)
                         Text(product.description).font(.subheadline).foregroundColor(.gray)
                     }
                 }
             }
             .navigationTitle("产品搜索")
         }
     }
 }
*/

这个Swift代码片段展示了几个生产级的实践：

Combine与Debounce: 防止对API的过度请求，提升用户体验和后端性能。
URLSession DataTaskPublisher: 使用Swift的现代化并发框架来处理网络请求。
安全提醒: 直接在客户端代码中包含API Key是极度危险的。在生产环境中，iOS客户端应该与一个受保护的后端API（API Gateway）通信，该API再将请求代理到Meilisearch，并在此过程中处理认证和授权。

局限性与未来迭代路径

当前这个方案已经解决了实时索引的核心问题，但在投入生产前，还有几个方面需要深化：

Schema演进: 如果products表的结构发生变化（例如，增加一个字段），当前的Python消费者可能会出错或忽略新字段。一个更完善的方案是集成Schema Registry（如Confluent Schema Registry），让消费者能够动态地适应上游的schema变更。
消费者伸缩性与容错: 当前服务是单实例运行。在数据变更非常频繁的场景下，单个消费者可能成为瓶颈。利用Kafka的消费者组特性，我们可以水平扩展多个消费者实例来并行处理分区。但这要求我们的处理逻辑是无状态的或者能够正确处理分区重平衡。
快照（Snapshot）处理: Debezium首次启动时会对表进行一次全量快照。这可能产生数百万条消息。我们的批处理逻辑需要能够优雅地处理这种突发的大流量，可能需要动态调整批次大小，或者在快照期间有不同的处理策略。
监控与告警: 整个管道的健康状况需要被严密监控。关键指标包括：Kafka消费者延迟（Consumer Lag）、Meilisearch的API响应时间和错误率、Python服务的CPU和内存使用率。这些指标应接入Prometheus等监控系统，并设置告警。

这个架构虽然组件多，配置复杂，但它提供了一个高度解耦、可扩展且具有弹性的实时搜索解决方案。它将数据库的写入性能与搜索索引的更新过程彻底分离，为iOS端提供了真正意义上的“即时”搜索体验。

Python iOS 开发 Meilisearch Kafka Debezium CDC

在Kubernetes上利用Cilium L7策略为Azure Functions与OpenCV工作负载构建零信任网络模型

2023-10-27 云原生

Kubernetes OpenCV Azure Functions Cilium eBPF KEDA

使用 Tyk Python 插件、NumPy 与 Argo CD 构建 API 流量异常检测系统

2023-10-27 API 网关

可观测性 NumPy Argo CD Tyk Cypress GitOps