Logo
Project Icon

jkes

将Java、Kafka和ElasticSearch集成的全功能搜索解决方案

Jkes搜索框架JavaKafkaElasticSearchGithub开源项目

Jkes是一个整合Java、Kafka和ElasticSearch技术的搜索框架。它采用注解驱动的JPA风格对象/文档映射，并提供RESTful API进行文档搜索。该框架支持自动索引更新和删除，同时提供灵活的查询接口。Jkes易于集成，可扩展性强，适合需要高性能搜索功能的Java应用。其架构支持多租户平台，为后续引入机器学习搜索排序功能做好了准备。

访问官网访问官网GithubGithub文档文档
介绍相关项目

Jkes

Jkes是一个基于Java、Kafka、ElasticSearch的搜索框架。Jkes提供了注解驱动的JPA风格的对象/文档映射，使用REST API进行文档搜索。

安装

可以参考jkes-integration-test项目快速掌握Jkes框架的使用方法。jkes-integration-test是我们用来测试功能完整性的一个Spring Boot应用。

  • 将jkes-index-connector和jkes-delete-connector安装到Kafka Connect的类路径
  • 安装Smart Chinese Analysis插件
sudo bin/elasticsearch-plugin install analysis-smartcn

配置

  • 引入jkes-spring-data-jpa依赖
  • 添加配置
@EnableAspectJAutoProxy
@EnableJkes
@Configuration
public class JkesConfig {

  @Bean
  public PlatformTransactionManager transactionManager(EntityManagerFactory factory, EventSupport eventSupport) {
    return new SearchPlatformTransactionManager(new JpaTransactionManager(factory), eventSupport);
  }
}
  • 提供JkesProperties Bean
@Component
@Configuration
public class JkesConf extends DefaultJkesPropertiesImpl {

    @PostConstruct
    public void setUp() {
        Config.setJkesProperties(this);
    }

    @Override
    public String getKafkaBootstrapServers() {
        return "k1-test.com:9292,k2-test.com:9292,k3-test.com:9292";
    }

    @Override
    public String getKafkaConnectServers() {
        return "http://k1-test.com:8084,http://k2-test.com:8084,http://k3-test.com:8084";
    }

    @Override
    public String getEsBootstrapServers() {
        return "http://es1-test.com:9200,http://es2-test.com:9200,http://es3-test.com:9200";
    }

    @Override
    public String getDocumentBasePackage() {
        return "com.timeyang.jkes.integration_test.domain";
    }

    @Override
    public String getClientId() {
        return "integration_test";
    }

}

这里可以很灵活，如果使用Spring Boot，可以使用@ConfigurationProperties提供配置

  • 增加索引管理端点 因为我们不知道客户端使用的是哪种Web技术，所以索引端点需要在客户端添加。例如在Spring MVC中，可以按照如下方式添加索引端点：
@RestController
@RequestMapping("/api/search")
public class SearchEndpoint {

    private Indexer indexer;

    @Autowired
    public SearchEndpoint(Indexer indexer) {
        this.indexer = indexer;
    }

    @RequestMapping(value = "/start_all", method = RequestMethod.POST)
    public void startAll() {
        indexer.startAll();
    }

    @RequestMapping(value = "/start/{entityClassName:.+}", method = RequestMethod.POST)
    public void start(@PathVariable("entityClassName") String entityClassName) {
        indexer.start(entityClassName);
    }

    @RequestMapping(value = "/stop_all", method = RequestMethod.PUT)
    public Map<String, Boolean> stopAll() {
        return indexer.stopAll();
    }

    @RequestMapping(value = "/stop/{entityClassName:.+}", method = RequestMethod.PUT)
    public Boolean stop(@PathVariable("entityClassName") String entityClassName) {
        return indexer.stop(entityClassName);
    }

    @RequestMapping(value = "/progress", method = RequestMethod.GET)
    public Map<String, IndexProgress> getProgress() {
        return indexer.getProgress();
    }

}

快速开始

索引API

使用com.timeyang.jkes.core.annotation包下相关注解标记实体

@lombok.Data
@Entity
@Document
public class Person extends AuditedEntity {

    // @Id将自动被识别
    // @Field(type = FieldType.Long)
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @MultiFields(
            mainField = @Field(type = FieldType.Text),
            otherFields = {
                    @InnerField(suffix = "raw", type = FieldType.Keyword),
                    @InnerField(suffix = "english", type = FieldType.Text, analyzer = "english")
            }
    )
    private String name;

    @Field(type = FieldType.Keyword)
    private String gender;

