jkes
v1.1.0
Jkes는 Java, Kafka 및 ElasticSearch를 기반으로 하는 검색 프레임워크입니다. Jkes는 문서 검색을 위해 REST API를 사용하여 주석 기반 JPA 스타일 개체/문서 매핑을 제공합니다.
jkes-integration-test 프로젝트를 참조하면 jkes 프레임워크 사용 방법을 빠르게 배울 수 있습니다. jkes-integration-test 기능적 무결성을 테스트하는 데 사용하는 Spring Boot 애플리케이션입니다.
jkes-index-connector 및 jkes-delete-connector 설치합니다.sudo bin/elasticsearch-plugin install analysis-smartcn @ EnableAspectJAutoProxy
@ EnableJkes
@ Configuration
public class JkesConfig {
@ Bean
public PlatformTransactionManager transactionManager ( EntityManagerFactory factory , EventSupport eventSupport ) {
return new SearchPlatformTransactionManager ( new JpaTransactionManager ( factory ), eventSupport );
}
} @ 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 사용하여 구성을 제공할 수 있습니다.
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 ();
}
} com.timeyang.jkes.core.annotation 패키지 아래의 관련 주석을 사용하여 엔터티를 표시합니다.
@ 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 ;
}
}엔터티가 업데이트되면 문서는 자동으로 ElasticSearch에 인덱싱되고, 엔터티가 삭제되면 해당 문서는 ElasticSearch에서 자동으로 삭제됩니다.
검색 서비스 jkes-search-service를 시작합니다. 검색 서비스는 나머지 검색 API를 제공하고 기본적으로 포트 9000에서 실행되는 Spring Boot 애플리케이션입니다.
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"
}
}
}
색인 생성 작동 방식:
@Document 로 주석이 달린 모든 엔터티를 검색하고 이에 대한 메타데이터를 구축합니다.index 및 mapping 구성을 생성한 후 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 다시 시작합니다.쿼리 작동 방식:
json 요청을 구문 분석하고 일부 사전 처리를 수행한 후 ElasticSearch Java Rest Client 사용하여 이를 ElasticSearch에 전달하고 응답을 구문 분석한 후 추가 처리 후 클라이언트에 반환합니다. jkes-core 전체 jkes 의 핵심 부분입니다. 주로 다음 기능을 포함합니다:
annotation 패키지는 jkes의 핵심 주석을 제공합니다.elasticsearch 패키지는 모든 문서에 대한 인덱스 생성/업데이트 및 매핑 업데이트와 같은 elasticsearch 관련 작업을 캡슐화합니다.kafka 패키지는 Kafka Producer, Kafka Json Serializer, 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-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-delete-connector
jkes-delete-connector kafka 클러스터에서 인덱스 삭제 이벤트( DeleteEvent )를 가져온 다음 Jest Client 사용하여 ElasticSearch에서 해당 문서를 삭제하는 데 사용되는 Kafka Connector 입니다.
Kafka Connect의 Rest Admin API를 사용하여 멀티 테넌트 플랫폼에서 문서 삭제 기능을 쉽게 구현합니다. 각 프로젝트에 대해 jkes-delete-connector 시작하기만 하면 해당 프로젝트에 대한 문서 삭제가 자동으로 처리됩니다. 이렇게 하면 새 프로젝트를 시작할 때마다 프로젝트의 문서 삭제를 처리하기 위해 Kafka 소비자를 수동으로 시작해야 하는 일이 방지됩니다. 이런 종류의 작업은 정기 구독을 통해 줄일 수 있지만 여전히 매우 융통성이 없습니다.
jkes-search-service
jkes-search-service 는 다양한 버전의 Rest 쿼리 API를 제공하는 편안한 검색 서비스입니다. 쿼리 서비스는 API 발전 및 호환성을 위해 다중 버전 API를 제공합니다.jkes-search-service 현재 URI 스타일 검색과 JSON 요청 본문 스타일 검색을 지원합니다.json 요청을 구문 분석하고 일부 사전 처리를 수행한 후 ElasticSearch Java Rest Client 사용하여 이를 ElasticSearch에 전달하고 응답을 구문 분석한 후 추가 처리 후 클라이언트에 반환합니다. 향후에는 zookeeper 기반의 인덱스 클러스터를 구축하여 클러스터 인덱스 관리 기능을 제공할 예정입니다.
jkes-integration-test 는功能测试위한 Spring Boot 통합 테스트 프로젝트입니다. 또한 일부 일반적인 작업의吞吐率측정합니다.
개발 버전을 빌드하려면 최신 버전의 Kafka가 필요합니다. 표준 수명 주기 단계를 사용하여 Maven으로 jke를 빌드할 수 있습니다.
이 프로젝트는 Apache License 2.0에 따라 라이센스가 부여됩니다.
