Download jkes - download do código-fonte jkes

jkes

Outro código-fonte

v1.1.0

Baixar

Brincadeiras

Jkes é uma estrutura de pesquisa baseada em Java, Kafka e ElasticSearch. Jkes fornece mapeamento de objeto/documento no estilo JPA baseado em anotações, usando a API REST para pesquisa de documentos.

Instalar

Você pode consultar o projeto jkes-integration-test para aprender rapidamente como usar a estrutura jkes. jkes-integration-test é um aplicativo Spring Boot que usamos para testar a integridade funcional.

Instale jkes-index-connector e jkes-delete-connector no classpath do Kafka Connect
Instale o plug-in de análise inteligente chinesa

sudo bin/elasticsearch-plugin install analysis-smartcn

Configuração

Introduzir a dependência jkes-spring-data-jpa
Adicionar configuração

 @ EnableAspectJAutoProxy
@ EnableJkes
@ Configuration
public class JkesConfig {

  @ Bean
  public PlatformTransactionManager transactionManager ( EntityManagerFactory factory , EventSupport eventSupport ) {

    return new SearchPlatformTransactionManager ( new JpaTransactionManager ( factory ), eventSupport );
  }
}

Fornecer feijão JkesProperties

 @ 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" ;
    }

}

Isso pode ser muito flexível. Se você usar Spring Boot, poderá usar @ConfigurationProperties para fornecer configuração.

Incluindo o terminal de gerenciamento de índice Como não sabemos qual tecnologia da web o cliente usa, o terminal de índice precisa ser adicionado no lado do cliente. Por exemplo, no Spring MVC , você pode adicionar o endpoint do índice da seguinte maneira

 @ 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 ();
    }

}

início rápido

API de índice

Use as anotações relevantes no pacote com.timeyang.jkes.core.annotation para marcar entidades

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

    // @Id will be identified automatically
    // @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 ;

    // don't add @Field to test whether ignored
    // @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 ;
    }
}

Quando uma entidade é atualizada, o documento é automaticamente indexado no ElasticSearch; quando uma entidade é excluída, o documento é automaticamente excluído do ElasticSearch;

API de pesquisa

Inicie o serviço de pesquisa jkes-search-service O serviço de pesquisa é um aplicativo Spring Boot que fornece API de pesquisa restante e é executado na porta 9000 por padrão.

Consulta URI

 curl -XPOST localhost:9000/api/v1/integration_test_person_group/person_group/_search?from=3&size=10

Consulta aninhada

 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
                }
              }
            }
          ]
        }
      }
    }
  }
}

consulta de correspondência

 integration_test_person_group/person_group/_search?from=0&size=10
{
  "query": {
      "match": {
        "interests": "Hadoop"
      }
    }
}

consulta booleana

 {
  "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
    }

  }

}

Filtragem de origem

 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" }
    }
}

prefixo

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

curinga

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

expressão regular

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

Como funciona o Jkes

Como funciona a indexação:

