andytruong/es-writer

Elastic Search Writer 镜像文档

镜像概述和主要用途

当多个服务直接向同一Elasticsearch服务器写入数据时，极易产生冲突。本镜像作为单一写入者，通过消费RabbitMQ消息实现对Elasticsearch的统一写入，有效避免数据冲突。服务无需知晓Elasticsearch服务器凭证，同时便于构建Elasticsearch集群。

核心功能和特性

• 冲突避免：作为单一写入者，消除多服务并发写入Elasticsearch的冲突问题
• 权限隔离：服务无需直接访问Elasticsearch，无需存储其凭证信息
• 消息驱动：基于RabbitMQ消息队列实现异步写入，解耦服务与Elasticsearch
• 灵活配置：支持RabbitMQ连接参数、Elasticsearch地址、消费模式等多维度配置
• 操作支持：处理索引创建、删除、文档索引、更新、删除等多种Elasticsearch操作
• 集群友好：简化Elasticsearch集群的部署与管理

使用场景和适用范围

适用于多服务需要向同一Elasticsearch服务器写入数据的场景，尤其适合：

• 微服务架构中多个服务需操作同一Elasticsearch实例
• 需统一管理Elasticsearch写入权限和流量的系统
• 要求避免并发写入冲突、保证数据一致性的应用

使用方法和配置说明

消息队列（Queueing）

服务需通过RabbitMQ发布消息至ES-Writer，消息需满足以下要求：

• 交换机（Exchange）：$exchange（自定义交换机名称）
• 路由键（Key）：$routingKey（自定义路由键）
• 消息体（Payload）：JSON对象，支持以下操作类型（示例路径）：
  • fixtures/indices/indices-create.json（创建索引）
  • fixtures/indices/indices-drop.json（删除索引）
  • fixtures/portal/portal-index.json（文档索引）
  • fixtures/portal/portal-update.json（文档更新）
  • fixtures/portal/portal-update-by-query.json（按查询更新）
  • fixtures/portal/portal-delete.json（文档删除）
  • fixtures/portal/portal-delete-by-query.json（按查询删除）

参考文档：https://www.elastic.co/guide/en/elasticsearch/reference/5.2/docs-bulk.html

消息消费（Consuming）

ES-Writer启动参数及环境变量配置：

/path/to/es-writer
    -url             RABBITMQ_URL<STRING=amqp://go1:go1@127.0.0.1:5672/>  # RabbitMQ连接URL
    -kind            RABBITMQ_KIND<STRING=topic>  # RabbitMQ交换机类型
    -exchange        RABBITMQ_EXCHANGE<STRING=events>  # 消息交换机名称
    -routing-key     RABBITMQ_ROUTING_KEY<STRING=es.writer.go1>  # 消息路由键
    -consumer-name   RABBITMQ_CONSUMER_NAME<STRING=es-writer>  # 消费者名称
    -queue-name      RABBITMQ_QUEUE_NAME<STRING=es-writer>  # 队列名称
    -url-contains     URL_CONTAINS<STRING>     # URL包含字符串过滤（示例：`award:`）
    -url-not-contains URL_NOT_CONTAINS<STRING> # URL不包含字符串过滤（示例：`award:`）
    -prefetch-count  RABBITMQ_PREFETCH_COUNT<INT=50>  # 预取消息数量
    -es-url          ELASTIC_SEARCH_URL<STRING=http://127.0.0.1:9200/?sniff=false>  # Elasticsearch连接URL
    -refresh         true # 可选，允许值：true、wait_for，默认true（索引刷新策略）
    -admin-port      ADMIN_PORT<STRING=:8001>  # 管理端口
    -debug           DEBUG<BOOL=true>  # 调试模式，默认false
    -single-active-consumer SINGLE_ACTIVE_CONSUMER<BOOL=true>  # 单活跃消费者模式，默认false（参考RabbitMQ文档）
    -bulk-timeout BULK_TIMEOUT=5m  # 批量操作超时时间

注意事项

以下请求类型不支持批量处理，需限制使用：

• /\_update_by_query（按查询更新）
• /\_delete_by_query（按查询删除）
• PUT /index_name（创建索引）
• DELETE /index_name（删除索引）

本地测试

bash
# 启动Elasticsearch
docker run -d -p 9200:9200 --rm --name es docker.xuanyuan.run/elasticsearch:5-alpine

# 启动RabbitMQ（带管理界面）
docker run -d -p 5672:5672 -p ***:*** --rm --name rabbitmq docker.xuanyuan.run/rabbitmq:3-management

# 运行测试
go test -race -v ./...