    @Field(type = FieldType.Integer)
    private Integer age;

    // 不添加@Field注解来测试是否被忽略
    // @Field(type = FieldType.Text)
    private String description;

    @Field(type = FieldType.Object)
    @ManyToOne(fetch = FetchType.EAGER)
    @JoinColumn(name = "group_id")
    private PersonGroup personGroup;

}

@lombok.Data
@Entity
@Document(type = "person_group", alias = "person_group_alias")
public class PersonGroup extends AuditedEntity {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String name;
    private String interests;
    @OneToMany(fetch = FetchType.EAGER, cascade = CascadeType.ALL, mappedBy = "personGroup", orphanRemoval = true)
    private List<Person> persons;
    private String description;

    @DocumentId
    @Field(type = FieldType.Long)
    public Long getId() {
        return id;
    }

    @MultiFields(
            mainField = @Field(type = FieldType.Text),
            otherFields = {
                    @InnerField(suffix = "raw", type = FieldType.Keyword),
                    @InnerField(suffix = "english", type = FieldType.Text, analyzer = "english")
            }
    )
    public String getName() {
        return name;
    }

    @Field(type = FieldType.Text)
    public String getInterests() {
        return interests;
    }

    @Field(type = FieldType.Nested)
    public List<Person> getPersons() {
        return persons;
    }

    /**
     * 不加Field注解，测试序列化时是否忽略
     */
    public String getDescription() {
        return description;
    }
}

当更新实体时，文档会被自动索引到ElasticSearch；删除实体时，文档会自动从ElasticSearch中删除。

搜索API

启动搜索服务jkes-search-service，搜索服务是一个Spring Boot应用，提供REST搜索API，默认运行在9000端口。

  • URI查询
curl -XPOST localhost:9000/api/v1/integration_test_person_group/person_group/_search?from=3&size=10
  • 嵌套查询
integration_test_person_group/person_group/_search?from=0&size=10
{
  "query": {
    "nested": {
      "path": "persons",
      "score_mode": "avg",
      "query": {
        "bool": {
          "must": [
            {
              "range": {
                "persons.age": {
                  "gt": 5
                }
              }
            }
          ]
        }
      }
    }
  }
}
  • 匹配查询