Quando o aplicativo é iniciado, o Jkes verifica todas as entidades anotadas com @Document e cria metadados para elas.
Com base nos metadados construídos, crie configurações index e mapping no formato Json e, em seguida, crie/atualize a configuração index por meio do ElasticSearch Java Rest Client .
Criar/atualizar Kafka ElasticSearch Connector para cada documento, usado para criar/atualizar documentos
Iniciar/atualizar Jkes Deleter Connector para todo o projeto, para exclusão de documentos
Método de operação de dados de interceptação. Empacote os dados retornados pelo método * save(*) como SaveEvent e salve-os no EventContainer ; use os parâmetros do método (* delete*(..) para gerar um DeleteEvent/DeleteAllEvent e salve-os no EventContainer .
Interceptar transações. Após a transação ser confirmada, use JkesKafkaProducer para enviar as entidades SaveEvent para Kafka, usará JkesJsonSerializer que fornecemos para serializar os dados especificados e, em seguida, enviá-los para Kafka.
Ao contrário de SaveEvent , DeleteEvent será serializado diretamente e enviado ao Kafka em vez de apenas enviar uma cópia dos dados.
Ao contrário de SaveEvent e DeleteEvent , DeleteAllEvent não envia dados para Kafka, mas exclui diretamente o index correspondente por meio ElasticSearch Java Rest Client , depois reconstrói o índice e reinicia Kafka ElasticSearch Connector

Como funciona a consulta:

O serviço de consulta é fornecido por meio da API REST
Não usamos o ElasticSearch diretamente para consulta porque precisamos usar o aprendizado de máquina para classificação de pesquisa nas versões subsequentes. O acoplamento direto com ElasticSearch aumentará a dificuldade de acesso à API de classificação de pesquisa.
O serviço de consulta é um aplicativo Spring Boot, empacotado como um espelho usando docker
O serviço de consulta fornece API multiversão para evolução e compatibilidade de API
O serviço de consulta analisa a solicitação json , executa algum pré-processamento e a encaminha para ElasticSearch usando ElasticSearch Java Rest Client , analisa a resposta e a retorna ao cliente após processamento adicional.
Para facilitar o desenvolvimento do cliente, o serviço de consulta fornece uma interface de interface de usuário de consulta. Os desenvolvedores podem copiar o corpo da solicitação json para o programa após obter os resultados esperados nesta página.

fluxograma

Introdução ao módulo

jkes-core

jkes-core é a parte central de todo o jkes . Inclui principalmente as seguintes funções:

O pacote annotation fornece as anotações principais do jkes
O pacote elasticsearch encapsula operações relacionadas elasticsearch , como criação/atualização de índices para todos os documentos e atualização de mapeamento
O pacote kafka fornece Kafka Producer, Kafka Json Serializer, Kafka Connect Client
O pacote metadata fornece construção de metadados de anotação básica e modelos estruturados.
O pacote event fornece modelos de eventos e contêineres
O pacote exception fornece exceções Jkes comuns
O pacote http encapsula solicitações http json comuns baseadas no Apache Http Client
O pacote support expõe o suporte à configuração principal do Jkes
O pacote util fornece algumas classes de ferramentas para facilitar o desenvolvimento. Tais como: Afirmações, ClassUtils, DocumentUtils, IOUtils, JsonUtils, ReflectionUtils, StringUtils

jkes-boot

jkes-boot é usado para integração com algumas estruturas de código aberto de terceiros.

Atualmente, fornecemos integração com spring data jpa por meio de jkes-spring-data-jpa . Usando o mecanismo AOP do Spring, o método Repository é interceptado e SaveEvent/DeleteEvent/DeleteAllEvent é gerado e salvo em EventContainer . Ao usar SearchPlatformTransactionManager que fornecemos, agrupamos gerenciadores de transações comumente usados (como JpaTransactionManager ) para fornecer funções de interceptação de transações.

Nas versões subsequentes, forneceremos integração com mais frameworks.

Descrição de jkes-spring-data-jpa :

A classe ContextSupport é usada para obter Repository Bean da fábrica de beans
@EnableJkes permite que os clientes habilitem facilmente a funcionalidade Jkes e fornece um modelo de configuração consistente com Spring
EventSupport lida com os detalhes dos eventos, gera eventos correspondentes e os armazena em EventContainer ao salvar e excluir dados e processa eventos correspondentes quando as transações são confirmadas e revertidas.
SearchPlatformTransactionManager envolve o gerenciador de transações do cliente e adiciona回调hook quando as transações são confirmadas e revertidas.
audit fornece uma classe pai AuditedEntity simples para facilitar a adição de funções de auditoria. As informações de versão podem ser usadas em conjunto com o mecanismo de versão do ElasticSearch para garantir que os dados do documento expirado não serão indexados.
O pacote exception encapsula exceções comuns
O pacote intercept fornece pontos e aspectos AOP
O pacote index fornece funcionalidade全量索引. Atualmente, fornecemos um mecanismo de indexação baseado em线程池e um mecanismo de indexação baseado em ForkJoin . Nas versões subsequentes, refatoraremos o código e adicionaremos um modelo生产者-消费者baseado no阻塞队列para fornecer desempenho de simultaneidade.

serviços jkes

jkes-services é usado principalmente para fornecer alguns serviços. Atualmente, jkes-services fornece os seguintes serviços:

jkes-delete-connector
- jkes-delete-connector é um Kafka Connector usado para obter eventos de exclusão de índice ( DeleteEvent ) do cluster kafka e, em seguida, usar Jest Client para excluir os documentos correspondentes no ElasticSearch.
- Com a ajuda da API rest admin do Kafka Connect, implementamos facilmente a função de exclusão de documentos em uma plataforma multilocatário. Basta iniciar um jkes-delete-connector para cada projeto e a exclusão de documentos desse projeto será tratada automaticamente. Isso evita que toda vez que iniciarmos um novo projeto, tenhamos que iniciar manualmente um Kafka Consumer para tratar da exclusão do documento do projeto. Embora este tipo de trabalho possa ser reduzido através de assinaturas regulares, ainda é muito inflexível.
jkes-search-service
- jkes-search-service é um serviço de pesquisa tranquilo que fornece várias versões da API de consulta restante. O serviço de consulta fornece API multiversão para evolução e compatibilidade de API
- jkes-search-service atualmente suporta pesquisa de estilo URI e pesquisa de estilo de corpo de solicitação JSON.
- Não usamos ElasticSearch diretamente para consulta, porque precisamos usar aprendizado de máquina para classificação de pesquisa nas versões subsequentes, e o acoplamento direto com ElasticSearch aumentará a dificuldade de acesso para classificação de pesquisa.
- O serviço de consulta é um aplicativo Spring Boot, empacotado como um espelho usando docker
- O serviço de consulta analisa a solicitação json , executa algum pré-processamento e a encaminha para ElasticSearch usando ElasticSearch Java Rest Client , analisa a resposta e a retorna ao cliente após processamento adicional.
- Para facilitar o desenvolvimento do cliente, o serviço de consulta fornece uma interface de interface de usuário de consulta. Os desenvolvedores podem copiar o corpo da solicitação json para o programa após obter os resultados esperados nesta página.

No futuro, construiremos um cluster de índice baseado no zookeeper para fornecer funções de gerenciamento de índice de cluster.

teste de integração jkes

jkes-integration-test é um projeto de teste de integração do Spring Boot para功能测试. Meça também吞吐率de algumas operações comuns

Desenvolvimento

Para construir uma versão de desenvolvimento, você precisará de uma versão recente do Kafka. Você pode construir jkes com Maven usando as fases padrão do ciclo de vida.