integration_test_person_group/person_group/_search?from=0&size=10
{
  "query": {
      "match": {
        "interests": "Hadoop"
      }
    }
}
  • 布尔查询
{
  "query": {
    "bool" : {
      "must" : {
        "match" : { "interests" : "Hadoop" }
      },
      "filter": {
        "term" : { "name.raw" : "name0" }
      },
      "should" : [
        { "match" : { "interests" : "Flink" } },
        {
            "nested" : {
                "path" : "persons",
                "score_mode" : "avg",

"query" : { "bool" : { "must" : [ { "match" : {"persons.name" : "name40"} }, { "match" : {"persons.interests" : "interests"} } ], "must_not" : { "range" : { "age" : { "gte" : 50, "lte" : 60 } } } } } }

], "minimum_should_match" : 1, "boost" : 1.0 }

}

}

- 源过滤

integration_test_person_group/person_group/_search { "_source": false, "query" : { "match" : { "name" : "name17" } } }

integration_test_person_group/person_group/_search { "_source": { "includes": [ "name", "persons." ], "excludes": [ "date", "version", "persons.age" ] }, "query" : { "match" : { "name" : "name17" } } }

- 前缀

integration_test_person_group/person_group/_search { "query": { "prefix" : { "name" : "name" } } }

- 通配符

integration_test_person_group/person_group/_search { "query": { "wildcard" : { "name" : "name*" } } }

- 正则表达式

integration_test_person_group/person_group/_search { "query": { "regexp":{ "name": "na.*17" } } }


## Jkes工作原理
索引工作原理：
- 应用启动时，Jkes扫描所有标注`@Document`注解的实体，为它们构建元数据。
- 基于构建的元数据，创建`index`和`mapping` Json格式的配置，然后通过`ElasticSearch Java Rest Client`创建/更新`index`配置。
- 为每个文档创建/更新`Kafka ElasticSearch Connector`，用于创建/更新文档
- 为整个项目启动/更新`Jkes Deleter Connector`，用于删除文档
- 拦截数据操作方法。将`* save(*)`方法返回的数据包装为`SaveEvent`保存到`EventContainer`；使用`(* delete*(..)`方法的参数，生成一个`DeleteEvent/DeleteAllEvent`保存到`EventContainer`。
- 拦截事务。在事务提交后使用`JkesKafkaProducer`发送`SaveEvent`中的实体到Kafka，Kafka会使用我们提供的`JkesJsonSerializer`序列化指定的数据，然后发送到Kafka。
- 与`SaveEvent`不同，`DeleteEvent`会直接被序列化，然后发送到Kafka，而不是只发送一份数据
- 与`SaveEvent`和`DeleteEvent`不同，`DeleteAllEvent`不会发送数据到Kafka，而是直接通过`ElasticSearch Java Rest Client`删除相应的`index`，然后重建该索引，重启`Kafka ElasticSearch Connector`

查询工作原理：
- 查询服务通过rest api提供
- 我们没有直接使用ElasticSearch进行查询，因为我们需要在后续版本使用机器学习进行搜索排序，而直接与ElasticSearch进行耦合，会增加搜索排序API的接入难度
- 查询服务是一个Spring Boot应用，使用docker打包为镜像
- 查询服务提供多版本API，用于API演进和兼容
- 查询服务解析`json`请求，进行一些预处理后，使用`ElasticSearch Java Rest Client`转发到ElasticSearch，将得到的响应进行解析，进一步处理后返回到客户端。
- 为了便于客户端人员开发，查询服务提供了一个[查询UI界面](http://localhost:9000/api/v1)，开发人员可以在这个页面得到预期结果后再把json请求体复制到程序中。

## 流程图
![Jkes流程图](https://yellow-cdn.veclightyear.com/0a4dffa0/da9ec837-8810-46ee-929c-8373f9843373.png)

## 模块介绍
### jkes-core
`jkes-core`是整个`jkes`的核心部分。主要包括以下功能：
- `annotation`包提供了jkes的核心注解
- `elasticsearch`包封装了`elasticsearch`相关的操作，如为所有的文档创建/更新索引，更新mapping
- `kafka`包提供了Kafka生产者，Kafka Json序列化器，Kafka Connect客户端
- `metadata`包提供了核心的注解元数据的构建与结构化模型
- `event`包提供了事件模型与容器
- `exception`包提供了常见的Jkes异常
- `http`包基于`Apache Http Client`封装了常见的http json请求
- `support`包暴露了Jkes核心配置支持
- `util`包提供了一些工具类，便于开发。如：Asserts, ClassUtils, DocumentUtils, IOUtils, JsonUtils, ReflectionUtils, StringUtils

### jkes-boot
`jkes-boot`用于与一些第三方开源框架进行集成。

目前，我们通过`jkes-spring-data-jpa`，提供了与`spring data jpa`的集成。通过使用Spring的AOP机制，对`Repository`方法进行拦截，生成`SaveEvent/DeleteEvent/DeleteAllEvent`保存到`EventContainer`。通过使用我们提供的`SearchPlatformTransactionManager`，对常用的事务管理器(如`JpaTransactionManager`)进行包装，提供事务拦截功能。

在后续版本，我们会提供与更多框架的集成。

`jkes-spring-data-jpa`说明：
- `ContextSupport`类用于从bean工厂获取`Repository Bean`
- `@EnableJkes`让客户端能够轻松开启Jkes的功能，提供了与Spring一致的配置模型
- `EventSupport`处理事件的细节，在保存和删除数据时生成相应事件存放到`EventContainer`，在事务提交和回滚时处理相应的事件
- `SearchPlatformTransactionManager`包装了客户端的事务管理器，在事务提交和回滚时加入了`回调hook`
- `audit`包提供了一个简单的`AuditedEntity`父类，方便添加审计功能，版本信息可用于结合`ElasticSearch`的版本机制保证不会索引过期文档数据
- `exception`包封装了常见异常
- `intercept`包提供了AOP切点和切面
- `index`包提供了`全量索引`功能。当前，我们提供了基于`线程池`的索引机制和基于`ForkJoin`的索引机制。在后续版本，我们会重构代码，增加基于`阻塞队列`的`生产者-消费者`模式，提供并发性能
### jkes-services
`jkes-services`主要用于提供一些服务。
目前，`jkes-services`提供了以下服务：
- `jkes-delete-connector`
    
  - `jkes-delete-connector`是一个`Kafka Connector`，用于从Kafka集群获取索引删除事件（`DeleteEvent`），然后使用`Jest Client`删除ElasticSearch中相应的文档。
   
  - 借助Kafka Connect的REST管理API，我们轻松地实现了多租户平台上的文档删除功能。只需为每个项目启动一个`jkes-delete-connector`，就可以自动处理该项目的文档删除工作。这避免了每次启动新项目时，都需要手动启动一个Kafka Consumer来处理该项目的文档删除工作。虽然可以通过正则订阅来减少这种工作，但仍然非常不灵活。
    
- `jkes-search-service`
  
  - `jkes-search-service`是一个RESTful搜索服务，提供了多个版本的REST查询API。查询服务提供多版本API，用于API演进和兼容。
  - `jkes-search-service`目前支持URI风格的搜索和JSON请求体风格的搜索。
  - 我们没有直接使用ElasticSearch进行查询，因为我们计划在后续版本中使用机器学习进行搜索排序，而直接与ElasticSearch耦合会增加搜索排序的接入难度。
  - 查询服务是一个Spring Boot应用程序，使用Docker打包为镜像。
  - 查询服务解析`json`请求，进行一些预处理后，使用`ElasticSearch Java REST Client`转发到ElasticSearch，将得到的响应进行解析，进一步处理后返回给客户端。
  - 为了方便客户端开发人员，查询服务提供了一个[查询UI界面](http://localhost:9000/api/v1)，开发人员可以在这个页面获得预期结果后再将JSON请求体复制到程序中。

后续，我们将基于`ZooKeeper`构建索引集群，提供集群索引管理功能。

### jkes-integration-test
`jkes-integration-test`是一个基于Spring Boot的集成测试项目，用于进行`功能测试`。同时测量一些常见操作的`吞吐率`。

## 开发
要构建开发版本，你需要使用较新版本的Kafka。你可以使用Maven的标准生命周期阶段来构建jkes。

## 贡献
- 源代码：https://github.com/chaokunyang/jkes
- 问题跟踪：https://github.com/chaokunyang/jkes/issues

## 许可证
本项目基于Apache License 2.0许可证。
相关项目
项目侧边栏1项目侧边栏2
推荐项目
Project Cover

豆包MarsCode

豆包 MarsCode 是一款革命性的编程助手，通过AI技术提供代码补全、单测生成、代码解释和智能问答等功能，支持100+编程语言，与主流编辑器无缝集成，显著提升开发效率和代码质量。

Project Cover

AI写歌

Suno AI是一个革命性的AI音乐创作平台，能在短短30秒内帮助用户创作出一首完整的歌曲。无论是寻找创作灵感还是需要快速制作音乐，Suno AI都是音乐爱好者和专业人士的理想选择。

Project Cover

有言AI

有言平台提供一站式AIGC视频创作解决方案，通过智能技术简化视频制作流程。无论是企业宣传还是个人分享，有言都能帮助用户快速、轻松地制作出专业级别的视频内容。

Project Cover

Kimi

Kimi AI助手提供多语言对话支持，能够阅读和理解用户上传的文件内容，解析网页信息，并结合搜索结果为用户提供详尽的答案。无论是日常咨询还是专业问题，Kimi都能以友好、专业的方式提供帮助。

Project Cover

阿里绘蛙

绘蛙是阿里巴巴集团推出的革命性AI电商营销平台。利用尖端人工智能技术，为商家提供一键生成商品图和营销文案的服务，显著提升内容创作效率和营销效果。适用于淘宝、天猫等电商平台，让商品第一时间被种草。

Project Cover

吐司

探索Tensor.Art平台的独特AI模型，免费访问各种图像生成与AI训练工具，从Stable Diffusion等基础模型开始，轻松实现创新图像生成。体验前沿的AI技术，推动个人和企业的创新发展。

Project Cover

SubCat字幕猫

SubCat字幕猫APP是一款创新的视频播放器，它将改变您观看视频的方式！SubCat结合了先进的人工智能技术，为您提供即时视频字幕翻译，无论是本地视频还是网络流媒体，让您轻松享受各种语言的内容。

Project Cover

美间AI

美间AI创意设计平台，利用前沿AI技术，为设计师和营销人员提供一站式设计解决方案。从智能海报到3D效果图，再到文案生成，美间让创意设计更简单、更高效。

Project Cover

AIWritePaper论文写作

AIWritePaper论文写作是一站式AI论文写作辅助工具，简化了选题、文献检索至论文撰写的整个过程。通过简单设定，平台可快速生成高质量论文大纲和全文，配合图表、参考文献等一应俱全，同时提供开题报告和答辩PPT等增值服务，保障数据安全，有效提升写作效率和论文质量。

使用协议隐私政策广告服务
投诉举报邮箱: service@vectorlightyear.com
@2024 懂AI·鲁ICP备2024100362号-6·鲁公网安备37021002001498号