版本列表 有找不到版本号的可以从此处找寻 SpringCloud Alibaba 项目参考 ------------------------------------------------------------------------------- <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> <java.version>1.8</java.version> <spring-boot.version>2.7.3</spring-boot.version> <spring-cloud.version>2021.0.4</spring-cloud.version> <spring-cloud-alibaba.version>2021.0.4.0</spring-cloud-alibaba.version> <spring-boot-admin.version>2.7.5</spring-boot-admin.version> <spring-boot.mybatis>2.2.2</spring-boot.mybatis> <swagger.fox.version>3.0.0</swagger.fox.version> <swagger.core.version>1.6.2</swagger.core.version> <tobato.version>1.27.2</tobato.version> <kaptcha.version>2.3.2</kaptcha.version> <pagehelper.boot.version>1.4.5</pagehelper.boot.version> <druid.version>1.2.12</druid.version> <dynamic-ds.version>3.5.2</dynamic-ds.version> <commons.io.version>2.11.0</commons.io.version> <commons.fileupload.version>1.4</commons.fileupload.version> <velocity.version>2.3</velocity.version> <fastjson.version>2.0.14</fastjson.version> <jjwt.version>0.9.1</jjwt.version> <minio.version>8.2.2</minio.version> <poi.version>4.1.2</poi.version> <commons-collections.version>3.2.2</commons-collections.version> <transmittable-thread-local.version>2.13.2</transmittable-thread-local.version> ------------------------------------------------------------------------------- SpringCloud 项目参考 ------------------------------------------------------------------------------- <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> <java.version>1.8</java.version> <spring-boot.version>2.3.2.RELEASE</spring-boot.version> <spring-cloud.version>Hoxton.SR7</spring-cloud.version> <spring-boot-admin.version>2.2.2</spring-boot-admin.version> <druid.version>1.1.14</druid.version> <bitwalker.version>1.19</bitwalker.version> <kaptcha.version>2.3.2</kaptcha.version> <swagger.version>2.7.0</swagger.version> <pagehelper.boot.version>1.2.12</pagehelper.boot.version> <mapper.starter.version>2.1.5</mapper.starter.version> <oshi.version>3.9.1</oshi.version> <commons.io.version>2.5</commons.io.version> <commons.fileupload.version>1.3.3</commons.fileupload.version> <jsoup.version>1.11.3</jsoup.version> <poi.version>3.17</poi.version> <velocity.version>1.7</velocity.version> <qiniu.version>[7.2.0, 7.2.99]</qiniu.version> <aliyun.oss.version>3.5.0</aliyun.oss.version> <qcloud.cos.version>5.6.3</qcloud.cos.version> <fastjson.version>1.2.70</fastjson.version> <lombok.version>1.18.4</lombok.version> <hutool.version>4.1.19</hutool.version> <common-pool.version>2.6.2</common-pool.version> <swagger.version>2.9.2</swagger.version> <swagger-ui.version>1.9.5</swagger-ui.version> <guava.version>26.0-jre</guava.version> <anji-plus.version>1.2.4</anji-plus.version> ------------------------------------------------------------------------------- SpringBoot 项目参考 ------------------------------------------------------------------------------- <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> <java.version>1.8</java.version> <maven-jar-plugin.version>3.1.1</maven-jar-plugin.version> <druid.version>1.2.11</druid.version> <bitwalker.version>1.21</bitwalker.version> <swagger.version>3.0.0</swagger.version> <kaptcha.version>2.3.2</kaptcha.version> <mybatis-spring-boot.version>2.2.2</mybatis-spring-boot.version> <pagehelper.boot.version>1.4.3</pagehelper.boot.version> <fastjson.version>2.0.14</fastjson.version> <oshi.version>6.2.2</oshi.version> <commons.io.version>2.11.0</commons.io.version> <commons.fileupload.version>1.4</commons.fileupload.version> <commons.collections.version>3.2.2</commons.collections.version> <poi.version>4.1.2</poi.version> <velocity.version>2.3</velocity.version> <jwt.version>0.9.1</jwt.version> Spring Spring框架基本的核心工具 <dependency> <groupId>org.springframework</groupId> <artifactId>spring-context-support</artifactId> </dependency> Spring-Web <dependency> <groupId>org.springframework</groupId> <artifactId>spring-web</artifactId> </dependency> Spring-webmvc <dependency> <groupId>org.springframework</groupId> <artifactId>spring-webmvc</artifactId> </dependency> Spring-context-support <dependency> <groupId>org.springframework</groupId> <artifactId>spring-context-support</artifactId> </dependency> SpringBoot SpringBoot Parent 模块 <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.2.5.RELEASE</version> </parent> SpringBoot 依赖配置 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-dependencies</artifactId> <version>2.5.14</version> <type>pom</type> <scope>import</scope> </dependency> spring security 安全认证 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-security</artifactId> </dependency> 自定义验证注解 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-validation</artifactId> </dependency> redis缓存操作 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> SpringBoot web容器 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> SpringBoot Aop 拦截器 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-aop</artifactId> </dependency> spring-boot-devtools <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-devtools</artifactId> <optional>true</optional> </dependency> SpringBoot Admin <dependency> <groupId>de.codecentric</groupId> <artifactId>spring-boot-admin-dependencies</artifactId> <version>2.2.2</version> <type>pom</type> <scope>import</scope> </dependency> 监控服务端 <dependency> <groupId>de.codecentric</groupId> <artifactId>spring-boot-admin-starter-server</artifactId> </dependency> undertow容器 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-undertow</artifactId> </dependency> Actuator 健康监控 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-actuator</artifactId> </dependency> JPA <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-jpa</artifactId> </dependency> SpringCloud 父依赖 <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-dependencies</artifactId> <version>Hoxton.SR6</version> <type>pom</type> <scope>import</scope> </dependency> Config <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-config</artifactId> </dependency> Config-Server <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-config-server</artifactId> </dependency> Eureka- Client <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId> </dependency> Eureka-Server <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-eureka-server</artifactId> </dependency> Hytrix <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-hystrix</artifactId> </dependency> Gateway <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-gateway</artifactId> </dependency> SpringCloud Alibaba 父依赖 <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-alibaba-dependencies</artifactId> <version>2021.0.4.0</version> <type>pom</type> <scope>import</scope> </dependency> Nacos discovery <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId> </dependency> Nacos Config <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId> </dependency> Sentinel <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId> </dependency> Sentinel Gateway <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-alibaba-sentinel-gateway</artifactId> </dependency> Sentinel Datasource Nacos <dependency> <groupId>com.alibaba.csp</groupId> <artifactId>sentinel-datasource-nacos</artifactId> </dependency> Loadbalancer <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-loadbalancer</artifactId> </dependency> openfeign <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-openfeign</artifactId> </dependency> Seata <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-seata</artifactId> </dependency> 数据库相关 Mysql驱动包 <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> </dependency> Druid连接池 <dependency> <groupId>com.alibaba</groupId> <artifactId>druid-spring-boot-starter</artifactId> <version>1.2.11</version> </dependency> Mybatis <dependency> <groupId>org.mybatis.spring.boot</groupId> <artifactId>mybatis-spring-boot-starter</artifactId <version>2.2.2</version> </dependency> tk-mapper <dependency> <groupId>tk.mybatis</groupId> <artifactId>mapper-spring-boot-starter</artifactId> <exclusions> <exclusion> <artifactId>persistence-api</artifactId> <groupId>javax.persistence</groupId> </exclusion> </exclusions> </dependency> pagehelper 分页插件 <dependency> <groupId>com.github.pagehelper</groupId> <artifactId>pagehelper-spring-boot-starter</artifactId> <version>1.4.3</version> </dependency> Swagger Swagger 3 依赖 <dependency> <groupId>io.springfox</groupId> <artifactId>springfox-boot-starter</artifactId> <version>3.0.0</version> <exclusions> <exclusion> <groupId>io.swagger</groupId> <artifactId>swagger-models</artifactId> </exclusion> </exclusions> </dependency> Swagger models <dependency> <groupId>io.swagger</groupId> <artifactId>swagger-models</artifactId> <version>1.6.2</version> </dependency> swagger-bootstrap-ui <dependency> <groupId>com.github.xiaoymin</groupId> <artifactId>swagger-bootstrap-ui</artifactId> <version>1.9.5</version> </dependency> swagger-annotations <dependency> <groupId>io.swagger</groupId> <artifactId>swagger-annotations</artifactId> <version>${swagger.core.version}</version> </dependency> knife4j <dependency> <groupId>com.github.xiaoymin</groupId> <artifactId>knife4j-spring-boot-starter</artifactId> <version>2.0.9</version> </dependency> 工具类 LomBok <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <scope>provided</scope> </dependency> io常用工具类 <dependency> <groupId>commons-io</groupId> <artifactId>commons-io</artifactId> <version>2.11.0</version> </dependency> 文件上传工具类 <dependency> <groupId>commons-fileupload</groupId> <artifactId>commons-fileupload</artifactId> <version>1.4</version> </dependency> Excel工具 <dependency> <groupId>org.apache.poi</groupId> <artifactId>poi-ooxml</artifactId> <version>4.1.2</version> </dependency> velocity代码生成使用模板 <dependency> <groupId>org.apache.velocity</groupId> <artifactId>velocity-engine-core</artifactId> <version>2.3</version> </dependency> Collections工具类 <dependency> <groupId>commons-collections</groupId> <artifactId>commons-collections</artifactId> <version>3.2.2</version> </dependency> Json工具类 <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> </dependency> FastJson <dependency> <groupId>com.alibaba.fastjson2</groupId> <artifactId>fastjson2</artifactId> <version>2.0.14</version> </dependency> JWT - Token生成与解析 <dependency> <groupId>io.jsonwebtoken</groupId> <artifactId>jjwt</artifactId> <version>0.9.1</version> </dependency> 验证码 <dependency> <groupId>com.github.penggle</groupId> <artifactId>kaptcha</artifactId> <version>2.3.2</version> </dependency> 定时任务 <dependency> <groupId>org.quartz-scheduler</groupId> <artifactId>quartz</artifactId> <exclusions> <exclusion> <groupId>com.mchange</groupId> <artifactId>c3p0</artifactId> </exclusion> </exclusions> </dependency> Yml解析器 <dependency> <groupId>org.yaml</groupId> <artifactId>snakeyaml</artifactId> </dependency> Jaxb <dependency> <groupId>javax.xml.bind</groupId> <artifactId>jaxb-api</artifactId> </dependency> pool 对象池 <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-pool2</artifactId> </dependency> HTML解析器 <dependency> <groupId>org.jsoup</groupId> <artifactId>jsoup</artifactId> <version>1.11.3</version> </dependency> Hutool 工具类 <dependency> <groupId>cn.hutool</groupId> <artifactId>hutool-all</artifactId> <version>4.1.19</version> </dependency> 其他依赖 解析客户端操作系统、浏览器等 <dependency> <groupId>eu.bitwalker</groupId> <artifactId>UserAgentUtils</artifactId> <version>1.21</version> </dependency> 获取系统信息 <dependency> <groupId>com.github.oshi</groupId> <artifactId>oshi-core</artifactId> <version>6.2.2</version> </dependency> Servlet 包 <dependency> <groupId>javax.servlet</groupId> <artifactId>javax.servlet-api</artifactId> </dependency> 阿里云OSS <dependency> <groupId>com.qiniu</groupId> <artifactId>qiniu-java-sdk</artifactId> <version>[7.2.0, 7.2.99]</version> </dependency> <dependency> <groupId>com.aliyun.oss</groupId> <artifactId>aliyun-sdk-oss</artifactId> <version>3.5.0</version> </dependency> <dependency> <groupId>com.qcloud</groupId> <artifactId>cos_api</artifactId> <version>5.6.3</version> <exclusions> <exclusion> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> </exclusion> </exclusions> </dependency> guava <dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>26.0-jre</version> </dependency> Activiti <dependency> <groupId>org.activiti</groupId> <artifactId>activiti-spring-boot-starter-basic</artifactI <version>5.22.0</version> <exclusions> <exclusion> <artifactId>activation</artifactId> <groupId>javax.activation</groupId> </exclusion> <exclusion> <artifactId>mybatis</artifactId> <groupId>org.mybatis</groupId> </exclusion> </exclusions> </dependency> Activiti 流程图 <dependency> <groupId>org.activiti</groupId> <artifactId>activiti-diagram-rest</artifactId> <version>5.22.0</version> <exclusions> <exclusion> <groupId>org.springframework.security</groupId> <artifactId>spring-security-config</artifactId> </exclusion> <exclusion> <groupId>org.springframework.security</groupId> <artifactId>spring-security-crypto</artifactId> </exclusion> <exclusion> <groupId>org.springframework.security</groupId> <artifactId>spring-security-web</artifactId> </exclusion> </exclusions> </dependency> Activiti 在线设计 <dependency> <groupId>org.activiti</groupId> <artifactId>activiti-modeler</artifactId> <version>${activiti.version}</version> <exclusions> <exclusion> <artifactId>commons-logging</artifactId> <groupId>commons-logging</groupId> </exclusion> <exclusion> <groupId>org.springframework.security</groupId> <artifactId>spring-security-config</artifactId> </exclusion> <exclusion> <groupId>org.springframework.security</groupId> <artifactId>spring-security-crypto</artifactId> </exclusion> <exclusion> <groupId>org.springframework.security</groupId> <artifactId>spring-security-web</artifactId> </exclusion> </exclusions> </dependency> FastDFS 分布式文件系统 <dependency> <groupId>com.github.tobato</groupId> <artifactId>fastdfs-client</artifactId> <version>1.27.2</version> </dependency> 线程传递值 <dependency> <groupId>com.alibaba</groupId> <artifactId>transmittable-thread-local</artifactId> <version>${transmittable-thread-local.version}</version> </dependency> 中央仓库 默认仓库 <repositories> <repository> <id>central</id> <name>Central Repository</name> <url>https://repo.maven.apache.org/maven2</url> <layout>default</layout> <snapshots> <enabled>false</enabled> </snapshots> </repository> </repositories> 阿里云 <repositories> <repository> <id>public</id> <name>aliyun nexus</name> <url>https://maven.aliyun.com/repository/public</url> <releases> <enabled>true</enabled> </releases> </repository> </repositories> <pluginRepositories> <pluginRepository> <id>public</id> <name>aliyun nexus</name> <url>https://maven.aliyun.com/repository/public</url> <releases> <enabled>true</enabled> </releases> <snapshots> <enabled>false</enabled> </snapshots> </pluginRepository> </pluginRepositories>
微服务概念 微服务是软件的一种现代方法，通过该方法，应用程序代码以可管理的小段形式交付 ，彼此独立。微服务是一种架构或者风格，它提倡将单一应用程序划分为一组小的服务，每个 服务运行在其独立的自己的进程中，服务之间互相协调，互相配合，为用户提供最终价值。 微服务的小规模和相对隔离性可以带来许多其他好处，例如更容易维护，提高生产率，更大的容错能力，更好的业务调整等。 主流框架 Spring Cloud ： Spring Cloud是关注全局的微服务协调整理治理框架，它将Spring Boot开发的一个个单体微服务整合并管理起来，为各个微服务之间提供，配置管理、服务发现、断路器、路由、微代理、事件总线、全局锁、决策竞选、分布式会话等等集成服务 Dubbo ：是阿里巴巴公司开源的一个高性能优秀的服务框架，使得应用可通过高性能的 RPC 实现服务的输出和输入功能，可以和 Spring框架无缝集成。Dubbo是一款高性能、轻量级的开源Java RPC框架，它提供了三大核心能力：面向接口的远程方法调用，智能容错和负载均衡，以及服务自动注册和发现。 主要技术点 服务的注册与发现 身份验证与授权 服务的伸缩控制 反向代理与负载均衡 路由控制 流量切换 日志管理性能度量、监控与调优 分布式跟踪 过载保护 服务降级服务部署与版本升级策略支持 错误处理 前情提要 微服务父模块依赖 <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.wcy.hand</groupId> <artifactId>Eureka-Study</artifactId> <packaging>pom</packaging> <version>1.0-SNAPSHOT</version> <modules> <module>Eureka-Client</module> <module>Eureka-Server</module> </modules> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.2.5.RELEASE</version> </parent> <dependencyManagement> <dependencies> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-dependencies</artifactId> <version>Hoxton.SR6</version> <type>pom</type> <scope>import</scope> </dependency> </dependencies> </dependencyManagement> </project> 微服务组件均配置于子模块中，故无需版本号 Eureka Spring-Cloud Eureka 是Spring Cloud集合中一个组件，它是对Eureka的集成，用于服务注册和发现。 Eureka是Netflix中的一个开源框架。它和 zookeeper、Consul一样，都是用于服务注册管理的，同样，Spring-Cloud 还集成了Zookeeper和Consul。 Eureka 采用了 C-S 的设计架构。 **Eureka Server 作为服务注册功能的服务器，它是服务注册中心** 。而系统中的其他微服务 使用 Eureka 的客户端连接到 Eureka Server并维持心跳连接 。这样系统的维护人员就可以通过 Eureka Server 来监控系统中各个微服务是否正常运行。Spring Cloud 的一些其他模块（比如Zuul）就可以通过 Eureka Server 来发现系统中的其他微服务，并执行相关的逻辑 Eureka上的服务实例可以分为两种：Eureka Server和Eureka Client。 为了便于理解，我们将Eureka client再分为Service Provider和Service Consumer。Eureka Server 提供服务注册和发现Service Provider 服务提供方，将自身服务注册到Eureka，从而使服务消费方能够找到Service Consumer服务消费方，从Eureka获取注册服务列表，从而能够消费服务 Eureka集群及搭建 [post cid="1086" cover=""/] Ribbon Spring Cloud Ribbon 是基于Netflix Ribbon实现的一套客户端负载均衡的工具 简单的说，Ribbon是Netflix发布的开源项目，主要功能是提供客户端的软件负载均衡算法，将Netflix的中间层服务连接在一起。Ribbon客户端组件提供一系列完善的配置项如连接超时，重试等。 简单的说，就是在配置文件中列出 Load Balancer（简称LB） 后面所有的机器，Ribbon会自动的帮助你基于某种规则（如简单轮询，随机连接等）去连接这些机器。我们也很容易 使用Ribbon实现自定义的负载均衡算法 Load Balance 负载均衡(Load Balance)，在微服务或分布式集群中经常用的一种应用。 负载均衡简单的说就是 将用户的请求平摊的分配到多个服务 上，从而达到系统的HA。常见的负载均衡有软件Nginx，LVS，硬件 F5等。相应的在中间件，例如：Dubbo和Spring Cloud中均给我们提供了负载均衡 Spring Cloud的负载均衡算法可以自定义 **。 负载均衡算法在客户端 客户端维护服务地址列表 使用步骤 引入maven依赖 将依赖放在客户端下 <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId> </dependency> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-ribbon</artifactId> </dependency> 配置文件 现在的 Ribbon 基本上不需要单独引用，因为绝大多数的注册中心（Eureka，consul等）都会集成它 采用Eureka的配置文件即可 server: port: 8989 spring: application: name: eureka-client eureka: client: service-url: service-url: defaultZone: http://localhost:8761/eureka,http://localhost:8762/eureka,http://localhost:8763/eureka # instance: # lease-expiration-duration-in-seconds: 10 # 修改eureka server默认接受心跳的最大时间 默认90s # lease-renewal-interval-in-seconds: 5 # 指定客户端多久向服务端发送一次心跳 默认30s @LoadBalance注解配置 import org.springframework.cloud.client.loadbalancer.LoadBalanced; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.client.RestTemplate; /** * @author wcy * @DATE 2022/9/27 */ @Configuration public class RestTemplateConfig { @Bean @LoadBalanced //开启负载均衡， public RestTemplate getRestTemplate(){ return new RestTemplate(); } } 主启动类添加注解 主启动类添加-****@EnableEurekaClient @RibbonClient**** 负载均衡策略 配置ribbon负载策略 配置类 配置类不能放在被 @ComponentScan 注解能扫描到的包下，否则将被所有的 ribbon client 共享 如 有A\B\C三种服务，如果ribbon配置类被该注解扫描到了，我们原本只想让A服务自己来遵循该负载策略，结果 我们所有的服务 A B C 全都按照这个策略来执行了，达不到独自配置的目的 import com.netflix.loadbalancer.IRule; import com.netflix.loadbalancer.RandomRule; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** * @author wcy * @date 2022/9/27 */ @Configuration public class RibbonConfig { @Bean public IRule iRule(){ return new RandomRule();//随机策略 } } 配置生效 在主启动类上添加 ribbon客户端注解 @RibbonClient(name = "PROVIDER-SERVICE",configuration = RibbonConfig.class) name 是指定哪个服务使用该策略， configuration 是我们写的配置类 @RibbonClients不同于@RibbonClient，它 可以为所有的Ribbon客户端提供默认配置。 @RibbonClients(value = {@RibbonClient(name = "user-service",configuration = NewRuleConfig.class)}) 或者 @RibbonClients(defaultConfiguration = DefaultRibbonConfig.class) 设置访问 @Autowired private RestTemplate restTemplate; @GetMapping("/users/{id}") public User findById(@PathVariable Long id) { // 这里用到了RestTemplate的占位符能力 User user = this.restTemplate.getForObject( "http://service-name/users/{id}", User.class, id); //用服务名称代替ip return user; } Hystrix 分布式系统面临的问题 复杂分布式体系结构中的应用程序有数十个依赖关系，每个依赖关系在某些时候将不可避免地失败。 多个微服务之间调用的时候，假设 微服务A调用微服务B和微服务C ， 微服务B和微服务C又调用其它的微服务 ，这就是所谓的“ 扇出 ”。如果扇出的链路上 某个微服务的调用响应时间过长或者不可用 ，对微服务A的调用就会占用越来越多的系统资源， 进而引起系统崩溃 ，所谓的“ 雪崩效应 ”也叫 服务雪崩 。 对于高流量的应用来说，单一的后端依赖可能会导致所有服务器上的所有资源都在几秒钟内饱和 比失败更糟糕的是，这些应用程序还可能导致服务之间的延迟增加，备份队列，线程和其他系统资源紧张，导致整个系统发生更多的级联故障。 这些都表示需要对故障和延迟进行隔离和管理，以便单个依赖关系的失败，不能取消整个应用程序或系统。 断路器Hystrix介绍 Hystrix是一个 用于处理分布式系统的延迟和容错 的开源库，在分布式系统里，许多依赖不可避免的会调用失败，比如超时、异常等问题 Hystrix能够保证在一个依赖出问题的情况下，不会导致整体服务失败，避免级联故障，以提高分布式系统的弹性。 断路器”本身是一种开关装置，当某个服务单元发生故障之后 通过断路器的故障监控（类似熔断保险丝），向调用方返回一个符合预期的、可处理的备选响应（FallBack），而不是长时间的等待或者抛出调用方无法处理的异常，这样就保证了服务调用方的线程不会被长时间、不必要地占用，从而避免了故障在分布式系统中的蔓延，乃至雪崩。 Hystrix功能包括：服务降级，服务熔断，服务限流，几近实时监控 服务熔断 熔断机制是应对雪崩效应的一种微服务链路保护机制。 当扇出链路的某个微服务不可用或者响应时间太长时，会进行服务的降级，进而熔断该节点微服务的调用，快速返回“错误”的响应信息。当检测到该节点微服务调用响应正常后恢复调用链路。 在Spring Cloud框架里熔断机制通过Hystrix实现。Hystrix会监控微服务间调用的状况，当失败的调用到一定阈值，缺省是5秒内20次调用失败就会启动熔断机制。熔断机制的注解是@HystrixCommand。 服务降级 服务降级服务降级的处理是在客户端完成的，与服务端没有关系。 整体资源快不够用了，忍痛将某些服务先关掉，待度过难关，再回来开启。 所谓 降级就是一般是从整体符合考虑，就是当某个服务熔断之后，服务器将不再被调用，此刻客户端可以自己准备一个本地的fallback回调，返回一个缺省值 这样做，虽然服务水平下降，但好歹可用， 比直接挂掉要强。 使用步骤 引入依赖 <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-hystrix</artifactId> </dependency> **** 所有需要被监控的微服务引入如下依赖**** <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-actuator</artifactId> </dependency> @HystrixCommand 核心配置 启用断路器 @HystrixProperty(name = "circuitBreaker.enabled", value = "true") 设置请求次数 @HystrixProperty(name = "circuitBreaker.requestVolumeThreshold", value = "10") 设置时间窗口期 @HystrixProperty(name = "circuitBreaker.sleepWindowInMilliseconds", value = "10000") 设置失败率 @HystrixProperty(name = "circuitBreaker.errorThresholdPercentage", value = "60") 如上设置的值，如果在10秒内，失败率达到请求次数（10）的百分之60，也就是6次就会 打开断路器；否则断路器依然关闭 当断路器开启时，所有的请求都不会转发，而是直接调用fallback方法 ；在一段时间后（默认5秒），断路器处于半开状态，尝试将请求转发 如果得到正确的响应，则将断路器关闭，恢复正常调用 ；否则断路器状态再次打开，重新计时后再进入半开状态 @Component public class PaymentService { @HystrixCommand(fallbackMethod = "breakFallback", commandProperties = { // 开启断路器 @HystrixProperty(name = "circuitBreaker.enabled", value = "true"), // 请求次数 @HystrixProperty(name = "circuitBreaker.requestVolumeThreshold", value = "10"), // 时间窗口期 @HystrixProperty(name = "circuitBreaker.sleepWindowInMilliseconds", value = "10000"), // 失败率 @HystrixProperty(name = "circuitBreaker.errorThresholdPercentage", value = "60") }) public String breakTest(Integer id) { if (id < 0) { throw new RuntimeException("ID 不能为负数"); } return "正常访问： " + IdUtil.simpleUUID(); } public String breakFallback(Integer id) { return Thread.currentThread().getName() + "ID 不能为负数 熔断服务降级FallBack!" + id; } } 启动类加上注解 熔断降级示例 以两个服务端一个客户端为例，降级在客户端进行，熔断在服务端进行 客户端 @FeignClient(name = "server",value = "server",fallback = MessageServiceFallback.class) public interface HelloFeign { /** * 正常调用 * @param id * @return */ @GetMapping("/normal/{id}") String normal(@PathVariable("id") int id); } FallBack配置类 @Component public class MessageServiceFallback implements HelloFeign { @Override public String normal(int id) { return "Error time out !!!"; } } 配置文件 hystrix: command: default: execution: isolation: thread: timeoutInMilliseconds: 1500 # 超时时间 接口层 @RestController public class HelloController { @Resource public HelloService helloService; @GetMapping("/normal") public void normal() { for (int i = 0; i < 30; i++) { System.out.println(helloService.normal(i)); } } } 服务端 接口层 @RestController public class HelloController { @Value("${server.port}") private int port; @Value("${spring.application.name}") private String name; @GetMapping("/normal/{id}") //表示只对该接口生效 @HystrixCommand(fallbackMethod = "breakFallback",commandProperties = { //开启熔断 @HystrixProperty(name="circuitBreaker.enabled",value = "true"), //设置Hystrix的超时时间，默认1s @HystrixProperty(name="execution.isolation.thread.timeoutInMilliseconds",value = "1000"), //监控时间 默认5000 毫秒 @HystrixProperty(name="circuitBreaker.sleepWindowInMilliseconds",value = "10000"), //失败次数。默认20次 @HystrixProperty(name="circuitBreaker.requestVolumeThreshold",value = "5"), //失败率 默认50% @HystrixProperty(name="circuitBreaker.errorThresholdPercentage",value = "50") }) public String normal(@PathVariable("id") int id) { if (id > 9 && id % 2 == 0) { try { sleep(1500); } catch (InterruptedException e) { e.printStackTrace(); } } return id + " : 你调用的服务名称为: " + name + " ,端口为 : " + port; } public String breakFallback(int id) { return id+": 端口 : "+ port+" 服务已熔断！！！"; } } 配置文件 hystrix: command: default: circuitBreaker: errorThresholdPercentage: 50 # 触发熔断错误比例阈值，默认值50% sleepWindowInMilliseconds: 10000 # 熔断后休眠时长，默认值5秒 requestVolumeThreshold: 10 # 熔断触发最小请求次数，默认值是20 execution: isolation: thread: timeoutInMilliseconds: 2000 # 熔断超时设置，默认为1秒 最终效果 0 : 你调用的服务名称为: server ,端口为 : 8888 1 : 你调用的服务名称为: server ,端口为 : 7777 2 : 你调用的服务名称为: server ,端口为 : 8888 3 : 你调用的服务名称为: server ,端口为 : 7777 4 : 你调用的服务名称为: server ,端口为 : 8888 5 : 你调用的服务名称为: server ,端口为 : 7777 6 : 你调用的服务名称为: server ,端口为 : 8888 7 : 你调用的服务名称为: server ,端口为 : 7777 8 : 你调用的服务名称为: server ,端口为 : 8888 9 : 你调用的服务名称为: server ,端口为 : 7777 Error time out !!! 11 : 你调用的服务名称为: server ,端口为 : 8888 Error time out !!! 13 : 你调用的服务名称为: server ,端口为 : 7777 Error time out !!! 15 : 你调用的服务名称为: server ,端口为 : 8888 Error time out !!! 17 : 你调用的服务名称为: server ,端口为 : 7777 Error time out !!! 19 : 你调用的服务名称为: server ,端口为 : 8888 Error time out !!! 21 : 你调用的服务名称为: server ,端口为 : 7777 Error time out !!! 23: 端口 : 8888 服务已熔断！！！ 24: 端口 : 7777 服务已熔断！！！ 25: 端口 : 8888 服务已熔断！！！ 26: 端口 : 7777 服务已熔断！！！ 27: 端口 : 8888 服务已熔断！！！ 28: 端口 : 7777 服务已熔断！！！ 29: 端口 : 8888 服务已熔断！！！ Hystrix Dashboard Delay: 该参数用来控制服务器上轮询监控信息的延迟时间，默认为2000毫秒，可以通过配置该属性来降低客户端的网络和 CPU消耗。 Title:该参数对应了头部标题Hystrix Stream之后的内容，默认会使用具体监控实例的URL，可以通过配置该信息来展示更合 适的标题。 监控查看：实心圆：共有两种含义。它通过颜色的变化代表了实例的健康程度，它的健康度从绿色<黄色<橙色<红色递减。 该实心圆除了颜色的变化之外，它的大小也会根据实例的请求流量发生变化，流量越大该实心圆就越大。所以通过该实心圆的展示 就可以在大量的实例中快速的发现故障实例和高压力实例。 Feign 声明式的服务调用Feign Feign简介 是一个声明式的Web服务客户端，使得编写Web服务客户端变得非常容易。 只需要创建一个接口，然后在上面添加注解即可。 Feign的作用 Feign 旨在使编写Java Http客户端变得更容易 。前面在使用Ribbon & RestTemplate时，利用RestTemplate对http请求的封装处理，形成了一套模版化的调用方法。但是在实际开发中，由于对服务依赖的调用可能不止一处， 往往一个接口会被多处调用，所以通常都会针对每个微服务自行封装一些客户端类来包装这些依赖服务的调用。 所以， Feign在此基础上做了进一步封装，由他来帮助我们定义和实现依赖服务接口的定义 。在Feign的实现下，我们只需创建一个接口并使用注解的方式来配置它(以前是Dao接口上面标注Mapper注解,现在是一个微服务接口上面 标注一个Feign注解即可 )，即可完成对服务提供方的接口绑定，简化了使用Spring cloud Ribbon时，自动封装服务调用客户端的开发量。 引入依赖 <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-openfeign</artifactId> </dependency> @FeignClient 定义一个FeignClient客户端，其实就是定义个一个接口，加上@feignClient注解 @FeignClient(value = "CART") public interface CartFeignClient { @PostMapping("/cart/{productId}") Long addCart(@PathVariable("productId")Long productId); } 上面是最简单的feign client的使用，声明完为feign client后，其他spring管理的类，如service就可以直接注入使用了，例如 //这里直接注入feign client @Autowired private CartFeignClient cartFeignClient; @PostMapping("/toCart/{productId}") public ResponseEntity addCart(@PathVariable("productId") Long productId){ Long result = cartFeignClient.addCart(productId); return ResponseEntity.ok(result); } 启动类添加注解 Gateway 引入依赖 <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-gateway</artifactId> </dependency> 配置文件 # 服务名称 spring: application: name: gateway # 开启 Gateway 服务注册中心服务发现 cloud: gateway: discovery: locator: enabled: true routes: - id: client uri: lb://client predicates: - Path=/client/** # Eureka Server 配置 eureka: client: service-url: defaultZone: http://localhost:8761/eureka/ # 配置Gateway日志等级，输出转发细节信息 logging: level: org.springframework.cloud.gateway: info server: port: 8080 启动后Eureka效果如图 通过8080端口我们即可访问其他模块的服务 我们在下图可以看到，gateway挑选了一个客户端服务进行调用 Zuul Zuul路由网关 Zuul包含了对 请求的路由和过滤 两个最主要的功能 其中 路由功能负责将外部请求转发到具体的微服务实例上 ，是实现外部访问统一入口的基础而过滤器功能则负责对请求的处理过程进行干预，是实现请求校验、服务聚合等功能的基础. Zuul和Eureka进行整合，将 Zuul自身注册为Eureka服务治理下的应用，同时从Eureka中获得其他微服务的消息 ，也即以后的访问微服务都是通过Zuul跳转后获得。 Zuul服务 最终还是会注册进Eureka。Zuul功能：服务代理、路由、过滤三大功能 引入依赖 <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-zuul</artifactId> </dependency> 配置文件 server: port: 9528 spring: application: name: cloud-zuul eureka: #将zuul注册进服务注册中心 client: service-url: defaultZone: http://localhost:8761/eureka,http://localhost:8762/eureka,http://localhost:8763/eureka register-with-eureka: true fetch-registry: true registry-fetch-interval-seconds: 10 instance: instance-id: Zuul-9528 #修改在eureka注册中心status列显示的服务实例id，默认是主机名+服务名+端口 prefer-ip-address: true #设置为true，将鼠标悬停在注册中心status列的服务实例上，浏览器左下角显示ip和端口 zuul: routes: payment: #payment是自定义的名字，随便定义 serviceId: eureka-client #真实的服务名字 path: /eureka-client/** #配置访问的路径，路径中带有eureka-client 的请求会被路由到真实的服务。 # ignored-services: cloud-payment-service #不能通过服务名字进行访问的服务的名字 # prefix: /cloud-service #为所有访问路径再添加前缀 启动类加入注解 在启动类上添加注解 @EnableZuulProxy和@EnableEurekaClient ，将Zuul网关服务注册到Eureka，可以通过服务名称调用其他的服务 Spring Cloud Config 微服务意味着要将单体应用中的业务拆分成一个个自服务， 每个服务的粒度相对较小，因此系统中会出现大量的服务。 由于每个服务都 需要必要的配置 才能运行，所以一套 集中式的、动态的配置管理 设施是必不可少的。 Spring Cloud Config为微服务架构中的微服务提供集中化的尾部配置支持，配置服务器为各个不同微服务应用的所有环境提供了一个中心化的外部配置。 Spring Cloud Config分为服务端和客户端两部分。 服务端也称为分布式配置中心，它是一个独立的微服务应用，用来连接配置服务器并未客户端提供获取配置信息，加密/解密信息等访问接口。 客户端则是通过指定的配置中心来管理应用资源，以及与业务相关的配置内容，并在启动的时候从配置中心获取和加载配置信息 。配置服务器默认采用git来存储配置信息，这样就有助于对环境配置进行版本管理，并且可以通过git客户端工具来方便的管理和访问配置内容。 功能介绍 集中管理配置文件 不同环境不同配置，动态化的配置更新，分环境部署，比如dev/test/prod/beta/release 运行期间动态调整配置，不再需要在每个服务器的机器上编写配置文件 ，服务会向配置中心统一拉取配置自己的信息运行期间动态调整配置，不再需要在每个服务部署的机器上编写配置文件，服务会想配置中心统 一拉取配置自己的信息 当配置发生变动时， 服务不需要重启即可感知到配置的变化并应用新的配置 将配置信息以REST接口的形式暴露 引入依赖 <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-config-server</artifactId> </dependency> 架构分解 网关集群：数据的聚合、实现对接入客户端的身份认证、防报文重放与防数据篡改、功能调用的业务鉴权、响应数据的脱敏、流量与并发控制等 业务集群：一般情况下移动端访问和浏览器访问的网关需要隔离，防止业务耦合 Local Cache：由于客户端访问业务可能需要调用多个服务聚合，所以本地缓存有效的降低了服务调用的频次，同时也提示了访问速度。本地缓存一般使用自动过期方式，业务场景中允许有一定的数据延时。 服务层：原子服务层，实现基础的增删改查功能，如果需要依赖其他服务需要在Service层主动调用 Remote Cache：访问DB前置一层分布式缓存，减少DB交互次数，提升系统的TPS DAL：数据访问层，如果单表数据量过大则需要通过DAL层做数据的分库分表处理。 MQ：消息队列用来解耦服务之间的依赖，异步调用可以通过MQ的方式来执行数据库主从：服务化过程中毕竟的阶段，用来提升系统的TPS
Kafka简介 消息队列 消息队列——用于存放消息的组件 程序员可以将消息放入到队列中，也可以从消息队列中获取消息 很多时候消息队列不是一个永久性的存储，是作为临时存储存在的（设定一个期限：设置消息在MQ中保存10天） 消息队列中间件：消息队列的组件，例如：Kafka、Active MQ、RabbitMQ、RocketMQ、ZeroMQ Kafka的应用场景 异步处理 可以将一些比较耗时的操作放在其他系统中，通过消息队列将需要进行处理的消息进行存储，其他系统可以消费消息队列中的数据 比较常见的：发送短信验证码、发送邮件 系统解耦 原先一个微服务是通过接口（HTTP）调用另一个微服务，这时候耦合很严重，只要接口发生变化就会导致系统不可用 使用消息队列可以将系统进行解耦合，现在第一个微服务可以将消息放入到消息队列中，另一个微服务可以从消息队列中把消息取出来进行处理。进行系统解耦 流量削峰 因为消息队列是低延迟、高可靠、高吞吐的，可以应对大量并发 日志处理 可以使用消息队列作为临时存储，或者一种通信管道 消息队列的两种模型 生产者、消费者模型 生产者负责将消息生产到MQ中 消费者负责从MQ中获取消息 生产者和消费者是解耦的，可能是生产者一个程序、消费者是另外一个程序 消息队列的模式 点对点：一个消费者消费一个消息 发布订阅：多个消费者可以消费一个消息 Kafka集群搭建 Kafka集群是必须要有ZooKeeper的 注意： 每一个Kafka的节点都需要修改broker.id（每个节点的标识，不能重复） log.dir数据存储目录需要配置 Kafka的生产者/消费者/工具 安装Kafka集群，可以测试以下 创建一个topic主题（消息都是存放在topic中，类似mysql建表的过程） 基于kafka的内置测试生产者脚本来读取标准输入（键盘输入）的数据，并放入到topic中 基于kafka的内置测试消费者脚本来消费topic中的数据 推荐大家开发的使用Kafka Tool 浏览Kafka集群节点、多少个topic、多少个分区 创建topic/删除topic 浏览ZooKeeper中的数据 Kafka的基准测试工具 Kafka中提供了内置的性能测试工具 生产者：测试生产每秒传输的数据量（多少条数据、多少M的数据） 5000000 records sent, 11825.446943 records/sec (11.28 MB/sec), 2757.61 ms avg latency 消费者：测试消费每条拉取的数据量 对比生产者和消费者：消费者的速度更快 Kafka Java API开发 生产者程序开发 创建连接 bootstrap.servers：Kafka的服务器地址 acks：表示当生产者生产数据到Kafka中，Kafka中会以什么样的策略返回 key.serializer：Kafka中的消息是以key、value键值对存储的，而且生产者生产的消息是需要在网络上传到的，这里指定的是StringSerializer方式，就是以字符串方式发送（将来还可以使用其他的一些序列化框架：Google ProtoBuf、Avro） value.serializer：同上 创建一个生产者对象KafkaProducer 调用send方法发送消息（ProducerRecor，封装是key-value键值对） 调用Future.get表示等带服务端的响应 关闭生产者 public class KafkaProducerTest { public static void main(String[] args) throws ExecutionException, InterruptedException { // 1. 创建用于连接Kafka的Properties配置 Properties props = new Properties(); props.put("bootstrap.servers", "ip:9092"); props.put("acks", "all"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 2. 创建一个生产者对象KafkaProducer KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props); // 3. 发送1-100的消息到指定的topic中 for(int i = 0; i < 100; ++i) { // 构建一条消息，直接new ProducerRecord ProducerRecord<String, String> producerRecord = new ProducerRecord<>("test", null, i + ""); Future<RecordMetadata> future = kafkaProducer.send(producerRecord); // 调用Future的get方法等待响应 future.get(); System.out.println("第" + i + "条消息写入成功！"); } // 4.关闭生产者 kafkaProducer.close(); } } 消费者程序开发 group.id：消费者组的概念，可以在一个消费组中包含多个消费者。如果若干个消费者的group.id是一样的，表示它们就在一个组中，一个组中的消费者是共同消费Kafka中topic的数据。 Kafka是一种拉消息模式的消息队列，在消费者中会有一个offset，表示从哪条消息开始拉取数据 kafkaConsumer.poll：Kafka的消费者API是一批一批数据的拉取 /** * 消费者程序 * * 1.创建Kafka消费者配置 * Properties props = new Properties(); * props.setProperty("bootstrap.servers", "ip:9092"); * props.setProperty("group.id", "test"); * props.setProperty("enable.auto.commit", "true"); * props.setProperty("auto.commit.interval.ms", "1000"); * props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); * props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); * * 2.创建Kafka消费者 * 3.订阅要消费的主题 * 4.使用一个while循环，不断从Kafka的topic中拉取消息 * 5.将将记录（record）的offset、key、value都打印出来 */ public class KafkaConsumerTest { public static void main(String[] args) { // 1.创建Kafka消费者配置 Properties props = new Properties(); props.setProperty("bootstrap.servers", "ip:9092"); // 消费者组（可以使用消费者组将若干个消费者组织到一起），共同消费Kafka中topic的数据 // 每一个消费者需要指定一个消费者组，如果消费者的组名是一样的，表示这几个消费者是一个组中的 props.setProperty("group.id", "test"); // 自动提交offset props.setProperty("enable.auto.commit", "true"); // 自动提交offset的时间间隔 props.setProperty("auto.commit.interval.ms", "1000"); // 拉取的key、value数据的 props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // 2.创建Kafka消费者 KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props); // 3. 订阅要消费的主题 // 指定消费者从哪个topic中拉取数据 kafkaConsumer.subscribe(Arrays.asList("test")); // 4.使用一个while循环，不断从Kafka的topic中拉取消息 while(true) { // Kafka的消费者一次拉取一批的数据 ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(5)); // 5.将将记录（record）的offset、key、value都打印出来 for (ConsumerRecord<String, String> consumerRecord : consumerRecords) { // 主题 String topic = consumerRecord.topic(); // offset：这条消息处于Kafka分区中的哪个位置 long offset = consumerRecord.offset(); // key\value String key = consumerRecord.key(); String value = consumerRecord.value(); System.out.println("topic: " + topic + " offset:" + offset + " key:" + key + " value:" + value); } } } } 生产者使用异步方式生产消息 使用匿名内部类实现Callback接口，该接口中表示Kafka服务器响应给客户端，会自动调用onCompletion方法 metadata：消息的元数据（属于哪个topic、属于哪个partition、对应的offset是什么） exception：这个对象Kafka生产消息封装了出现的异常，如果为null，表示发送成功，如果不为null，表示出现异常。 // 二、使用异步回调的方式发送消息 ProducerRecord<String, String> producerRecord = new ProducerRecord<>("test", null, i + ""); kafkaProducer.send(producerRecord, new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { // 1. 判断发送消息是否成功 if(exception == null) { // 发送成功 // 主题 String topic = metadata.topic(); // 分区id int partition = metadata.partition(); // 偏移量 long offset = metadata.offset(); System.out.println("topic:" + topic + " 分区id：" + partition + " 偏移量：" + offset); } else { // 发送出现错误 System.out.println("生产消息出现异常！"); // 打印异常消息 System.out.println(exception.getMessage()); // 打印调用栈 System.out.println(exception.getStackTrace()); } } }); Kafka中的重要概念 broker Kafka服务器进程，生产者、消费者都要连接broker 一个集群由多个broker组成，功能实现Kafka集群的负载均衡、容错 producer：生产者 consumer：消费者 topic：主题，一个Kafka集群中，可以包含多个topic。一个topic可以包含多个分区 是一个逻辑结构，生产、消费消息都需要指定topic partition：Kafka集群的分布式就是由分区来实现的。一个topic中的消息可以分布在topic中的不同partition中 replica：副本，实现Kafkaf集群的容错，实现partition的容错。一个topic至少应该包含大于1个的副本 consumer group：消费者组，一个消费者组中的消费者可以共同消费topic中的分区数据。每一个消费者组都一个唯一的名字。配置group.id一样的消费者是属于同一个组中 offset：偏移量。相对消费者、partition来说，可以通过offset来拉取数据 消费者组 一个消费者组中可以包含多个消费者，共同来消费topic中的数据 一个topic中如果只有一个分区，那么这个分区只能被某个组中的一个消费者消费 有多少个分区，那么就可以被同一个组内的多少个消费者消费 幂等性 生产者消息重复问题 Kafka生产者生产消息到partition，如果直接发送消息，kafka会将消息保存到分区中，但Kafka会返回一个ack给生产者，表示当前操作是否成功，是否已经保存了这条消息。如果ack响应的过程失败了，此时生产者会重试，继续发送没有发送成功的消息，Kafka又会保存一条一模一样的消息 在Kafka中可以开启幂等性 当Kafka的生产者生产消息时，会增加一个pid（生产者的唯一编号）和sequence number（针对消息的一个递增序列） 发送消息，会连着pid和sequence number一块发送 kafka接收到消息，会将消息和pid、sequence number一并保存下来 如果ack响应失败，生产者重试，再次发送消息时，Kafka会根据pid、sequence number是否需要再保存一条消息 判断条件：生产者发送过来的sequence number 是否小于等于 partition中消息对应的sequence 事务编程 开启事务的条件 生产者 // 开启事务必须要配置事务的ID props.put("transactional.id", "dwd_user"); 消费者 // 配置事务的隔离级别 props.put("isolation.level","read_committed"); // 关闭自动提交，一会我们需要手动来提交offset，通过事务来维护offset props.setProperty("enable.auto.commit", "false"); 生产者 初始化事务 开启事务 需要使用producer来将消费者的offset提交到事务中 提交事务 如果出现异常回滚事务 如果使用了事务，不要使用异步发送 public class TransactionProgram { public static void main(String[] args) { // 1. 调用之前实现的方法，创建消费者、生产者对象 KafkaConsumer<String, String> consumer = createConsumer(); KafkaProducer<String, String> producer = createProducer(); // 2. 生产者调用initTransactions初始化事务 producer.initTransactions(); // 3. 编写一个while死循环，在while循环中不断拉取数据，进行处理后，再写入到指定的topic while(true) { try { // (1) 生产者开启事务 producer.beginTransaction(); // 这个Map保存了topic对应的partition的偏移量 Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>(); // 从topic中拉取一批的数据 // (2) 消费者拉取消息 ConsumerRecords<String, String> concumserRecordArray = consumer.poll(Duration.ofSeconds(5)); // (3) 遍历拉取到的消息，并进行预处理 for (ConsumerRecord<String, String> cr : concumserRecordArray) { // 将1转换为男，0转换为女 String msg = cr.value(); String[] fieldArray = msg.split(","); // 将消息的偏移量保存 // 消费的是ods_user中的数据 String topic = cr.topic(); int partition = cr.partition(); long offset = cr.offset(); int i = 1 / 0; // offset + 1：offset是当前消费的记录（消息）对应在partition中的offset，而我们希望下一次能继续从下一个消息消息 // 必须要+1，从能消费下一条消息 offsetMap.put(new TopicPartition(topic, partition), new OffsetAndMetadata(offset + 1)); // 将字段进行替换 if(fieldArray != null && fieldArray.length > 2) { String sexField = fieldArray[1]; if(sexField.equals("1")) { fieldArray[1] = "男"; } else if(sexField.equals("0")){ fieldArray[1] = "女"; } } // 重新拼接字段 msg = fieldArray[0] + "," + fieldArray[1] + "," + fieldArray[2]; // (4) 生产消息到dwd_user topic中 ProducerRecord<String, String> dwdMsg = new ProducerRecord<>("dwd_user", msg); // 发送消息 Future<RecordMetadata> future = producer.send(dwdMsg); try { future.get(); } catch (Exception e) { e.printStackTrace(); producer.abortTransaction(); } // new Callback() // { // @Override // public void onCompletion(RecordMetadata metadata, Exception exception) { // // 生产消息没有问题 // if(exception == null) { // System.out.println("发送成功:" + dwdMsg); // } // else { // System.out.println("生产消息失败:"); // System.out.println(exception.getMessage()); // System.out.println(exception.getStackTrace()); // } // } // }); } producer.sendOffsetsToTransaction(offsetMap, "ods_user"); // (6) 提交事务 producer.commitTransaction(); }catch (Exception e) { e.printStackTrace(); // (7) 捕获异常，如果出现异常，则取消事务 producer.abortTransaction(); } } } // 一、创建一个消费者来消费ods_user中的数据 private static KafkaConsumer<String, String> createConsumer() { // 1. 配置消费者的属性（添加对事务的支持） Properties props = new Properties(); props.setProperty("bootstrap.servers", "ip:9092"); props.setProperty("group.id", "ods_user"); // 配置事务的隔离级别 props.put("isolation.level","read_committed"); // 关闭自动提交，一会我们需要手动来提交offset，通过事务来维护offset props.setProperty("enable.auto.commit", "false"); // 反序列化器 props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // 2. 构建消费者对象 KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props); // 3. 订阅一个topic kafkaConsumer.subscribe(Arrays.asList("ods_user")); return kafkaConsumer; } // 二、编写createProducer方法，用来创建一个带有事务配置的生产者 private static KafkaProducer<String, String> createProducer() { // 1. 配置生产者带有事务配置的属性 Properties props = new Properties(); props.put("bootstrap.servers", "ip:9092"); // 开启事务必须要配置事务的ID props.put("transactional.id", "dwd_user"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 2. 构建生产者 KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props); return kafkaProducer; } } Kafka中的分区副本机制 生产者的分区写入策略 轮询（按照消息尽量保证每个分区的负载）策略，消息会均匀地分布到每个partition 写入消息的时候，key为null的时候，默认使用的是轮询策略 随机策略（不使用） 按key写入策略，key.hash() % 分区的数量 自定义分区策略（类似于MapReduce指定分区） 乱序问题 在Kafka中生产者是有写入策略，如果topic有多个分区，就会将数据分散在不同的partition中存储 当partition数量大于1的时候，数据（消息）会打散分布在不同的partition中 如果只有一个分区，消息是有序的 消费组Consumer Group Rebalance机制 再均衡：在某些情况下，消费者组中的消费者消费的分区会产生变化，会导致消费者分配不均匀（例如：有两个消费者消费3个，因为某个partition崩溃了，还有一个消费者当前没有分区要削峰），Kafka Consumer Group就会启用rebalance机制，重新平衡这个Consumer Group内的消费者消费的分区分配。 触发时机 消费者数量发生变化 某个消费者crash 新增消费者 topic的数量发生变化 某个topic被删除 partition的数量发生变化 删除partition 新增partition 不良影响 发生rebalance，所有的consumer将不再工作，共同来参与再均衡，直到每个消费者都已经被成功分配所需要消费的分区为止（rebalance结束） 消费者的分区分配策略 分区分配策略：保障每个消费者尽量能够均衡地消费分区的数据，不能出现某个消费者消费分区的数量特别多，某个消费者消费的分区特别少 Range分配策略（范围分配策略）：Kafka默认的分配策略 n：分区的数量 / 消费者数量 m：分区的数量 % 消费者数量 前m个消费者消费n+1个分区 剩余的消费者消费n个分区 RoundRobin分配策略（轮询分配策略） 消费者挨个分配消费的分区 Striky粘性分配策略 在没有发生rebalance跟轮询分配策略是一致的 发生了rebalance，轮询分配策略，重新走一遍轮询分配的过程。而粘性会保证跟上一次的尽量一致，只是将新的需要分配的分区，均匀的分配到现有可用的消费者中即可 减少上下文的切换 副本的ACK机制 producer是不断地往Kafka中写入数据，写入数据会有一个返回结果，表示是否写入成功。这里对应有一个ACKs的配置。 acks = 0：生产者只管写入，不管是否写入成功，可能会数据丢失。性能是最好的 acks = 1：生产者会等到leader分区写入成功后，返回成功，接着发送下一条 acks = -1/all：确保消息写入到leader分区、还确保消息写入到对应副本都成功后，接着发送下一条，性能是最差的 根据业务情况来选择ack机制，是要求性能最高，一部分数据丢失影响不大，可以选择0/1。如果要求数据一定不能丢失，就得配置为-1/all。 分区中是有leader和follower的概念，为了确保消费者消费的数据是一致的，只能从分区leader去读写消息，follower做的事情就是同步数据，Backup。 高级API（High-Level API）、低级API（Low-Level API） 高级API就是直接让Kafka帮助管理、处理分配、数据 offset存储在ZK中 由kafka的rebalance来控制消费者分配的分区 开发起来比较简单，无需开发者关注底层细节 无法做到细粒度的控制 低级API：由编写的程序自己控制逻辑 自己来管理Offset，可以将offset存储在ZK、MySQL、Redis、HBase、Flink的状态存储 指定消费者拉取某个分区的数据 可以做到细粒度的控制 原有的Kafka的策略会失效，需要我们自己来实现消费机制 Kafka原理 leader和follower Kafka中的leader和follower是相对分区有意义，不是相对broker Kafka在创建topic的时候，会尽量分配分区的leader在不同的broker中，其实就是负载均衡 leader职责：读写数据 follower职责：同步数据、参与选举（leader crash之后，会选举一个follower重新成为分区的leader 注意和ZooKeeper区分 ZK的leader负责读、写，follower可以读取 Kafka的leader负责读写、follower不能读写数据（确保每个消费者消费的数据是一致的），Kafka一个topic有多个分区leader，一样可以实现数据操作的负载均衡 AR\ISR\OSR AR表示一个topic下的所有副本 ISR：In Sync Replicas，正在同步的副本（可以理解为当前有几个follower是存活的） OSR：Out of Sync Replicas，不再同步的副本 AR = ISR + OSR leader选举 Controller：controller是kafka集群的老大，是针对Broker的一个角色 Controller是高可用的，是用过ZK来进行选举 Leader：是针对partition的一个角色 Leader是通过ISR来进行快速选举 如果Kafka是基于ZK来进行选举，ZK的压力可能会比较大。例如：某个节点崩溃，这个节点上不仅仅只有一个leader，是有不少的leader需要选举。通过ISR快速进行选举。 leader的负载均衡 如果某个broker crash之后，就可能会导致partition的leader分布不均匀，就是一个broker上存在一个topic下不同partition的leader 通过以下指令，可以将leader分配到优先的leader对应的broker，确保leader是均匀分配的 bin/kafka-leader-election.sh --bootstrap-server ip:9092 --topic test --partition=2 --election-type preferred Kafka读写流程 写流程 通过ZooKeeper找partition对应的leader，leader是负责写的 producer开始写入数据 ISR里面的follower开始同步数据，并返回给leader ACK 返回给producer ACK 读流程 通过ZooKeeper找partition对应的leader，leader是负责读的 通过ZooKeeper找到消费者对应的offset 然后开始从offset往后顺序拉取数据 提交offset（自动提交——每隔多少秒提交一次offset、手动提交——放入到事务中提交） Kafka的物理存储 Kafka的数据组织结构 topic partition segment .log数据文件 .index（稀疏索引） .timeindex（根据时间做的索引） 深入了解读数据的流程 消费者的offset是一个针对partition全局offset 可以根据这个offset找到segment段 接着需要将全局的offset转换成segment的局部offset 根据局部的offset，就可以从（.index稀疏索引）找到对应的数据位置 开始顺序读取 消息传递的语义性 Flink里面有对应的每种不同机制的保证，提供Exactly-Once保障（二阶段事务提交方式） At-most once：最多一次（只管把数据消费到，不管有没有成功，可能会有数据丢失） At-least once：最少一次（有可能会出现重复消费） Exactly-Once：仅有一次（事务性性的保障，保证消息有且仅被处理一次） Kafka的消息不丢失 broker消息不丢失：因为有副本relicas的存在，会不断地从leader中同步副本，所以，一个broker crash，不会导致数据丢失，除非是只有一个副本。 生产者消息不丢失：ACK机制（配置成ALL/-1）、配置0或者1有可能会存在丢失 消费者消费不丢失：重点控制offset At-least once：一种数据可能会重复消费 Exactly-Once：仅被一次消费 数据积压 数据积压指的是消费者因为有一些外部的IO、一些比较耗时的操作（Full GC——Stop the world），就会造成消息在partition中一直存在得不到消费，就会产生数据积压 在企业中，我们要有监控系统，如果出现这种情况，需要尽快处理。虽然后续的Spark Streaming/Flink可以实现背压机制，但是数据累积太多一定对实时系统它的实时性是有说影响的 数据清理&配额限速 数据清理 Log Deletion（日志删除）：如果消息达到一定的条件（时间、日志大小、offset大小），Kafka就会自动将日志设置为待删除（segment端的后缀名会以 .delete结尾），日志管理程序会定期清理这些日志 默认是7天过期 Log Compaction（日志合并） 如果在一些key-value数据中，一个key可以对应多个不同版本的value 经过日志合并，就会只保留最新的一个版本 配额限速 可以限制Producer、Consumer的速率 防止Kafka的速度过快，占用整个服务器（broker）的所有IO资源
分布式搜索引擎 -数据聚合 详细项目可见：ElasticSearch-search 学习目标 1.数据聚合 聚合（aggregations）可以让我们极其方便的实现对数据的统计、分析、运算。例如： 什么品牌的手机最受欢迎？ 这些手机的平均价格、最高价格、最低价格？ 这些手机每月的销售情况如何？ 实现这些统计功能的比数据库的sql要方便的多，而且查询速度非常快，可以实现近实时搜索效果。 1.1.聚合的种类 聚合常见的有三类： 桶（Bucket）聚合：用来对文档做分组 TermAggregation：按照文档字段值分组，例如按照品牌值分组、按照国家分组 Date Histogram：按照日期阶梯分组，例如一周为一组，或者一月为一组 度量（Metric）聚合：用以计算一些值，比如：最大值、最小值、平均值等 Avg：求平均值 Max：求最大值 Min：求最小值 Stats：同时求max、min、avg、sum等 管道（pipeline）聚合：其它聚合的结果为基础做聚合 注意：参加聚合的字段必须是keyword、日期、数值、布尔类型 1.2.DSL实现聚合 现在，我们要统计所有数据中的酒店品牌有几种，其实就是按照品牌对数据分组。此时可以根据酒店品牌的名称做聚合，也就是Bucket聚合。 1.2.1.Bucket聚合语法 语法如下： GET /hotel/_search { "size": 0, // 设置size为0，结果中不包含文档，只包含聚合结果 "aggs": { // 定义聚合 "brandAgg": { //给聚合起个名字 "terms": { // 聚合的类型，按照品牌值聚合，所以选择term "field": "brand", // 参与聚合的字段 "size": 20 // 希望获取的聚合结果数量 } } } } 结果如图： 1.2.2.聚合结果排序 默认情况下，Bucket聚合会统计Bucket内的文档数量，记为_count，并且按照_count降序排序。 我们可以指定order属性，自定义聚合的排序方式： GET /hotel/_search { "size": 0, "aggs": { "brandAgg": { "terms": { "field": "brand", "order": { "_count": "asc" // 按照_count升序排列 }, "size": 20 } } } } 1.2.3.限定聚合范围 默认情况下，Bucket聚合是对索引库的所有文档做聚合，但真实场景下，用户会输入搜索条件，因此聚合必须是对搜索结果聚合。那么聚合必须添加限定条件。 我们可以限定要聚合的文档范围，只要添加query条件即可： GET /hotel/_search { "query": { "range": { "price": { "lte": 200 // 只对200元以下的文档聚合 } } }, "size": 0, "aggs": { "brandAgg": { "terms": { "field": "brand", "size": 20 } } } } 这次，聚合得到的品牌明显变少了： 1.2.4.Metric聚合语法 上节课，我们对酒店按照品牌分组，形成了一个个桶。现在我们需要对桶内的酒店做运算，获取每个品牌的用户评分的min、max、avg等值。 这就要用到Metric聚合了，例如stat聚合：就可以获取min、max、avg等结果。 语法如下： GET /hotel/_search { "size": 0, "aggs": { "brandAgg": { "terms": { "field": "brand", "size": 20 }, "aggs": { // 是brands聚合的子聚合，也就是分组后对每组分别计算 "score_stats": { // 聚合名称 "stats": { // 聚合类型，这里stats可以计算min、max、avg等 "field": "score" // 聚合字段，这里是score } } } } } } 这次的score_stats聚合是在brandAgg的聚合内部嵌套的子聚合。因为我们需要在每个桶分别计算。 另外，我们还可以给聚合结果做个排序，例如按照每个桶的酒店平均分做排序： 1.2.5.小结 aggs代表聚合，与query同级，此时query的作用是？ 限定聚合的的文档范围 聚合必须的三要素： 聚合名称 聚合类型 聚合字段 聚合可配置属性有： size：指定聚合结果数量 order：指定聚合结果排序方式 field：指定聚合字段 1.3.RestAPI实现聚合 1.3.1.API语法 聚合条件与query条件同级别，因此需要使用request.source()来指定聚合条件。 聚合条件的语法： 聚合的结果也与查询结果不同，API也比较特殊。不过同样是JSON逐层解析： 1.3.2.业务需求 需求：搜索页面的品牌、城市等信息不应该是在页面写死，而是通过聚合索引库中的酒店数据得来的： 分析： 目前，页面的城市列表、星级列表、品牌列表都是写死的，并不会随着搜索结果的变化而变化。但是用户搜索条件改变时，搜索结果会跟着变化。 例如：用户搜索“东方明珠”，那搜索的酒店肯定是在上海东方明珠附近，因此，城市只能是上海，此时城市列表中就不应该显示北京、深圳、杭州这些信息了。 也就是说，搜索结果中包含哪些城市，页面就应该列出哪些城市；搜索结果中包含哪些品牌，页面就应该列出哪些品牌。 如何得知搜索结果中包含哪些品牌？如何得知搜索结果中包含哪些城市？ 使用聚合功能，利用Bucket聚合，对搜索结果中的文档基于品牌分组、基于城市分组，就能得知包含哪些品牌、哪些城市了。 因为是对搜索结果聚合，因此聚合是限定范围的聚合，也就是说聚合的限定条件跟搜索文档的条件一致。 查看浏览器可以发现，前端其实已经发出了这样的一个请求： 请求参数与搜索文档的参数完全一致。 返回值类型就是页面要展示的最终结果： 结果是一个Map结构： key是字符串，城市、星级、品牌、价格 value是集合，例如多个城市的名称 1.3.3.业务实现 在cn.wcy.hotel.web包的HotelController中添加一个方法，遵循下面的要求： 请求方式：POST 请求路径：/hotel/filters 请求参数：RequestParams，与搜索文档的参数一致 返回值类型：Map<String, List<String>> 代码： @PostMapping("filters") public Map<String, List<String>> getFilters(@RequestBody RequestParams params){ return hotelService.getFilters(params); } 这里调用了IHotelService中的getFilters方法，尚未实现。 在cn.wcy.hotel.service.IHotelService中定义新方法： Map<String, List<String>> filters(RequestParams params); 在cn.wcy.hotel.service.impl.HotelService中实现该方法： @Override public Map<String, List<String>> filters(RequestParams params) { try { // 1.准备Request SearchRequest request = new SearchRequest("hotel"); // 2.准备DSL // 2.1.query buildBasicQuery(params, request); // 2.2.设置size request.source().size(0); // 2.3.聚合 buildAggregation(request); // 3.发出请求 SearchResponse response = client.search(request, RequestOptions.DEFAULT); // 4.解析结果 Map<String, List<String>> result = new HashMap<>(); Aggregations aggregations = response.getAggregations(); // 4.1.根据品牌名称，获取品牌结果 List<String> brandList = getAggByName(aggregations, "brandAgg"); result.put("品牌", brandList); // 4.2.根据品牌名称，获取品牌结果 List<String> cityList = getAggByName(aggregations, "cityAgg"); result.put("城市", cityList); // 4.3.根据品牌名称，获取品牌结果 List<String> starList = getAggByName(aggregations, "starAgg"); result.put("星级", starList); return result; } catch (IOException e) { throw new RuntimeException(e); } } private void buildAggregation(SearchRequest request) { request.source().aggregation(AggregationBuilders .terms("brandAgg") .field("brand") .size(100) ); request.source().aggregation(AggregationBuilders .terms("cityAgg") .field("city") .size(100) ); request.source().aggregation(AggregationBuilders .terms("starAgg") .field("starName") .size(100) ); } private List<String> getAggByName(Aggregations aggregations, String aggName) { // 4.1.根据聚合名称获取聚合结果 Terms brandTerms = aggregations.get(aggName); // 4.2.获取buckets List<? extends Terms.Bucket> buckets = brandTerms.getBuckets(); // 4.3.遍历 List<String> brandList = new ArrayList<>(); for (Terms.Bucket bucket : buckets) { // 4.4.获取key String key = bucket.getKeyAsString(); brandList.add(key); } return brandList; } 2.自动补全 当用户在搜索框输入字符时，我们应该提示出与该字符有关的搜索项，如图： 这种根据用户输入的字母，提示完整词条的功能，就是自动补全了。 因为需要根据拼音字母来推断，因此要用到拼音分词功能。 2.1.拼音分词器 要实现根据字母做补全，就必须对文档按照拼音分词。在GitHub上恰好有elasticsearch的拼音分词插件。地址：https://github.com/medcl/elasticsearch-analysis-pinyin 安装方式与IK分词器一样，分三步： ​ ①解压 ​ ②上传到虚拟机中，elasticsearch的plugin目录 ​ ③重启elasticsearch ​ ④测试 详细安装步骤可以参考IK分词器的安装过程。 测试用法如下： POST /_analyze { "text": "如家酒店还不错", "analyzer": "pinyin" } 结果： 2.2.自定义分词器 默认的拼音分词器会将每个汉字单独分为拼音，而我们希望的是每个词条形成一组拼音，需要对拼音分词器做个性化定制，形成自定义分词器。 elasticsearch中分词器（analyzer）的组成包含三部分： character filters：在tokenizer之前对文本进行处理。例如删除字符、替换字符 tokenizer：将文本按照一定的规则切割成词条（term）。例如keyword，就是不分词；还有ik_smart tokenizer filter：将tokenizer输出的词条做进一步处理。例如大小写转换、同义词处理、拼音处理等 文档分词时会依次由这三部分来处理文档： 声明自定义分词器的语法如下： PUT /test { "settings": { "analysis": { "analyzer": { // 自定义分词器 "my_analyzer": { // 分词器名称 "tokenizer": "ik_max_word", "filter": "py" } }, "filter": { // 自定义tokenizer filter "py": { // 过滤器名称 "type": "pinyin", // 过滤器类型，这里是pinyin "keep_full_pinyin": false, "keep_joined_full_pinyin": true, "keep_original": true, "limit_first_letter_length": 16, "remove_duplicated_term": true, "none_chinese_pinyin_tokenize": false } } } }, "mappings": { "properties": { "name": { "type": "text", "analyzer": "my_analyzer", "search_analyzer": "ik_smart" } } } } 测试： 总结： 如何使用拼音分词器？ ①下载pinyin分词器 ②解压并放到elasticsearch的plugin目录 ③重启即可 如何自定义分词器？ ①创建索引库时，在settings中配置，可以包含三部分 ②character filter ③tokenizer ④filter 拼音分词器注意事项？ 为了避免搜索到同音字，搜索时不要使用拼音分词器 2.3.自动补全查询 elasticsearch提供了Completion Suggester查询来实现自动补全功能。这个查询会匹配以用户输入内容开头的词条并返回。为了提高补全查询的效率，对于文档中字段的类型有一些约束： 参与补全查询的字段必须是completion类型。 字段的内容一般是用来补全的多个词条形成的数组。 比如，一个这样的索引库： // 创建索引库 PUT test { "mappings": { "properties": { "title":{ "type": "completion" } } } } 然后插入下面的数据： // 示例数据 POST test/_doc { "title": ["Sony", "WH-1000XM3"] } POST test/_doc { "title": ["SK-II", "PITERA"] } POST test/_doc { "title": ["Nintendo", "switch"] } 查询的DSL语句如下： // 自动补全查询 GET /test/_search { "suggest": { "title_suggest": { "text": "s", // 关键字 "completion": { "field": "title", // 补全查询的字段 "skip_duplicates": true, // 跳过重复的 "size": 10 // 获取前10条结果 } } } } 2.4.实现酒店搜索框自动补全 现在，我们的hotel索引库还没有设置拼音分词器，需要修改索引库中的配置。但是我们知道索引库是无法修改的，只能删除然后重新创建。 另外，我们需要添加一个字段，用来做自动补全，将brand、suggestion、city等都放进去，作为自动补全的提示。 因此，总结一下，我们需要做的事情包括： 修改hotel索引库结构，设置自定义拼音分词器 修改索引库的name、all字段，使用自定义分词器 索引库添加一个新字段suggestion，类型为completion类型，使用自定义的分词器 给HotelDoc类添加suggestion字段，内容包含brand、business 重新导入数据到hotel库 2.4.1.修改酒店映射结构 代码如下： // 酒店数据索引库 PUT /hotel { "settings": { "analysis": { "analyzer": { "text_anlyzer": { "tokenizer": "ik_max_word", "filter": "py" }, "completion_analyzer": { "tokenizer": "keyword", "filter": "py" } }, "filter": { "py": { "type": "pinyin", "keep_full_pinyin": false, "keep_joined_full_pinyin": true, "keep_original": true, "limit_first_letter_length": 16, "remove_duplicated_term": true, "none_chinese_pinyin_tokenize": false } } } }, "mappings": { "properties": { "id":{ "type": "keyword" }, "name":{ "type": "text", "analyzer": "text_anlyzer", "search_analyzer": "ik_smart", "copy_to": "all" }, "address":{ "type": "keyword", "index": false }, "price":{ "type": "integer" }, "score":{ "type": "integer" }, "brand":{ "type": "keyword", "copy_to": "all" }, "city":{ "type": "keyword" }, "starName":{ "type": "keyword" }, "business":{ "type": "keyword", "copy_to": "all" }, "location":{ "type": "geo_point" }, "pic":{ "type": "keyword", "index": false }, "all":{ "type": "text", "analyzer": "text_anlyzer", "search_analyzer": "ik_smart" }, "suggestion":{ "type": "completion", "analyzer": "completion_analyzer" } } } } 2.4.2.修改HotelDoc实体 HotelDoc中要添加一个字段，用来做自动补全，内容可以是酒店品牌、城市、商圈等信息。按照自动补全字段的要求，最好是这些字段的数组。 因此我们在HotelDoc中添加一个suggestion字段，类型为List<String>，然后将brand、city、business等信息放到里面。 代码如下： package cn.wcy.hotel.pojo; import lombok.Data; import lombok.NoArgsConstructor; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; @Data @NoArgsConstructor public class HotelDoc { private Long id; private String name; private String address; private Integer price; private Integer score; private String brand; private String city; private String starName; private String business; private String location; private String pic; private Object distance; private Boolean isAD; private List<String> suggestion; public HotelDoc(Hotel hotel) { this.id = hotel.getId(); this.name = hotel.getName(); this.address = hotel.getAddress(); this.price = hotel.getPrice(); this.score = hotel.getScore(); this.brand = hotel.getBrand(); this.city = hotel.getCity(); this.starName = hotel.getStarName(); this.business = hotel.getBusiness(); this.location = hotel.getLatitude() + ", " + hotel.getLongitude(); this.pic = hotel.getPic(); // 组装suggestion if(this.business.contains("/")){ // business有多个值，需要切割 String[] arr = this.business.split("/"); // 添加元素 this.suggestion = new ArrayList<>(); this.suggestion.add(this.brand); Collections.addAll(this.suggestion, arr); }else { this.suggestion = Arrays.asList(this.brand, this.business); } } } 2.4.3.重新导入 重新执行之前编写的导入数据功能，可以看到新的酒店数据中包含了suggestion： 2.4.4.自动补全查询的JavaAPI 之前我们学习了自动补全查询的DSL，而没有学习对应的JavaAPI，这里给出一个示例： 而自动补全的结果也比较特殊，解析的代码如下： 2.4.5.实现搜索框自动补全 查看前端页面，可以发现当我们在输入框键入时，前端会发起ajax请求： 返回值是补全词条的集合，类型为List<String> 1）在cn.wcy.hotel.web包下的HotelController中添加新接口，接收新的请求： @GetMapping("suggestion") public List<String> getSuggestions(@RequestParam("key") String prefix) { return hotelService.getSuggestions(prefix); } 2）在cn.wcy.hotel.service包下的IhotelService中添加方法： List<String> getSuggestions(String prefix); 3）在cn.wcy.hotel.service.impl.HotelService中实现该方法： @Override public List<String> getSuggestions(String prefix) { try { // 1.准备Request SearchRequest request = new SearchRequest("hotel"); // 2.准备DSL request.source().suggest(new SuggestBuilder().addSuggestion( "suggestions", SuggestBuilders.completionSuggestion("suggestion") .prefix(prefix) .skipDuplicates(true) .size(10) )); // 3.发起请求 SearchResponse response = client.search(request, RequestOptions.DEFAULT); // 4.解析结果 Suggest suggest = response.getSuggest(); // 4.1.根据补全查询名称，获取补全结果 CompletionSuggestion suggestions = suggest.getSuggestion("suggestions"); // 4.2.获取options List<CompletionSuggestion.Entry.Option> options = suggestions.getOptions(); // 4.3.遍历 List<String> list = new ArrayList<>(options.size()); for (CompletionSuggestion.Entry.Option option : options) { String text = option.getText().toString(); list.add(text); } return list; } catch (IOException e) { throw new RuntimeException(e); } } 3.数据同步 elasticsearch中的酒店数据来自于mysql数据库，因此mysql数据发生改变时，elasticsearch也必须跟着改变，这个就是elasticsearch与mysql之间的数据同步。 3.1.思路分析 常见的数据同步方案有三种： 同步调用 异步通知 监听binlog 3.1.1.同步调用 方案一：同步调用 基本步骤如下： hotel-demo对外提供接口，用来修改elasticsearch中的数据 酒店管理服务在完成数据库操作后，直接调用hotel-demo提供的接口， 3.1.2.异步通知 方案二：异步通知 流程如下： hotel-admin对mysql数据库数据完成增、删、改后，发送MQ消息 hotel-demo监听MQ，接收到消息后完成elasticsearch数据修改 3.1.3.监听binlog 方案三：监听binlog 流程如下： 给mysql开启binlog功能 mysql完成增、删、改操作都会记录在binlog中 hotel-demo基于canal监听binlog变化，实时更新elasticsearch中的内容 3.1.4.选择 方式一：同步调用 优点：实现简单，粗暴 缺点：业务耦合度高 方式二：异步通知 优点：低耦合，实现难度一般 缺点：依赖mq的可靠性 方式三：监听binlog 优点：完全解除服务间耦合 缺点：开启binlog增加数据库负担、实现复杂度高 3.2.实现数据同步 3.2.1.思路 步骤： hotel-admin项目，启动并测试酒店数据的CRUD 声明exchange、queue、RoutingKey 在hotel-admin中的增、删、改业务中完成消息发送 在hotel-demo中完成消息监听，并更新elasticsearch中数据 启动并测试数据同步功能 3.2.2.打开项目 运行后，访问 http://localhost:8099 其中包含了酒店的CRUD功能： 3.2.3.声明交换机、队列 MQ结构如图： 1）引入依赖 在hotel-admin、hotel-demo中引入rabbitmq的依赖： <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> 2）声明队列交换机名称 在hotel-admin和hotel-demo中的cn.wcy.hotel.constatnts包下新建一个类MqConstants： package cn.wcy.hotel.constatnts; public class MqConstants { /** * 交换机 */ public final static String HOTEL_EXCHANGE = "hotel.topic"; /** * 监听新增和修改的队列 */ public final static String HOTEL_INSERT_QUEUE = "hotel.insert.queue"; /** * 监听删除的队列 */ public final static String HOTEL_DELETE_QUEUE = "hotel.delete.queue"; /** * 新增或修改的RoutingKey */ public final static String HOTEL_INSERT_KEY = "hotel.insert"; /** * 删除的RoutingKey */ public final static String HOTEL_DELETE_KEY = "hotel.delete"; } 3）声明队列交换机 在hotel-demo中，定义配置类，声明队列、交换机： package cn.wcy.hotel.config; import cn.wcy.hotel.constants.MqConstants; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.TopicExchange; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class MqConfig { @Bean public TopicExchange topicExchange(){ return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false); } @Bean public Queue insertQueue(){ return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true); } @Bean public Queue deleteQueue(){ return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true); } @Bean public Binding insertQueueBinding(){ return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY); } @Bean public Binding deleteQueueBinding(){ return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY); } } 3.2.4.发送MQ消息 在hotel-admin中的增、删、改业务中分别发送MQ消息： 3.2.5.接收MQ消息 hotel-demo接收到MQ消息要做的事情包括： 新增消息：根据传递的hotel的id查询hotel信息，然后新增一条数据到索引库 删除消息：根据传递的hotel的id删除索引库中的一条数据 1）首先在hotel-demo的cn.wcy.hotel.service包下的IHotelService中新增新增、删除业务 void deleteById(Long id); void insertById(Long id); 2）给hotel-demo中的cn.wcy.hotel.service.impl包下的HotelService中实现业务： @Override public void deleteById(Long id) { try { // 1.准备Request DeleteRequest request = new DeleteRequest("hotel", id.toString()); // 2.发送请求 client.delete(request, RequestOptions.DEFAULT); } catch (IOException e) { throw new RuntimeException(e); } } @Override public void insertById(Long id) { try { // 0.根据id查询酒店数据 Hotel hotel = getById(id); // 转换为文档类型 HotelDoc hotelDoc = new HotelDoc(hotel); // 1.准备Request对象 IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString()); // 2.准备Json文档 request.source(JSON.toJSONString(hotelDoc), XContentType.JSON); // 3.发送请求 client.index(request, RequestOptions.DEFAULT); } catch (IOException e) { throw new RuntimeException(e); } } 3）编写监听器 在hotel-demo中的cn.wcy.hotel.mq包新增一个类： package cn.wcy.hotel.mq; import cn.wcy.hotel.constants.MqConstants; import cn.wcy.hotel.service.IHotelService; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @Component public class HotelListener { @Autowired private IHotelService hotelService; /** * 监听酒店新增或修改的业务 * @param id 酒店id */ @RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE) public void listenHotelInsertOrUpdate(Long id){ hotelService.insertById(id); } /** * 监听酒店删除的业务 * @param id 酒店id */ @RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE) public void listenHotelDelete(Long id){ hotelService.deleteById(id); } } 4.集群 单机的elasticsearch做数据存储，必然面临两个问题：海量数据存储问题、单点故障问题。 海量数据存储问题：将索引库从逻辑上拆分为N个分片（shard），存储到多个节点 单点故障问题：将分片数据在不同节点备份（replica ） ES集群相关概念: 集群（cluster）：一组拥有共同的 cluster name 的 节点。 节点（node) ：集群中的一个 Elasticearch 实例 分片（shard）：索引可以被拆分为不同的部分进行存储，称为分片。在集群环境下，一个索引的不同分片可以拆分到不同的节点中 解决问题：数据量太大，单点存储量有限的问题。 此处，我们把数据分成3片：shard0、shard1、shard2 主分片（Primary shard）：相对于副本分片的定义。 副本分片（Replica shard）每个主分片可以有一个或者多个副本，数据和主分片一样。 ​ 数据备份可以保证高可用，但是每个分片备份一份，所需要的节点数量就会翻一倍，成本实在是太高了！ 为了在高可用和成本间寻求平衡，我们可以这样做： 首先对数据分片，存储到不同节点 然后对每个分片进行备份，放到对方节点，完成互相备份 这样可以大大减少所需要的服务节点数量，如图，我们以3分片，每个分片备份一份为例： 现在，每个分片都有1个备份，存储在3个节点： node0：保存了分片0和1 node1：保存了分片0和2 node2：保存了分片1和2 4.1.搭建ES集群 预先安装es 4.2.集群脑裂问题 4.2.1.集群职责划分 elasticsearch中集群节点有不同的职责划分： 默认情况下，集群中的任何一个节点都同时具备上述四种角色。 但是真实的集群一定要将集群职责分离： master节点：对CPU要求高，但是内存要求第 data节点：对CPU和内存要求都高 coordinating节点：对网络带宽、CPU要求高 职责分离可以让我们根据不同节点的需求分配不同的硬件去部署。而且避免业务之间的互相干扰。 一个典型的es集群职责划分如图： 4.2.2.脑裂问题 脑裂是因为集群中的节点失联导致的。 例如一个集群中，主节点与其它节点失联： 此时，node2和node3认为node1宕机，就会重新选主： 当node3当选后，集群继续对外提供服务，node2和node3自成集群，node1自成集群，两个集群数据不同步，出现数据差异。 当网络恢复后，因为集群中有两个master节点，集群状态的不一致，出现脑裂的情况： 解决脑裂的方案是，要求选票超过 ( eligible节点数量 + 1 ）/ 2 才能当选为主，因此eligible节点数量最好是奇数。对应配置项是discovery.zen.minimum_master_nodes，在es7.0以后，已经成为默认配置，因此一般不会发生脑裂问题 例如：3个节点形成的集群，选票必须超过 （3 + 1） / 2 ，也就是2票。node3得到node2和node3的选票，当选为主。node1只有自己1票，没有当选。集群中依然只有1个主节点，没有出现脑裂。 4.2.3.小结 master eligible节点的作用是什么？ 参与集群选主 主节点可以管理集群状态、管理分片信息、处理创建和删除索引库的请求 data节点的作用是什么？ 数据的CRUD coordinator节点的作用是什么？ 路由请求到其它节点 合并查询到的结果，返回给用户 4.3.集群分布式存储 当新增文档时，应该保存到不同分片，保证数据均衡，那么coordinating node如何确定数据该存储到哪个分片呢？ 4.3.1.分片存储测试 插入三条数据： 测试可以看到，三条数据分别在不同分片： 结果： 4.3.2.分片存储原理 elasticsearch会通过hash算法来计算文档应该存储到哪个分片： 说明： _routing默认是文档的id 算法与分片数量有关，因此索引库一旦创建，分片数量不能修改！ 新增文档的流程如下： 解读： 1）新增一个id=1的文档 2）对id做hash运算，假如得到的是2，则应该存储到shard-2 3）shard-2的主分片在node3节点，将数据路由到node3 4）保存文档 5）同步给shard-2的副本replica-2，在node2节点 6）返回结果给coordinating-node节点 4.4.集群分布式查询 elasticsearch的查询分成两个阶段： scatter phase：分散阶段，coordinating node会把请求分发到每一个分片 gather phase：聚集阶段，coordinating node汇总data node的搜索结果，并处理为最终结果集返回给用户 4.5.集群故障转移 集群的master节点会监控集群中的节点状态，如果发现有节点宕机，会立即将宕机节点的分片数据迁移到其它节点，确保数据安全，这个叫做故障转移。 1）例如一个集群结构如图： 现在，node1是主节点，其它两个节点是从节点。 2）突然，node1发生了故障： 宕机后的第一件事，需要重新选主，例如选中了node2： node2成为主节点后，会检测集群监控状态，发现：shard-1、shard-0没有副本节点。因此需要将node1上的数据迁移到node2、node3：
分布式搜索引擎 ElasticSearch 在昨天的学习中，我们已经导入了大量数据到elasticsearch中，实现了elasticsearch的数据存储功能。但elasticsearch最擅长的还是搜索和数据分析。 所以今天，我们研究下elasticsearch的数据搜索功能。我们会分别使用DSL和RestClient实现搜索。 学习目标 1.DSL查询文档 elasticsearch的查询依然是基于JSON风格的DSL来实现的。 1.1.DSL查询分类 Elasticsearch提供了基于JSON的DSL（Domain Specific Language）来定义查询。常见的查询类型包括： 查询所有：查询出所有数据，一般测试用。例如：match_all 全文检索（full text）查询：利用分词器对用户输入内容分词，然后去倒排索引库中匹配。例如： match_query multi_match_query 精确查询：根据精确词条值查找数据，一般是查找keyword、数值、日期、boolean等类型字段。例如： ids range term 地理（geo）查询：根据经纬度查询。例如： geo_distance geo_bounding_box 复合（compound）查询：复合查询可以将上述各种查询条件组合起来，合并查询条件。例如： bool function_score 查询的语法基本一致： GET /indexName/_search { "query": { "查询类型": { "查询条件": "条件值" } } } 我们以查询所有为例，其中： 查询类型为match_all 没有查询条件 // 查询所有 GET /indexName/_search { "query": { "match_all": { } } } 其它查询无非就是查询类型、查询条件的变化。 1.2.全文检索查询 1.2.1.使用场景 全文检索查询的基本流程如下： 对用户搜索的内容做分词，得到词条 根据词条去倒排索引库中匹配，得到文档id 根据文档id找到文档，返回给用户 比较常用的场景包括： 商城的输入框搜索 百度输入框搜索 例如京东： 因为是拿着词条去匹配，因此参与搜索的字段也必须是可分词的text类型的字段。 1.2.2.基本语法 常见的全文检索查询包括： match查询：单字段查询 multi_match查询：多字段查询，任意一个字段符合条件就算符合查询条件 match查询语法如下： GET /indexName/_search { "query": { "match": { "FIELD": "TEXT" } } } mulit_match语法如下： GET /indexName/_search { "query": { "multi_match": { "query": "TEXT", "fields": ["FIELD1", " FIELD12"] } } } 1.2.3.示例 match查询示例： multi_match查询示例： 可以看到，两种查询结果是一样的，为什么？ 因为我们将brand、name、business值都利用copy_to复制到了all字段中。因此你根据三个字段搜索，和根据all字段搜索效果当然一样了。 但是，搜索字段越多，对查询性能影响越大，因此建议采用copy_to，然后单字段查询的方式。 1.2.4.总结 match和multi_match的区别是什么？ match：根据一个字段查询 multi_match：根据多个字段查询，参与查询字段越多，查询性能越差 1.3.精准查询 精确查询一般是查找keyword、数值、日期、boolean等类型字段。所以不会对搜索条件分词。常见的有： term：根据词条精确值查询 range：根据值的范围查询 1.3.1.term查询 因为精确查询的字段搜是不分词的字段，因此查询的条件也必须是不分词的词条。查询时，用户输入的内容跟自动值完全匹配时才认为符合条件。如果用户输入的内容过多，反而搜索不到数据。 语法说明： // term查询 GET /indexName/_search { "query": { "term": { "FIELD": { "value": "VALUE" } } } } 示例： 当我搜索的是精确词条时，能正确查询出结果： 但是，当我搜索的内容不是词条，而是多个词语形成的短语时，反而搜索不到： 1.3.2.range查询 范围查询，一般应用在对数值类型做范围过滤的时候。比如做价格范围过滤。 基本语法： // range查询 GET /indexName/_search { "query": { "range": { "FIELD": { "gte": 10, // 这里的gte代表大于等于，gt则代表大于 "lte": 20 // lte代表小于等于，lt则代表小于 } } } } 示例： 1.3.3.总结 精确查询常见的有哪些？ term查询：根据词条精确匹配，一般搜索keyword类型、数值类型、布尔类型、日期类型字段 range查询：根据数值范围查询，可以是数值、日期的范围 1.4.地理坐标查询 所谓的地理坐标查询，其实就是根据经纬度查询，官方文档：https://www.elastic.co/guide/en/elasticsearch/reference/current/geo-queries.html 常见的使用场景包括： 携程：搜索我附近的酒店 滴滴：搜索我附近的出租车 微信：搜索我附近的人 附近的酒店： 附近的车： 1.4.1.矩形范围查询 矩形范围查询，也就是geo_bounding_box查询，查询坐标落在某个矩形范围的所有文档： 查询时，需要指定矩形的左上、右下两个点的坐标，然后画出一个矩形，落在该矩形内的都是符合条件的点。 语法如下： // geo_bounding_box查询 GET /indexName/_search { "query": { "geo_bounding_box": { "FIELD": { "top_left": { // 左上点 "lat": 31.1, "lon": 121.5 }, "bottom_right": { // 右下点 "lat": 30.9, "lon": 121.7 } } } } } 这种并不符合“附近的人”这样的需求，所以我们就不做了。 1.4.2.附近查询 附近查询，也叫做距离查询（geo_distance）：查询到指定中心点小于某个距离值的所有文档。 换句话来说，在地图上找一个点作为圆心，以指定距离为半径，画一个圆，落在圆内的坐标都算符合条件： 语法说明： // geo_distance 查询 GET /indexName/_search { "query": { "geo_distance": { "distance": "15km", // 半径 "FIELD": "31.21,121.5" // 圆心 } } } 示例： 我们先搜索陆家嘴附近15km的酒店： 发现共有47家酒店。 然后把半径缩短到3公里： 可以发现，搜索到的酒店数量减少到了5家。 1.5.复合查询 复合（compound）查询：复合查询可以将其它简单查询组合起来，实现更复杂的搜索逻辑。常见的有两种： fuction score：算分函数查询，可以控制文档相关性算分，控制文档排名 bool query：布尔查询，利用逻辑关系组合多个其它的查询，实现复杂搜索 1.5.1.相关性算分 当我们利用match查询时，文档结果会根据与搜索词条的关联度打分（_score），返回结果时按照分值降序排列。 例如，我们搜索 "虹桥如家"，结果如下： [ { "_score" : 17.850193, "_source" : { "name" : "虹桥如家酒店真不错", } }, { "_score" : 12.259849, "_source" : { "name" : "外滩如家酒店真不错", } }, { "_score" : 11.91091, "_source" : { "name" : "迪士尼如家酒店真不错", } } ] 在elasticsearch中，早期使用的打分算法是TF-IDF算法，公式如下： 在后来的5.1版本升级中，elasticsearch将算法改进为BM25算法，公式如下： TF-IDF算法有一各缺陷，就是词条频率越高，文档得分也会越高，单个词条对文档影响较大。而BM25则会让单个词条的算分有一个上限，曲线更加平滑： 小结：elasticsearch会根据词条和文档的相关度做打分，算法由两种： TF-IDF算法 BM25算法，elasticsearch5.1版本后采用的算法 1.5.2.算分函数查询 根据相关度打分是比较合理的需求，但合理的不一定是产品经理需要的。 以百度为例，你搜索的结果中，并不是相关度越高排名越靠前，而是谁掏的钱多排名就越靠前。如图： 要想认为控制相关性算分，就需要利用elasticsearch中的function score 查询了。 1）语法说明 function score 查询中包含四部分内容： 原始查询条件：query部分，基于这个条件搜索文档，并且基于BM25算法给文档打分，原始算分（query score) 过滤条件：filter部分，符合该条件的文档才会重新算分 算分函数：符合filter条件的文档要根据这个函数做运算，得到的函数算分（function score），有四种函数 weight：函数结果是常量 field_value_factor：以文档中的某个字段值作为函数结果 random_score：以随机数作为函数结果 script_score：自定义算分函数算法 运算模式：算分函数的结果、原始查询的相关性算分，两者之间的运算方式，包括： multiply：相乘 replace：用function score替换query score 其它，例如：sum、avg、max、min function score的运行流程如下： 1）根据原始条件查询搜索文档，并且计算相关性算分，称为原始算分（query score） 2）根据过滤条件，过滤文档 3）符合过滤条件的文档，基于算分函数运算，得到函数算分（function score） 4）将原始算分（query score）和函数算分（function score）基于运算模式做运算，得到最终结果，作为相关性算分。 因此，其中的关键点是： 过滤条件：决定哪些文档的算分被修改 算分函数：决定函数算分的算法 运算模式：决定最终算分结果 2）示例 需求：给“如家”这个品牌的酒店排名靠前一些 翻译一下这个需求，转换为之前说的四个要点： 原始条件：不确定，可以任意变化 过滤条件：brand = "如家" 算分函数：可以简单粗暴，直接给固定的算分结果，weight 运算模式：比如求和 因此最终的DSL语句如下： GET /hotel/_search { "query": { "function_score": { "query": { .... }, // 原始查询，可以是任意条件 "functions": [ // 算分函数 { "filter": { // 满足的条件，品牌必须是如家 "term": { "brand": "如家" } }, "weight": 2 // 算分权重为2 } ], "boost_mode": "sum" // 加权模式，求和 } } } 测试，在未添加算分函数时，如家得分如下： 添加了算分函数后，如家得分就提升了： 3）小结 function score query定义的三要素是什么？ 过滤条件：哪些文档要加分 算分函数：如何计算function score 加权方式：function score 与 query score如何运算 1.5.3.布尔查询 布尔查询是一个或多个查询子句的组合，每一个子句就是一个子查询。子查询的组合方式有： must：必须匹配每个子查询，类似“与” should：选择性匹配子查询，类似“或” must_not：必须不匹配，不参与算分，类似“非” filter：必须匹配，不参与算分 比如在搜索酒店时，除了关键字搜索外，我们还可能根据品牌、价格、城市等字段做过滤： 每一个不同的字段，其查询的条件、方式都不一样，必须是多个不同的查询，而要组合这些查询，就必须用bool查询了。 需要注意的是，搜索时，参与打分的字段越多，查询的性能也越差。因此这种多条件查询时，建议这样做： 搜索框的关键字搜索，是全文检索查询，使用must查询，参与算分 其它过滤条件，采用filter查询。不参与算分 1）语法示例： GET /hotel/_search { "query": { "bool": { "must": [ {"term": {"city": "上海" }} ], "should": [ {"term": {"brand": "皇冠假日" }}, {"term": {"brand": "华美达" }} ], "must_not": [ { "range": { "price": { "lte": 500 } }} ], "filter": [ { "range": {"score": { "gte": 45 } }} ] } } } 2）示例 需求：搜索名字包含“如家”，价格不高于400，在坐标31.21,121.5周围10km范围内的酒店。 分析： 名称搜索，属于全文检索查询，应该参与算分。放到must中 价格不高于400，用range查询，属于过滤条件，不参与算分。放到must_not中 周围10km范围内，用geo_distance查询，属于过滤条件，不参与算分。放到filter中 3）小结 bool查询有几种逻辑关系？ must：必须匹配的条件，可以理解为“与” should：选择性匹配的条件，可以理解为“或” must_not：必须不匹配的条件，不参与打分 filter：必须匹配的条件，不参与打分 2.搜索结果处理 搜索的结果可以按照用户指定的方式去处理或展示。 2.1.排序 elasticsearch默认是根据相关度算分（_score）来排序，但是也支持自定义方式对搜索结果排序。可以排序字段类型有：keyword类型、数值类型、地理坐标类型、日期类型等。 2.1.1.普通字段排序 keyword、数值、日期类型排序的语法基本一致。 语法： GET /indexName/_search { "query": { "match_all": {} }, "sort": [ { "FIELD": "desc" // 排序字段、排序方式ASC、DESC } ] } 排序条件是一个数组，也就是可以写多个排序条件。按照声明的顺序，当第一个条件相等时，再按照第二个条件排序，以此类推 示例： 需求描述：酒店数据按照用户评价（score)降序排序，评价相同的按照价格(price)升序排序 2.1.2.地理坐标排序 地理坐标排序略有不同。 语法说明： GET /indexName/_search { "query": { "match_all": {} }, "sort": [ { "_geo_distance" : { "FIELD" : "纬度，经度", // 文档中geo_point类型的字段名、目标坐标点 "order" : "asc", // 排序方式 "unit" : "km" // 排序的距离单位 } } ] } 这个查询的含义是： 指定一个坐标，作为目标点 计算每一个文档中，指定字段（必须是geo_point类型）的坐标 到目标点的距离是多少 根据距离排序 示例： 需求描述：实现对酒店数据按照到你的位置坐标的距离升序排序 提示：获取你的位置的经纬度的方式：https://lbs.amap.com/demo/jsapi-v2/example/map/click-to-get-lnglat/ 假设我的位置是：31.034661，121.612282，寻找我周围距离最近的酒店。 2.2.分页 elasticsearch 默认情况下只返回top10的数据。而如果要查询更多数据就需要修改分页参数了。elasticsearch中通过修改from、size参数来控制要返回的分页结果： from：从第几个文档开始 size：总共查询几个文档 类似于mysql中的limit ?, ? 2.2.1.基本的分页 分页的基本语法如下： GET /hotel/_search { "query": { "match_all": {} }, "from": 0, // 分页开始的位置，默认为0 "size": 10, // 期望获取的文档总数 "sort": [ {"price": "asc"} ] } 2.2.2.深度分页问题 现在，我要查询990~1000的数据，查询逻辑要这么写： GET /hotel/_search { "query": { "match_all": {} }, "from": 990, // 分页开始的位置，默认为0 "size": 10, // 期望获取的文档总数 "sort": [ {"price": "asc"} ] } 这里是查询990开始的数据，也就是 第990~第1000条 数据。 不过，elasticsearch内部分页时，必须先查询 0~1000条，然后截取其中的990 ~ 1000的这10条： 查询TOP1000，如果es是单点模式，这并无太大影响。 但是elasticsearch将来一定是集群，例如我集群有5个节点，我要查询TOP1000的数据，并不是每个节点查询200条就可以了。 因为节点A的TOP200，在另一个节点可能排到10000名以外了。 因此要想获取整个集群的TOP1000，必须先查询出每个节点的TOP1000，汇总结果后，重新排名，重新截取TOP1000。 那如果我要查询9900~10000的数据呢？是不是要先查询TOP10000呢？那每个节点都要查询10000条？汇总到内存中？ 当查询分页深度较大时，汇总数据过多，对内存和CPU会产生非常大的压力，因此elasticsearch会禁止from+ size 超过10000的请求。 针对深度分页，ES提供了两种解决方案，官方文档： search after：分页时需要排序，原理是从上一次的排序值开始，查询下一页数据。官方推荐使用的方式。 scroll：原理将排序后的文档id形成快照，保存在内存。官方已经不推荐使用。 2.2.3.小结 分页查询的常见实现方案以及优缺点： from + size： 优点：支持随机翻页 缺点：深度分页问题，默认查询上限（from + size）是10000 场景：百度、京东、谷歌、淘宝这样的随机翻页搜索 after search： 优点：没有查询上限（单次查询的size不超过10000） 缺点：只能向后逐页查询，不支持随机翻页 场景：没有随机翻页需求的搜索，例如手机向下滚动翻页 scroll： 优点：没有查询上限（单次查询的size不超过10000） 缺点：会有额外内存消耗，并且搜索结果是非实时的 场景：海量数据的获取和迁移。从ES7.1开始不推荐，建议用 after search方案。 2.3.高亮 2.3.1.高亮原理 什么是高亮显示呢？ 我们在百度，京东搜索时，关键字会变成红色，比较醒目，这叫高亮显示： 高亮显示的实现分为两步： 1）给文档中的所有关键字都添加一个标签，例如标签 2）页面给标签编写CSS样式 2.3.2.实现高亮 高亮的语法： GET /hotel/_search { "query": { "match": { "FIELD": "TEXT" // 查询条件，高亮一定要使用全文检索查询 } }, "highlight": { "fields": { // 指定要高亮的字段 "FIELD": { "pre_tags": "", // 用来标记高亮字段的前置标签 "post_tags": "" // 用来标记高亮字段的后置标签 } } } } 注意： 高亮是对关键字高亮，因此搜索条件必须带有关键字，而不能是范围这样的查询。 默认情况下，高亮的字段，必须与搜索指定的字段一致，否则无法高亮 如果要对非搜索字段高亮，则需要添加一个属性：required_field_match=false 示例： 2.4.总结 查询的DSL是一个大的JSON对象，包含下列属性： query：查询条件 from和size：分页条件 sort：排序条件 highlight：高亮条件 示例： 3.RestClient查询文档 文档的查询同样适用昨天学习的 RestHighLevelClient对象，基本步骤包括： 1）准备Request对象 2）准备请求参数 3）发起请求 4）解析响应 3.1.快速入门 我们以match_all查询为例 3.1.1.发起查询请求 代码解读： 第一步，创建SearchRequest对象，指定索引库名 第二步，利用request.source()构建DSL，DSL中可以包含查询、分页、排序、高亮等 query()：代表查询条件，利用QueryBuilders.matchAllQuery()构建一个match_all查询的DSL 第三步，利用client.search()发送请求，得到响应 这里关键的API有两个，一个是request.source()，其中包含了查询、排序、分页、高亮等所有功能： 另一个是QueryBuilders，其中包含match、term、function_score、bool等各种查询： 3.1.2.解析响应 响应结果的解析： elasticsearch返回的结果是一个JSON字符串，结构包含： hits：命中的结果 total：总条数，其中的value是具体的总条数值 max_score：所有结果中得分最高的文档的相关性算分 hits：搜索结果的文档数组，其中的每个文档都是一个json对象 _source：文档中的原始数据，也是json对象 因此，我们解析响应结果，就是逐层解析JSON字符串，流程如下： SearchHits：通过response.getHits()获取，就是JSON中的最外层的hits，代表命中的结果 SearchHits#getTotalHits().value：获取总条数信息 SearchHits#getHits()：获取SearchHit数组，也就是文档数组 SearchHit#getSourceAsString()：获取文档结果中的_source，也就是原始的json文档数据 3.1.3.完整代码 完整代码如下： @Test void testMatchAll() throws IOException { // 1.准备Request SearchRequest request = new SearchRequest("hotel"); // 2.准备DSL request.source() .query(QueryBuilders.matchAllQuery()); // 3.发送请求 SearchResponse response = client.search(request, RequestOptions.DEFAULT); // 4.解析响应 handleResponse(response); } private void handleResponse(SearchResponse response) { // 4.解析响应 SearchHits searchHits = response.getHits(); // 4.1.获取总条数 long total = searchHits.getTotalHits().value; System.out.println("共搜索到" + total + "条数据"); // 4.2.文档数组 SearchHit[] hits = searchHits.getHits(); // 4.3.遍历 for (SearchHit hit : hits) { // 获取文档source String json = hit.getSourceAsString(); // 反序列化 HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class); System.out.println("hotelDoc = " + hotelDoc); } } 3.1.4.小结 查询的基本步骤是： 创建SearchRequest对象 准备Request.source()，也就是DSL。 ① QueryBuilders来构建查询条件 ② 传入Request.source() 的 query() 方法 发送请求，得到结果 解析结果（参考JSON结果，从外到内，逐层解析） 3.2.match查询 全文检索的match和multi_match查询与match_all的API基本一致。差别是查询条件，也就是query的部分。 因此，Java代码上的差异主要是request.source().query()中的参数了。同样是利用QueryBuilders提供的方法： 而结果解析代码则完全一致，可以抽取并共享。 完整代码如下： @Test void testMatch() throws IOException { // 1.准备Request SearchRequest request = new SearchRequest("hotel"); // 2.准备DSL request.source() .query(QueryBuilders.matchQuery("all", "如家")); // 3.发送请求 SearchResponse response = client.search(request, RequestOptions.DEFAULT); // 4.解析响应 handleResponse(response); } 3.3.精确查询 精确查询主要是两者： term：词条精确匹配 range：范围查询 与之前的查询相比，差异同样在查询条件，其它都一样。 查询条件构造的API如下： 3.4.布尔查询 布尔查询是用must、must_not、filter等方式组合其它查询，代码示例如下： 可以看到，API与其它查询的差别同样是在查询条件的构建，QueryBuilders，结果解析等其他代码完全不变。 完整代码如下： @Test void testBool() throws IOException { // 1.准备Request SearchRequest request = new SearchRequest("hotel"); // 2.准备DSL // 2.1.准备BooleanQuery BoolQueryBuilder boolQuery = QueryBuilders.boolQuery(); // 2.2.添加term boolQuery.must(QueryBuilders.termQuery("city", "杭州")); // 2.3.添加range boolQuery.filter(QueryBuilders.rangeQuery("price").lte(250)); request.source().query(boolQuery); // 3.发送请求 SearchResponse response = client.search(request, RequestOptions.DEFAULT); // 4.解析响应 handleResponse(response); } 3.5.排序、分页 搜索结果的排序和分页是与query同级的参数，因此同样是使用request.source()来设置。 对应的API如下： 完整代码示例： @Test void testPageAndSort() throws IOException { // 页码，每页大小 int page = 1, size = 5; // 1.准备Request SearchRequest request = new SearchRequest("hotel"); // 2.准备DSL // 2.1.query request.source().query(QueryBuilders.matchAllQuery()); // 2.2.排序 sort request.source().sort("price", SortOrder.ASC); // 2.3.分页 from、size request.source().from((page - 1) * size).size(5); // 3.发送请求 SearchResponse response = client.search(request, RequestOptions.DEFAULT); // 4.解析响应 handleResponse(response); } 3.6.高亮 高亮的代码与之前代码差异较大，有两点： 查询的DSL：其中除了查询条件，还需要添加高亮条件，同样是与query同级。 结果解析：结果除了要解析_source文档数据，还要解析高亮结果 3.6.1.高亮请求构建 高亮请求的构建API如下： 上述代码省略了查询条件部分，但是大家不要忘了：高亮查询必须使用全文检索查询，并且要有搜索关键字，将来才可以对关键字高亮。 完整代码如下： @Test void testHighlight() throws IOException { // 1.准备Request SearchRequest request = new SearchRequest("hotel"); // 2.准备DSL // 2.1.query request.source().query(QueryBuilders.matchQuery("all", "如家")); // 2.2.高亮 request.source().highlighter(new HighlightBuilder().field("name").requireFieldMatch(false)); // 3.发送请求 SearchResponse response = client.search(request, RequestOptions.DEFAULT); // 4.解析响应 handleResponse(response); } 3.6.2.高亮结果解析 高亮的结果与查询的文档结果默认是分离的，并不在一起。 因此解析高亮的代码需要额外处理： 代码解读： 第一步：从结果中获取source。hit.getSourceAsString()，这部分是非高亮结果，json字符串。还需要反序列为HotelDoc对象 第二步：获取高亮结果。hit.getHighlightFields()，返回值是一个Map，key是高亮字段名称，值是HighlightField对象，代表高亮值 第三步：从map中根据高亮字段名称，获取高亮字段值对象HighlightField 第四步：从HighlightField中获取Fragments，并且转为字符串。这部分就是真正的高亮字符串了 第五步：用高亮的结果替换HotelDoc中的非高亮结果 完整代码如下： private void handleResponse(SearchResponse response) { // 4.解析响应 SearchHits searchHits = response.getHits(); // 4.1.获取总条数 long total = searchHits.getTotalHits().value; System.out.println("共搜索到" + total + "条数据"); // 4.2.文档数组 SearchHit[] hits = searchHits.getHits(); // 4.3.遍历 for (SearchHit hit : hits) { // 获取文档source String json = hit.getSourceAsString(); // 反序列化 HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class); // 获取高亮结果 Map<String, HighlightField> highlightFields = hit.getHighlightFields(); if (!CollectionUtils.isEmpty(highlightFields)) { // 根据字段名获取高亮结果 HighlightField highlightField = highlightFields.get("name"); if (highlightField != null) { // 获取高亮值 String name = highlightField.getFragments()[0].string(); // 覆盖非高亮结果 hotelDoc.setName(name); } } System.out.println("hotelDoc = " + hotelDoc); } }
分布式搜索引擎 -- ElasticSearch基础 学习目标 1.初识elasticsearch 1.1.了解ES 1.1.1.elasticsearch的作用 elasticsearch是一款非常强大的开源搜索引擎，具备非常多强大功能，可以帮助我们从海量数据中快速找到需要的内容 例如： 在GitHub搜索代码 1.1.2.ELK技术栈 elasticsearch结合kibana、Logstash、Beats，也就是elastic stack（ELK）。被广泛应用在日志数据分析、实时监控等领域： 而elasticsearch是elastic stack的核心，负责存储、搜索、分析数据。 1.1.3.为什么不是其他搜索技术？ 目前比较知名的搜索引擎技术排名： 虽然在早期，Apache Solr是最主要的搜索引擎技术，但随着发展elasticsearch已经渐渐超越了Solr，独占鳌头： 1.1.5.总结 什么是elasticsearch？ 一个开源的分布式搜索引擎，可以用来实现搜索、日志统计、分析、系统监控等功能 什么是elastic stack（ELK）？ 是以elasticsearch为核心的技术栈，包括beats、Logstash、kibana、elasticsearch 什么是Lucene？ 是Apache的开源搜索引擎类库，提供了搜索引擎的核心API 1.2.倒排索引 倒排索引的概念是基于MySQL这样的正向索引而言的。 1.2.1.正向索引 那么什么是正向索引呢？例如给下表（tb_goods）中的id创建索引： 如果是根据id查询，那么直接走索引，查询速度非常快。 但如果是基于title做模糊查询，只能是逐行扫描数据，流程如下： 1）用户搜索数据，条件是title符合"%手机%" 2）逐行获取数据，比如id为1的数据 3）判断数据中的title是否符合用户搜索条件 4）如果符合则放入结果集，不符合则丢弃。回到步骤1 逐行扫描，也就是全表扫描，随着数据量增加，其查询效率也会越来越低。当数据量达到数百万时，就是一场灾难。 1.2.2.倒排索引 倒排索引中有两个非常重要的概念： 文档（Document）：用来搜索的数据，其中的每一条数据就是一个文档。例如一个网页、一个商品信息 词条（Term）：对文档数据或用户搜索数据，利用某种算法分词，得到的具备含义的词语就是词条。例如：我是中国人，就可以分为：我、是、中国人、中国、国人这样的几个词条 创建倒排索引是对正向索引的一种特殊处理，流程如下： 将每一个文档的数据利用算法分词，得到一个个词条 创建表，每行数据包括词条、词条所在文档id、位置等信息 因为词条唯一性，可以给词条创建索引，例如hash表结构索引 如图： 倒排索引的搜索流程如下（以搜索"华为手机"为例）： 1）用户输入条件"华为手机"进行搜索。 2）对用户输入内容分词，得到词条：华为、手机。 3）拿着词条在倒排索引中查找，可以得到包含词条的文档id：1、2、3。 4）拿着文档id到正向索引中查找具体文档。 如图： 虽然要先查询倒排索引，再查询倒排索引，但是无论是词条、还是文档id都建立了索引，查询速度非常快！无需全表扫描。 1.2.3.正向和倒排 那么为什么一个叫做正向索引，一个叫做倒排索引呢？ 正向索引是最传统的，根据id索引的方式。但根据词条查询时，必须先逐条获取每个文档，然后判断文档中是否包含所需要的词条，是根据文档找词条的过程。 而倒排索引则相反，是先找到用户要搜索的词条，根据词条得到保护词条的文档的id，然后根据id获取文档。是根据词条找文档的过程。 是不是恰好反过来了？ 那么两者方式的优缺点是什么呢？ 正向索引： 优点： 可以给多个字段创建索引 根据索引字段搜索、排序速度非常快 缺点： 根据非索引字段，或者索引字段中的部分词条查找时，只能全表扫描。 倒排索引： 优点： 根据词条搜索、模糊搜索时，速度非常快 缺点： 只能给词条创建索引，而不是字段 无法根据字段做排序 1.3.es的一些概念 elasticsearch中有很多独有的概念，与mysql中略有差别，但也有相似之处。 1.3.1.文档和字段 elasticsearch是面向文档（Document）存储的，可以是数据库中的一条商品数据，一个订单信息。文档数据会被序列化为json格式后存储在elasticsearch中： 而Json文档中往往包含很多的字段（Field），类似于数据库中的列。 1.3.2.索引和映射 索引（Index），就是相同类型的文档的集合。 例如： 所有用户文档，就可以组织在一起，称为用户的索引； 所有商品的文档，可以组织在一起，称为商品的索引； 所有订单的文档，可以组织在一起，称为订单的索引； 因此，我们可以把索引当做是数据库中的表。 数据库的表会有约束信息，用来定义表的结构、字段的名称、类型等信息。因此，索引库中就有映射（mapping），是索引中文档的字段约束信息，类似表的结构约束。 1.3.3.mysql与elasticsearch 我们统一的把mysql与elasticsearch的概念做一下对比： MySQL Elasticsearch 说明 Table Index 索引(index)，就是文档的集合，类似数据库的表(table) Row Document 文档（Document），就是一条条的数据，类似数据库中的行（Row），文档都是JSON格式 Column Field 字段（Field），就是JSON文档中的字段，类似数据库中的列（Column） Schema Mapping Mapping（映射）是索引中文档的约束，例如字段类型约束。类似数据库的表结构（Schema） SQL DSL DSL是elasticsearch提供的JSON风格的请求语句，用来操作elasticsearch，实现CRUD 是不是说，我们学习了elasticsearch就不再需要mysql了呢？ 并不是如此，两者各自有自己的擅长支出： Mysql：擅长事务类型操作，可以确保数据的安全和一致性 Elasticsearch：擅长海量数据的搜索、分析、计算 因此在企业中，往往是两者结合使用： 对安全性要求较高的写操作，使用mysql实现 对查询性能要求较高的搜索需求，使用elasticsearch实现 两者再基于某种方式，实现数据的同步，保证一致性 2.索引库操作 索引库就类似数据库表，mapping映射就类似表的结构。 我们要向es中存储数据，必须先创建“库”和“表”。 2.1.mapping映射属性 mapping是对索引库中文档的约束，常见的mapping属性包括： type：字段数据类型，常见的简单类型有： 字符串：text（可分词的文本）、keyword（精确值，例如：品牌、国家、ip地址） 数值：long、integer、short、byte、double、float、 布尔：boolean 日期：date 对象：object index：是否创建索引，默认为true analyzer：使用哪种分词器 properties：该字段的子字段 例如下面的json文档： { "age": 21, "weight": 52.1, "isMarried": false, "info": "你长得真可爱", "email": "wcy@wcy.cn", "score": [99.1, 99.5, 98.9], "name": { "firstName": "云", "lastName": "赵" } } 对应的每个字段映射（mapping）： age：类型为 integer；参与搜索，因此需要index为true；无需分词器 weight：类型为float；参与搜索，因此需要index为true；无需分词器 isMarried：类型为boolean；参与搜索，因此需要index为true；无需分词器 info：类型为字符串，需要分词，因此是text；参与搜索，因此需要index为true；分词器可以用ik_smart email：类型为字符串，但是不需要分词，因此是keyword；不参与搜索，因此需要index为false；无需分词器 score：虽然是数组，但是我们只看元素的类型，类型为float；参与搜索，因此需要index为true；无需分词器 name：类型为object，需要定义多个子属性 name.firstName；类型为字符串，但是不需要分词，因此是keyword；参与搜索，因此需要index为true；无需分词器 name.lastName；类型为字符串，但是不需要分词，因此是keyword；参与搜索，因此需要index为true；无需分词器 2.2.索引库的CRUD 这里我们统一使用Kibana编写DSL的方式来演示。 2.2.1.创建索引库和映射 基本语法： 请求方式：PUT 请求路径：/索引库名，可以自定义 请求参数：mapping映射 格式： PUT /索引库名称 { "mappings": { "properties": { "字段名":{ "type": "text", "analyzer": "ik_smart" }, "字段名2":{ "type": "keyword", "index": "false" }, "字段名3":{ "properties": { "子字段": { "type": "keyword" } } }, // ...略 } } } 示例： PUT /heima { "mappings": { "properties": { "info":{ "type": "text", "analyzer": "ik_smart" }, "email":{ "type": "keyword", "index": "falsae" }, "name":{ "properties": { "firstName": { "type": "keyword" } } }, // ... 略 } } } 2.2.2.查询索引库 基本语法： 请求方式：GET 请求路径：/索引库名 请求参数：无 格式： GET /索引库名 示例： 2.2.3.修改索引库 倒排索引结构虽然不复杂，但是一旦数据结构改变（比如改变了分词器），就需要重新创建倒排索引，这简直是灾难。因此索引库一旦创建，无法修改mapping。 虽然无法修改mapping中已有的字段，但是却允许添加新的字段到mapping中，因为不会对倒排索引产生影响。 语法说明： PUT /索引库名/_mapping { "properties": { "新字段名":{ "type": "integer" } } } 示例： 2.2.4.删除索引库 语法： 请求方式：DELETE 请求路径：/索引库名 请求参数：无 格式： DELETE /索引库名 在kibana中测试： 2.2.5.总结 索引库操作有哪些？ 创建索引库：PUT /索引库名 查询索引库：GET /索引库名 删除索引库：DELETE /索引库名 添加字段：PUT /索引库名/_mapping 3.文档操作 3.1.新增文档 语法： POST /索引库名/_doc/文档id { "字段1": "值1", "字段2": "值2", "字段3": { "子属性1": "值3", "子属性2": "值4" }, // ... } 示例： POST /heima/_doc/1 { "info": "黑马程序员Java讲师", "email": "zy@wcy.cn", "name": { "firstName": "云", "lastName": "赵" } } 响应： 3.2.查询文档 根据rest风格，新增是post，查询应该是get，不过查询一般都需要条件，这里我们把文档id带上。 语法： GET /{索引库名称}/_doc/{id} 通过kibana查看数据： GET /heima/_doc/1 查看结果： 3.3.删除文档 删除使用DELETE请求，同样，需要根据id进行删除： 语法： DELETE /{索引库名}/_doc/id值 示例： # 根据id删除数据 DELETE /heima/_doc/1 结果： 3.4.修改文档 修改有两种方式： 全量修改：直接覆盖原来的文档 增量修改：修改文档中的部分字段 3.4.1.全量修改 全量修改是覆盖原来的文档，其本质是： 根据指定的id删除文档 新增一个相同id的文档 注意：如果根据id删除时，id不存在，第二步的新增也会执行，也就从修改变成了新增操作了。 语法： PUT /{索引库名}/_doc/文档id { "字段1": "值1", "字段2": "值2", // ... 略 } 示例： PUT /heima/_doc/1 { "info": "你长得真可爱", "email": "wcy@wcy.cn", "name": { "firstName": "云", "lastName": "赵" } } 3.4.2.增量修改 增量修改是只修改指定id匹配的文档中的部分字段。 语法： POST /{索引库名}/_update/文档id { "doc": { "字段名": "新的值", } } 示例： POST /heima/_update/1 { "doc": { "email": "ZhaoYun@wcy.cn" } } 3.5.总结 文档操作有哪些？ 创建文档：POST /{索引库名}/_doc/文档id { json文档 } 查询文档：GET /{索引库名}/_doc/文档id 删除文档：DELETE /{索引库名}/_doc/文档id 修改文档： 全量修改：PUT /{索引库名}/_doc/文档id { json文档 } 增量修改：POST /{索引库名}/_update/文档id { "doc": {字段}} 4.RestAPI ES官方提供了各种不同语言的客户端，用来操作ES。这些客户端的本质就是组装DSL语句，通过http请求发送给ES。官方文档地址：https://www.elastic.co/guide/en/elasticsearch/client/index.html 其中的Java Rest Client又包括两种： Java Low Level Rest Client Java High Level Rest Client 我们学习的是Java HighLevel Rest Client客户端API 4.0.导入Demo工程 4.0.1.导入数据 数据结构如下： CREATE TABLE `tb_hotel` ( `id` bigint(20) NOT NULL COMMENT '酒店id', `name` varchar(255) NOT NULL COMMENT '酒店名称；例：7天酒店', `address` varchar(255) NOT NULL COMMENT '酒店地址；例：航头路', `price` int(10) NOT NULL COMMENT '酒店价格；例：329', `score` int(2) NOT NULL COMMENT '酒店评分；例：45，就是4.5分', `brand` varchar(32) NOT NULL COMMENT '酒店品牌；例：如家', `city` varchar(32) NOT NULL COMMENT '所在城市；例：上海', `star_name` varchar(16) DEFAULT NULL COMMENT '酒店星级，从低到高分别是：1星到5星，1钻到5钻', `business` varchar(255) DEFAULT NULL COMMENT '商圈；例：虹桥', `latitude` varchar(32) NOT NULL COMMENT '纬度；例：31.2497', `longitude` varchar(32) NOT NULL COMMENT '经度；例：120.3925', `pic` varchar(255) DEFAULT NULL COMMENT '酒店图片；例:/img/1.jpg', PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; 4.0.2.导入项目 tb_hotel.sql 4.0.3.mapping映射分析 创建索引库，最关键的是mapping映射，而mapping映射要考虑的信息包括： 字段名 字段数据类型 是否参与搜索 是否需要分词 如果分词，分词器是什么？ 其中： 字段名、字段数据类型，可以参考数据表结构的名称和类型 是否参与搜索要分析业务来判断，例如图片地址，就无需参与搜索 是否分词呢要看内容，内容如果是一个整体就无需分词，反之则要分词 分词器，我们可以统一使用ik_max_word 来看下酒店数据的索引库结构: PUT /hotel { "mappings": { "properties": { "id": { "type": "keyword" }, "name":{ "type": "text", "analyzer": "ik_max_word", "copy_to": "all" }, "address":{ "type": "keyword", "index": false }, "price":{ "type": "integer" }, "score":{ "type": "integer" }, "brand":{ "type": "keyword", "copy_to": "all" }, "city":{ "type": "keyword", "copy_to": "all" }, "starName":{ "type": "keyword" }, "business":{ "type": "keyword" }, "location":{ "type": "geo_point" }, "pic":{ "type": "keyword", "index": false }, "all":{ "type": "text", "analyzer": "ik_max_word" } } } } 几个特殊字段说明： location：地理坐标，里面包含精度、纬度 all：一个组合字段，其目的是将多字段的值 利用copy_to合并，提供给用户搜索 地理坐标说明： copy_to说明： 4.0.4.初始化RestClient 在elasticsearch提供的API中，与elasticsearch一切交互都封装在一个名为RestHighLevelClient的类中，必须先完成这个对象的初始化，建立与elasticsearch的连接。 分为三步： 1）引入es的RestHighLevelClient依赖： <dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>elasticsearch-rest-high-level-client</artifactId> </dependency> 2）因为SpringBoot默认的ES版本是7.6.2，所以我们需要覆盖默认的ES版本： <properties> <java.version>1.8</java.version> <elasticsearch.version>7.12.1</elasticsearch.version> </properties> 3）初始化RestHighLevelClient： 初始化的代码如下： RestHighLevelClient client = new RestHighLevelClient(RestClient.builder( HttpHost.create("http://ip:9200") )); 这里为了单元测试方便，我们创建一个测试类HotelIndexTest，然后将初始化的代码编写在@BeforeEach方法中： import org.apache.http.HttpHost; import org.elasticsearch.client.RestHighLevelClient; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import java.io.IOException; public class HotelIndexTest { private RestHighLevelClient client; @BeforeEach void setUp() { this.client = new RestHighLevelClient(RestClient.builder( HttpHost.create("http://ip:9200") )); } @AfterEach void tearDown() throws IOException { this.client.close(); } } 4.1.创建索引库 4.1.1.代码解读 创建索引库的API如下： 代码分为三步： 1）创建Request对象。因为是创建索引库的操作，因此Request是CreateIndexRequest。 2）添加请求参数，其实就是DSL的JSON参数部分。因为json字符串很长，这里是定义了静态字符串常量MAPPING_TEMPLATE，让代码看起来更加优雅。 3）发送请求，client.indices()方法的返回值是IndicesClient类型，封装了所有与索引库操作有关的方法。 4.1.2.完整示例 在hotel-demo的cn.wcy.hotel.constants包下，创建一个类，定义mapping映射的JSON字符串常量： public class HotelConstants { public static final String MAPPING_TEMPLATE = "{\n" + " \"mappings\": {\n" + " \"properties\": {\n" + " \"id\": {\n" + " \"type\": \"keyword\"\n" + " },\n" + " \"name\":{\n" + " \"type\": \"text\",\n" + " \"analyzer\": \"ik_max_word\",\n" + " \"copy_to\": \"all\"\n" + " },\n" + " \"address\":{\n" + " \"type\": \"keyword\",\n" + " \"index\": false\n" + " },\n" + " \"price\":{\n" + " \"type\": \"integer\"\n" + " },\n" + " \"score\":{\n" + " \"type\": \"integer\"\n" + " },\n" + " \"brand\":{\n" + " \"type\": \"keyword\",\n" + " \"copy_to\": \"all\"\n" + " },\n" + " \"city\":{\n" + " \"type\": \"keyword\",\n" + " \"copy_to\": \"all\"\n" + " },\n" + " \"starName\":{\n" + " \"type\": \"keyword\"\n" + " },\n" + " \"business\":{\n" + " \"type\": \"keyword\"\n" + " },\n" + " \"location\":{\n" + " \"type\": \"geo_point\"\n" + " },\n" + " \"pic\":{\n" + " \"type\": \"keyword\",\n" + " \"index\": false\n" + " },\n" + " \"all\":{\n" + " \"type\": \"text\",\n" + " \"analyzer\": \"ik_max_word\"\n" + " }\n" + " }\n" + " }\n" + "}"; } 在hotel-demo中的HotelIndexTest测试类中，编写单元测试，实现创建索引： @Test void createHotelIndex() throws IOException { // 1.创建Request对象 CreateIndexRequest request = new CreateIndexRequest("hotel"); // 2.准备请求的参数：DSL语句 request.source(MAPPING_TEMPLATE, XContentType.JSON); // 3.发送请求 client.indices().create(request, RequestOptions.DEFAULT); } 4.2.删除索引库 删除索引库的DSL语句非常简单： DELETE /hotel 与创建索引库相比： 请求方式从PUT变为DELTE 请求路径不变 无请求参数 所以代码的差异，注意体现在Request对象上。依然是三步走： 1）创建Request对象。这次是DeleteIndexRequest对象 2）准备参数。这里是无参 3）发送请求。改用delete方法 在hotel-demo中的HotelIndexTest测试类中，编写单元测试，实现删除索引： @Test void testDeleteHotelIndex() throws IOException { // 1.创建Request对象 DeleteIndexRequest request = new DeleteIndexRequest("hotel"); // 2.发送请求 client.indices().delete(request, RequestOptions.DEFAULT); } 4.3.判断索引库是否存在 判断索引库是否存在，本质就是查询，对应的DSL是： GET /hotel 因此与删除的Java代码流程是类似的。依然是三步走： 1）创建Request对象。这次是GetIndexRequest对象 2）准备参数。这里是无参 3）发送请求。改用exists方法 @Test void testExistsHotelIndex() throws IOException { // 1.创建Request对象 GetIndexRequest request = new GetIndexRequest("hotel"); // 2.发送请求 boolean exists = client.indices().exists(request, RequestOptions.DEFAULT); // 3.输出 System.err.println(exists ? "索引库已经存在！" : "索引库不存在！"); } 4.4.总结 JavaRestClient操作elasticsearch的流程基本类似。核心是client.indices()方法来获取索引库的操作对象。 索引库操作的基本步骤： 初始化RestHighLevelClient 创建XxxIndexRequest。XXX是Create、Get、Delete 准备DSL（ Create时需要，其它是无参） 发送请求。调用RestHighLevelClient#indices().xxx()方法，xxx是create、exists、delete 5.RestClient操作文档 为了与索引库操作分离，我们再次参加一个测试类，做两件事情： 初始化RestHighLevelClient 我们的酒店数据在数据库，需要利用IHotelService去查询，所以注入这个接口 import cn.wcy.hotel.pojo.Hotel; import cn.wcy.hotel.service.IHotelService; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import java.io.IOException; import java.util.List; @SpringBootTest public class HotelDocumentTest { @Autowired private IHotelService hotelService; private RestHighLevelClient client; @BeforeEach void setUp() { this.client = new RestHighLevelClient(RestClient.builder( HttpHost.create("http://ip:9200") )); } @AfterEach void tearDown() throws IOException { this.client.close(); } } 5.1.新增文档 我们要将数据库的酒店数据查询出来，写入elasticsearch中。 5.1.1.索引库实体类 数据库查询后的结果是一个Hotel类型的对象。结构如下： @Data @TableName("tb_hotel") public class Hotel { @TableId(type = IdType.INPUT) private Long id; private String name; private String address; private Integer price; private Integer score; private String brand; private String city; private String starName; private String business; private String longitude; private String latitude; private String pic; } 与我们的索引库结构存在差异： longitude和latitude需要合并为location 因此，我们需要定义一个新的类型，与索引库结构吻合： import lombok.Data; import lombok.NoArgsConstructor; @Data @NoArgsConstructor public class HotelDoc { private Long id; private String name; private String address; private Integer price; private Integer score; private String brand; private String city; private String starName; private String business; private String location; private String pic; public HotelDoc(Hotel hotel) { this.id = hotel.getId(); this.name = hotel.getName(); this.address = hotel.getAddress(); this.price = hotel.getPrice(); this.score = hotel.getScore(); this.brand = hotel.getBrand(); this.city = hotel.getCity(); this.starName = hotel.getStarName(); this.business = hotel.getBusiness(); this.location = hotel.getLatitude() + ", " + hotel.getLongitude(); this.pic = hotel.getPic(); } } 5.1.2.语法说明 新增文档的DSL语句如下： POST /{索引库名}/_doc/1 { "name": "Jack", "age": 21 } 对应的java代码如图： 可以看到与创建索引库类似，同样是三步走： 1）创建Request对象 2）准备请求参数，也就是DSL中的JSON文档 3）发送请求 变化的地方在于，这里直接使用client.xxx()的API，不再需要client.indices()了。 5.1.3.完整代码 我们导入酒店数据，基本流程一致，但是需要考虑几点变化： 酒店数据来自于数据库，我们需要先查询出来，得到hotel对象 hotel对象需要转为HotelDoc对象 HotelDoc需要序列化为json格式 因此，代码整体步骤如下： 1）根据id查询酒店数据Hotel 2）将Hotel封装为HotelDoc 3）将HotelDoc序列化为JSON 4）创建IndexRequest，指定索引库名和id 5）准备请求参数，也就是JSON文档 6）发送请求 在hotel-demo的HotelDocumentTest测试类中，编写单元测试： @Test void testAddDocument() throws IOException { // 1.根据id查询酒店数据 Hotel hotel = hotelService.getById(61083L); // 2.转换为文档类型 HotelDoc hotelDoc = new HotelDoc(hotel); // 3.将HotelDoc转json String json = JSON.toJSONString(hotelDoc); // 1.准备Request对象 IndexRequest request = new IndexRequest("hotel").id(hotelDoc.getId().toString()); // 2.准备Json文档 request.source(json, XContentType.JSON); // 3.发送请求 client.index(request, RequestOptions.DEFAULT); } 5.2.查询文档 5.2.1.语法说明 查询的DSL语句如下： GET /hotel/_doc/{id} 非常简单，因此代码大概分两步： 准备Request对象 发送请求 不过查询的目的是得到结果，解析为HotelDoc，因此难点是结果的解析。完整代码如下： 可以看到，结果是一个JSON，其中文档放在一个_source属性中，因此解析就是拿到_source，反序列化为Java对象即可。 与之前类似，也是三步走： 1）准备Request对象。这次是查询，所以是GetRequest 2）发送请求，得到结果。因为是查询，这里调用client.get()方法 3）解析结果，就是对JSON做反序列化 5.2.2.完整代码 在hotel-demo的HotelDocumentTest测试类中，编写单元测试： @Test void testGetDocumentById() throws IOException { // 1.准备Request GetRequest request = new GetRequest("hotel", "61082"); // 2.发送请求，得到响应 GetResponse response = client.get(request, RequestOptions.DEFAULT); // 3.解析响应结果 String json = response.getSourceAsString(); HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class); System.out.println(hotelDoc); } 5.3.删除文档 删除的DSL为是这样的： DELETE /hotel/_doc/{id} 与查询相比，仅仅是请求方式从DELETE变成GET，可以想象Java代码应该依然是三步走： 1）准备Request对象，因为是删除，这次是DeleteRequest对象。要指定索引库名和id 2）准备参数，无参 3）发送请求。因为是删除，所以是client.delete()方法 在hotel-demo的HotelDocumentTest测试类中，编写单元测试： @Test void testDeleteDocument() throws IOException { // 1.准备Request DeleteRequest request = new DeleteRequest("hotel", "61083"); // 2.发送请求 client.delete(request, RequestOptions.DEFAULT); } 5.4.修改文档 5.4.1.语法说明 修改我们讲过两种方式： 全量修改：本质是先根据id删除，再新增 增量修改：修改文档中的指定字段值 在RestClient的API中，全量修改与新增的API完全一致，判断依据是ID： 如果新增时，ID已经存在，则修改 如果新增时，ID不存在，则新增 这里不再赘述，我们主要关注增量修改。 代码示例如图： 与之前类似，也是三步走： 1）准备Request对象。这次是修改，所以是UpdateRequest 2）准备参数。也就是JSON文档，里面包含要修改的字段 3）更新文档。这里调用client.update()方法 5.4.2.完整代码 在hotel-demo的HotelDocumentTest测试类中，编写单元测试： @Test void testUpdateDocument() throws IOException { // 1.准备Request UpdateRequest request = new UpdateRequest("hotel", "61083"); // 2.准备请求参数 request.doc( "price", "952", "starName", "四钻" ); // 3.发送请求 client.update(request, RequestOptions.DEFAULT); } 5.5.批量导入文档 案例需求：利用BulkRequest批量将数据库数据导入到索引库中。 步骤如下： 利用mybatis-plus查询酒店数据 将查询到的酒店数据（Hotel）转换为文档类型数据（HotelDoc） 利用JavaRestClient中的BulkRequest批处理，实现批量新增文档 5.5.1.语法说明 批量处理BulkRequest，其本质就是将多个普通的CRUD请求组合在一起发送。 其中提供了一个add方法，用来添加其他请求： 可以看到，能添加的请求包括： IndexRequest，也就是新增 UpdateRequest，也就是修改 DeleteRequest，也就是删除 因此Bulk中添加了多个IndexRequest，就是批量新增功能了。示例： 其实还是三步走： 1）创建Request对象。这里是BulkRequest 2）准备参数。批处理的参数，就是其它Request对象，这里就是多个IndexRequest 3）发起请求。这里是批处理，调用的方法为client.bulk()方法 我们在导入酒店数据时，将上述代码改造成for循环处理即可。 5.5.2.完整代码 在hotel-demo的HotelDocumentTest测试类中，编写单元测试： @Test void testBulkRequest() throws IOException { // 批量查询酒店数据 List<Hotel> hotels = hotelService.list(); // 1.创建Request BulkRequest request = new BulkRequest(); // 2.准备参数，添加多个新增的Request for (Hotel hotel : hotels) { // 2.1.转换为文档类型HotelDoc HotelDoc hotelDoc = new HotelDoc(hotel); // 2.2.创建新增文档的Request对象 request.add(new IndexRequest("hotel") .id(hotelDoc.getId().toString()) .source(JSON.toJSONString(hotelDoc), XContentType.JSON)); } // 3.发送请求 client.bulk(request, RequestOptions.DEFAULT); } 5.6.小结 文档操作的基本步骤： 初始化RestHighLevelClient 创建XxxRequest。XXX是Index、Get、Update、Delete、Bulk 准备参数（Index、Update、Bulk时需要） 发送请求。调用RestHighLevelClient#.xxx()方法，xxx是index、get、update、delete、bulk 解析结果（Get时需要）
RabbitMQ 1.初识MQ 1.1.同步和异步通讯 微服务间通讯有同步和异步两种方式： 同步通讯：就像打电话，需要实时响应。 异步通讯：就像发邮件，不需要马上回复。 两种方式各有优劣，打电话可以立即得到响应，但是你却不能跟多个人同时通话。发送邮件可以同时与多个人收发邮件，但是往往响应会有延迟。 1.1.1.同步通讯 我们之前学习的Feign调用就属于同步方式，虽然调用可以实时得到结果，但存在下面的问题： 总结： 同步调用的优点： 时效性较强，可以立即得到结果 同步调用的问题： 耦合度高 性能和吞吐能力下降 有额外的资源消耗 有级联失败问题 1.1.2.异步通讯 异步调用则可以避免上述问题： 我们以购买商品为例，用户支付后需要调用订单服务完成订单状态修改，调用物流服务，从仓库分配响应的库存并准备发货。 在事件模式中，支付服务是事件发布者（publisher），在支付完成后只需要发布一个支付成功的事件（event），事件中带上订单id。 订单服务和物流服务是事件订阅者（Consumer），订阅支付成功的事件，监听到事件后完成自己业务即可。 为了解除事件发布者与订阅者之间的耦合，两者并不是直接通信，而是有一个中间人（Broker）。发布者发布事件到Broker，不关心谁来订阅事件。订阅者从Broker订阅事件，不关心谁发来的消息。 Broker 是一个像数据总线一样的东西，所有的服务要接收数据和发送数据都发到这个总线上，这个总线就像协议一样，让服务间的通讯变得标准和可控。 好处： 吞吐量提升：无需等待订阅者处理完成，响应更快速 故障隔离：服务没有直接调用，不存在级联失败问题 调用间没有阻塞，不会造成无效的资源占用 耦合度极低，每个服务都可以灵活插拔，可替换 流量削峰：不管发布事件的流量波动多大，都由Broker接收，订阅者可以按照自己的速度去处理事件 缺点： 架构复杂了，业务没有明显的流程线，不好管理 需要依赖于Broker的可靠、安全、性能 好在现在开源软件或云平台上 Broker 的软件是非常成熟的，比较常见的一种就是我们今天要学习的MQ技术。 1.2.技术对比： MQ，中文是消息队列（MessageQueue），字面来看就是存放消息的队列。也就是事件驱动架构中的Broker。 比较常见的MQ实现： ActiveMQ RabbitMQ RocketMQ Kafka 几种常见MQ的对比： RabbitMQ ActiveMQ RocketMQ Kafka 公司/社区 Rabbit Apache 阿里 Apache 开发语言 Erlang Java Java Scala&Java 协议支持 AMQP，XMPP，SMTP，STOMP OpenWire,STOMP，REST,XMPP,AMQP 自定义协议 自定义协议 可用性 高 一般 高 高 单机吞吐量 一般 差 高 非常高 消息延迟 微秒级 毫秒级 毫秒级 毫秒以内 消息可靠性 高 一般 高 一般 追求可用性：Kafka、 RocketMQ 、RabbitMQ 追求可靠性：RabbitMQ、RocketMQ 追求吞吐能力：RocketMQ、Kafka 追求消息低延迟：RabbitMQ、Kafka 1.3.安装RabbitMQ 单机部署 我们在Centos7虚拟机中使用Docker来安装。 下载镜像 方式一：在线拉取 docker pull rabbitmq:3-management 上传到虚拟机中后，使用命令加载镜像即可： docker load -i mq.tar 安装MQ 执行下面的命令来运行MQ容器： docker run \ -e RABBITMQ_DEFAULT_USER=wcy\ -e RABBITMQ_DEFAULT_PASS=123321 \ --name mq \ --hostname mq1 \ -p 15672:15672 \ -p 5672:5672 \ -d \ rabbitmq:3-management 集群部署 接下来，我们看看如何安装RabbitMQ的集群。 集群分类 在RabbitMQ的官方文档中，讲述了两种集群的配置方式： 普通模式：普通模式集群不进行数据同步，每个MQ都有自己的队列、数据信息（其它元数据信息如交换机等会同步）。例如我们有2个MQ：mq1，和mq2，如果你的消息在mq1，而你连接到了mq2，那么mq2会去mq1拉取消息，然后返回给你。如果mq1宕机，消息就会丢失。 镜像模式：与普通模式不同，队列会在各个mq的镜像节点之间同步，因此你连接到任何一个镜像节点，均可获取到消息。而且如果一个节点宕机，并不会导致数据丢失。不过，这种方式增加了数据同步的带宽消耗。 我们先来看普通模式集群。 设置网络 首先，我们需要让3台MQ互相知道对方的存在。 分别在3台机器中，设置 /etc/hosts文件，添加如下内容： xxx.xxx.xxx.101 mq1 xxx.xxx.xxx.102 mq2 xxx.xxx.xxx.103 mq3 并在每台机器上测试，是否可以ping通对方： MQ的基本结构： 2.快速入门 2.1 RabbitMQ中的角色： publisher：生产者 consumer：消费者 exchange个：交换机，负责消息路由 queue：队列，存储消息 virtualHost：虚拟主机，隔离不同租户的exchange、queue、消息的隔离 2.2.RabbitMQ消息模型 RabbitMQ官方提供了5个不同的Demo示例，对应了不同的消息模型： 2.4.入门案例 简单队列模式的模型图： 官方的HelloWorld是基于最基础的消息队列模型来实现的，只包括三个角色： publisher：消息发布者，将消息发送到队列queue queue：消息队列，负责接受并缓存消息 consumer：订阅队列，处理队列中的消息 2.4.1.publisher实现 思路： 建立连接 创建Channel 声明队列 发送消息 关闭连接和channel 代码实现： import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import org.junit.Test; import java.io.IOException; import java.util.concurrent.TimeoutException; public class PublisherTest { @Test public void testSendMessage() throws IOException, TimeoutException { // 1.建立连接 ConnectionFactory factory = new ConnectionFactory(); // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码 factory.setHost("yourIp"); factory.setPort(5672); factory.setVirtualHost("/"); factory.setUsername("wcy"); factory.setPassword("123321"); // 1.2.建立连接 Connection connection = factory.newConnection(); // 2.创建通道Channel Channel channel = connection.createChannel(); // 3.创建队列 String queueName = "simple.queue"; channel.queueDeclare(queueName, false, false, false, null); // 4.发送消息 String message = "hello, rabbitmq!"; channel.basicPublish("", queueName, null, message.getBytes()); System.out.println("发送消息成功：【" + message + "】"); // 5.关闭通道和连接 channel.close(); connection.close(); } } 2.4.2.consumer实现 代码思路： 建立连接 创建Channel 声明队列 订阅消息 代码实现： import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class ConsumerTest { public static void main(String[] args) throws IOException, TimeoutException { // 1.建立连接 ConnectionFactory factory = new ConnectionFactory(); // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码 factory.setHost("yourIp"); factory.setPort(5672); factory.setVirtualHost("/"); factory.setUsername("wcy"); factory.setPassword("123321"); // 1.2.建立连接 Connection connection = factory.newConnection(); // 2.创建通道Channel Channel channel = connection.createChannel(); // 3.创建队列 String queueName = "simple.queue"; channel.queueDeclare(queueName, false, false, false, null); // 4.订阅消息 channel.basicConsume(queueName, true, new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // 5.处理消息 String message = new String(body); System.out.println("接收到消息：【" + message + "】"); } }); System.out.println("等待接收消息。。。。"); } } 2.5.总结 基本消息队列的消息发送流程： 建立connection 创建channel 利用channel声明队列 利用channel向队列发送消息 基本消息队列的消息接收流程： 建立connection 创建channel 利用channel声明队列 定义consumer的消费行为handleDelivery() 利用channel将消费者与队列绑定 3.SpringAMQP SpringAMQP是基于RabbitMQ封装的一套模板，并且还利用SpringBoot对其实现了自动装配，使用起来非常方便。 SpringAmqp的官方地址：https://spring.io/projects/spring-amqp SpringAMQP提供了三个功能： 自动声明队列、交换机及其绑定关系 基于注解的监听器模式，异步接收消息 封装了RabbitTemplate工具，用于发送消息 3.1.Basic Queue 简单队列模型 在父工程mq-demo中引入依赖 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> 3.1.1.消息发送 首先配置MQ地址，在publisher服务的application.yml中添加配置： spring: rabbitmq: host: youIp # 主机名 port: 5672 # 端口 virtual-host: / # 虚拟主机 username: wcy# 用户名 password: 123321 # 密码 然后在publisher服务中编写测试类SpringAmqpTest，并利用RabbitTemplate实现消息发送： import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; @RunWith(SpringRunner.class) @SpringBootTest public class SpringAmqpTest { @Autowired private RabbitTemplate rabbitTemplate; @Test public void testSimpleQueue() { // 队列名称 String queueName = "simple.queue"; // 消息 String message = "hello, spring amqp!"; // 发送消息 rabbitTemplate.convertAndSend(queueName, message); } } 3.1.2.消息接收 首先配置MQ地址，在consumer服务的application.yml中添加配置： spring: rabbitmq: host: yourIp# 主机名 port: 5672 # 端口 virtual-host: / # 虚拟主机 username: wcy# 用户名 password: 123321 # 密码 然后在consumer服务的cn.wcy.mq.listener包中新建一个类SpringRabbitListener，代码如下： import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class SpringRabbitListener { @RabbitListener(queues = "simple.queue") public void listenSimpleQueueMessage(String msg) throws InterruptedException { System.out.println("spring 消费者接收到消息：【" + msg + "】"); } } 3.1.3.测试 启动consumer服务，然后在publisher服务中运行测试代码，发送MQ消息 3.2.WorkQueue Work queues，也被称为（Task queues），任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。 当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。 此时就可以使用work 模型，多个消费者共同处理消息处理，速度就能大大提高了。 3.2.1.消息发送 这次我们循环发送，模拟大量消息堆积现象。 在publisher服务中的SpringAmqpTest类中添加一个测试方法： /** * workQueue * 向队列中不停发送消息，模拟消息堆积。 */ @Test public void testWorkQueue() throws InterruptedException { // 队列名称 String queueName = "simple.queue"; // 消息 String message = "hello, message_"; for (int i = 0; i < 50; i++) { // 发送消息 rabbitTemplate.convertAndSend(queueName, message + i); Thread.sleep(20); } } 3.2.2.消息接收 要模拟多个消费者绑定同一个队列，我们在consumer服务的SpringRabbitListener中添加2个新的方法： @RabbitListener(queues = "simple.queue") public void listenWorkQueue1(String msg) throws InterruptedException { System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now()); Thread.sleep(20); } @RabbitListener(queues = "simple.queue") public void listenWorkQueue2(String msg) throws InterruptedException { System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now()); Thread.sleep(200); } 注意到这个消费者sleep了1000秒，模拟任务耗时。 3.2.3.测试 启动ConsumerApplication后，在执行publisher服务中刚刚编写的发送测试方法testWorkQueue。 可以看到消费者1很快完成了自己的25条消息。消费者2却在缓慢的处理自己的25条消息。 也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。这样显然是有问题的。 3.2.4.能者多劳 在spring中有一个简单的配置，可以解决这个问题。我们修改consumer服务的application.yml文件，添加配置： spring: rabbitmq: listener: simple: prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息 3.2.5.总结 Work模型的使用： 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理 通过设置prefetch来控制消费者预取的消息数量 3.3.发布/订阅 发布订阅的模型如图： 可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化： Publisher：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机） Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有以下3种类型： Fanout：广播，将消息交给所有绑定到交换机的队列 Direct：定向，把消息交给符合指定routing key 的队列 Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列 Consumer：消费者，与以前一样，订阅队列，没有变化 Queue：消息队列也与以前一样，接收消息、缓存消息。 Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！ 3.4.Fanout Fanout，英文翻译是扇出，我觉得在MQ中叫广播更合适。 在广播模式下，消息发送流程是这样的： 1） 可以有多个队列 2） 每个队列都要绑定到Exchange（交换机） 3） 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定 4） 交换机把消息发送给绑定过的所有队列 5） 订阅队列的消费者都能拿到消息 我们的计划是这样的： 创建一个交换机 wcy.fanout，类型是Fanout 创建两个队列fanout.queue1和fanout.queue2，绑定到交换机wcy.fanout 3.4.1.声明队列和交换机 Spring提供了一个接口Exchange，来表示所有不同类型的交换机： 在consumer中创建一个类，声明队列和交换机： import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class FanoutConfig { /** * 声明交换机 * @return Fanout类型交换机 */ @Bean public FanoutExchange fanoutExchange(){ return new FanoutExchange("wcy.fanout"); } /** * 第1个队列 */ @Bean public Queue fanoutQueue1(){ return new Queue("fanout.queue1"); } /** * 绑定队列和交换机 */ @Bean public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){ return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange); } /** * 第2个队列 */ @Bean public Queue fanoutQueue2(){ return new Queue("fanout.queue2"); } /** * 绑定队列和交换机 */ @Bean public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){ return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange); } } 3.4.2.消息发送 在publisher服务的SpringAmqpTest类中添加测试方法： @Test public void testFanoutExchange() { // 队列名称 String exchangeName = "wcy.fanout"; // 消息 String message = "hello, everyone!"; rabbitTemplate.convertAndSend(exchangeName, "", message); } 3.4.3.消息接收 在consumer服务的SpringRabbitListener中添加两个方法，作为消费者： @RabbitListener(queues = "fanout.queue1") public void listenFanoutQueue1(String msg) { System.out.println("消费者1接收到Fanout消息：【" + msg + "】"); } @RabbitListener(queues = "fanout.queue2") public void listenFanoutQueue2(String msg) { System.out.println("消费者2接收到Fanout消息：【" + msg + "】"); } 3.4.4.总结 交换机的作用是什么？ 接收publisher发送的消息 将消息按照规则路由到与之绑定的队列 不能缓存消息，路由失败，消息丢失 FanoutExchange的会将消息路由到每个绑定的队列 声明队列、交换机、绑定关系的Bean是什么？ Queue FanoutExchange Binding 3.5.Direct 在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。 在Direct模型下： 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key） 消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKey。 Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息 案例需求如下： 利用@RabbitListener声明Exchange、Queue、RoutingKey 在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2 在publisher中编写测试方法，向wcy. direct发送消息 3.5.1.基于注解声明队列和交换机 基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。 在consumer的SpringRabbitListener中添加两个消费者，同时基于注解来声明队列和交换机： @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "direct.queue1"), exchange = @Exchange(name = "wcy.direct", type = ExchangeTypes.DIRECT), key = {"red", "blue"} )) public void listenDirectQueue1(String msg){ System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】"); } @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "direct.queue2"), exchange = @Exchange(name = "wcy.direct", type = ExchangeTypes.DIRECT), key = {"red", "yellow"} )) public void listenDirectQueue2(String msg){ System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】"); } 3.5.2.消息发送 在publisher服务的SpringAmqpTest类中添加测试方法： @Test public void testSendDirectExchange() { // 交换机名称 String exchangeName = "wcy.direct"; // 消息 String message = "红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！"; // 发送消息 rabbitTemplate.convertAndSend(exchangeName, "red", message); } 3.5.3.总结 描述下Direct交换机与Fanout交换机的差异？ Fanout交换机将消息路由给每一个与之绑定的队列 Direct交换机根据RoutingKey判断路由给哪个队列 如果多个队列具有相同的RoutingKey，则与Fanout功能类似 基于@RabbitListener注解声明队列和交换机有哪些常见注解？ @Queue @Exchange 3.6.Topic 3.6.1.说明 Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！ Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert 通配符规则： #：匹配一个或多个词 *：匹配不多不少恰好1个词 举例： item.#：能够匹配item.spu.insert 或者 item.spu item.*：只能匹配item.spu ​ 图示： 解释： Queue1：绑定的是china.# ，因此凡是以 china.开头的routing key 都会被匹配到。包括china.news和china.weather Queue2：绑定的是#.news ，因此凡是以 .news结尾的 routing key 都会被匹配。包括china.news和japan.news 案例需求： 实现思路如下： 并利用@RabbitListener声明Exchange、Queue、RoutingKey 在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2 在publisher中编写测试方法，向wcy. topic发送消息 3.6.2.消息发送 在publisher服务的SpringAmqpTest类中添加测试方法： /** * topicExchange */ @Test public void testSendTopicExchange() { // 交换机名称 String exchangeName = "wcy.topic"; // 消息 String message = "喜报！孙悟空大战哥斯拉，胜!"; // 发送消息 rabbitTemplate.convertAndSend(exchangeName, "china.news", message); } 3.6.3.消息接收 在consumer服务的SpringRabbitListener中添加方法： @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "topic.queue1"), exchange = @Exchange(name = "wcy.topic", type = ExchangeTypes.TOPIC), key = "china.#" )) public void listenTopicQueue1(String msg){ System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】"); } @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "topic.queue2"), exchange = @Exchange(name = "wcy.topic", type = ExchangeTypes.TOPIC), key = "#.news" )) public void listenTopicQueue2(String msg){ System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】"); } 3.6.4.总结 描述下Direct交换机与Topic交换机的差异？ Topic交换机接收的消息RoutingKey必须是多个单词，以 **.** 分割 Topic交换机与队列绑定时的bindingKey可以指定通配符 #：代表0个或多个词 *：代表1个词 3.7.消息转换器 之前说过，Spring会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。 只不过，默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题： 数据体积过大 有安全漏洞 可读性差 我们来测试一下。 3.7.1.测试默认转换器 我们修改消息发送的代码，发送一个Map对象： @Test public void testSendMap() throws InterruptedException { // 准备消息 Map<String,Object> msg = new HashMap<>(); msg.put("name", "Jack"); msg.put("age", 21); // 发送消息 rabbitTemplate.convertAndSend("simple.queue","", msg); } 停止consumer服务 发送消息后查看控制台： 3.7.2.配置JSON转换器 显然，JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高，因此可以使用JSON方式来做序列化和反序列化。 在publisher和consumer两个服务中都引入依赖： <dependency> <groupId>com.fasterxml.jackson.dataformat</groupId> <artifactId>jackson-dataformat-xml</artifactId> <version>2.9.10</version> </dependency> 配置消息转换器。 在启动类中添加一个Bean即可： @Bean public MessageConverter jsonMessageConverter(){ return new Jackson2JsonMessageConverter(); }
1.Nacos配置管理 Nacos除了可以做注册中心，同样可以做配置管理来使用。 1.1.统一配置管理 当微服务部署的实例越来越多，达到数十、数百时，逐个修改微服务配置就会让人抓狂，而且很容易出错。我们需要一种统一配置管理方案，可以集中管理所有实例的配置。 Nacos一方面可以将配置集中管理，另一方可以在配置变更时，及时通知微服务，实现配置的热更新。 1.1.1.在nacos中添加配置文件 如何在nacos中管理配置呢？ 然后在弹出的表单中，填写配置信息： 注意：项目的核心配置，需要热更新的配置才有放到nacos管理的必要。基本不会变更的一些配置还是保存在微服务本地比较好。 1.1.2.从微服务拉取配置 微服务要拉取nacos中管理的配置，并且与本地的application.yml配置合并，才能完成项目启动。 但如果尚未读取application.yml，又如何得知nacos地址呢？ 因此spring引入了一种新的配置文件：bootstrap.yaml文件，会在application.yml之前被读取，流程如下： 1）引入nacos-config依赖 首先，在user-service服务中，引入nacos-config的客户端依赖： <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId> </dependency> 2）添加bootstrap.yaml 然后，在user-service中添加一个bootstrap.yaml文件，内容如下： spring: application: name: userservice # 服务名称 profiles: active: dev #开发环境，这里是dev cloud: nacos: server-addr: localhost:8848 # Nacos地址 config: file-extension: yaml # 文件后缀名 这里会根据spring.cloud.nacos.server-addr获取nacos地址，再根据 ${spring.application.name}-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}作为文件id，来读取配置。 本例中，就是去读取userservice-dev.yaml： 3）读取nacos配置 在user-service中的UserController中添加业务逻辑，读取pattern.dateformat配置： 完整代码： package cn.itcast.user.web; import cn.itcast.user.pojo.User; import cn.itcast.user.service.UserService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.web.bind.annotation.*; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; @Slf4j @RestController @RequestMapping("/user") public class UserController { @Autowired private UserService userService; @Value("${pattern.dateformat}") private String dateformat; @GetMapping("now") public String now(){ return LocalDateTime.now().format(DateTimeFormatter.ofPattern(dateformat)); } // ...略 } 在页面访问，可以看到效果： 1.2.配置热更新 我们最终的目的，是修改nacos中的配置后，微服务中无需重启即可让配置生效，也就是配置热更新。 要实现配置热更新，可以使用两种方式： 1.2.1.方式一 在@Value注入的变量所在类上添加注解@RefreshScope： 1.2.2.方式二 使用@ConfigurationProperties注解代替@Value注解。 在user-service服务中，添加一个类，读取patterrn.dateformat属性： package cn.itcast.user.config; import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; @Component @Data @ConfigurationProperties(prefix = "pattern") public class PatternProperties { private String dateformat; } 在UserController中使用这个类代替@Value： 完整代码： package cn.itcast.user.web; import cn.itcast.user.config.PatternProperties; import cn.itcast.user.pojo.User; import cn.itcast.user.service.UserService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; @Slf4j @RestController @RequestMapping("/user") public class UserController { @Autowired private UserService userService; @Autowired private PatternProperties patternProperties; @GetMapping("now") public String now(){ return LocalDateTime.now().format(DateTimeFormatter.ofPattern(patternProperties.getDateformat())); } // 略 } 1.3.配置共享 其实微服务启动时，会去nacos读取多个配置文件，例如： [spring.application.name]-[spring.profiles.active].yaml，例如：userservice-dev.yaml [spring.application.name].yaml，例如：userservice.yaml 而[spring.application.name].yaml不包含环境，因此可以被多个环境共享。 下面我们通过案例来测试配置共享 1）添加一个环境共享配置 我们在nacos中添加一个userservice.yaml文件： 2）在user-service中读取共享配置 在user-service服务中，修改PatternProperties类，读取新添加的属性： 在user-service服务中，修改UserController，添加一个方法： 3）运行两个UserApplication，使用不同的profile 修改UserApplication2这个启动项，改变其profile值： 这样，UserApplication(8081)使用的profile是dev，UserApplication2(8082)使用的profile是test。 启动UserApplication和UserApplication2 访问http://localhost:8081/user/prop，结果： 访问http://localhost:8082/user/prop，结果： 可以看出来，不管是dev，还是test环境，都读取到了envSharedValue这个属性的值。 4）配置共享的优先级 当nacos、服务本地同时出现相同属性时，优先级有高低之分： 1.4.搭建Nacos集群 Nacos生产环境下一定要部署为集群状态，部署方式参考课前资料中的文档： 2.Feign远程调用 先来看我们以前利用RestTemplate发起远程调用的代码： 存在下面的问题： •代码可读性差，编程体验不统一 •参数复杂URL难以维护 Feign是一个声明式的http客户端，官方地址：https://github.com/OpenFeign/feign 其作用就是帮助我们优雅的实现http请求的发送，解决上面提到的问题。 2.1.Feign替代RestTemplate Fegin的使用步骤如下： 1）引入依赖 我们在order-service服务的pom文件中引入feign的依赖： <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-openfeign</artifactId> </dependency> 2）添加注解 在order-service的启动类添加注解开启Feign的功能： 3）编写Feign的客户端 在order-service中新建一个接口，内容如下： package cn.itcast.order.client; import cn.itcast.order.pojo.User; import org.springframework.cloud.openfeign.FeignClient; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; @FeignClient("userservice") public interface UserClient { @GetMapping("/user/{id}") User findById(@PathVariable("id") Long id); } 这个客户端主要是基于SpringMVC的注解来声明远程调用的信息，比如： 服务名称：userservice 请求方式：GET 请求路径：/user/{id} 请求参数：Long id 返回值类型：User 这样，Feign就可以帮助我们发送http请求，无需自己使用RestTemplate来发送了。 4）测试 修改order-service中的OrderService类中的queryOrderById方法，使用Feign客户端代替RestTemplate： 是不是看起来优雅多了。 5）总结 使用Feign的步骤： ① 引入依赖 ② 添加@EnableFeignClients注解 ③ 编写FeignClient接口 ④ 使用FeignClient中定义的方法代替RestTemplate 2.2.自定义配置 Feign可以支持很多的自定义配置，如下表所示： 类型 作用 说明 feign.Logger.Level 修改日志级别 包含四种不同的级别：NONE、BASIC、HEADERS、FULL feign.codec.Decoder 响应结果的解析器 http远程调用的结果做解析，例如解析json字符串为java对象 feign.codec.Encoder 请求参数编码 将请求参数编码，便于通过http请求发送 feign. Contract 支持的注解格式 默认是SpringMVC的注解 feign. Retryer 失败重试机制 请求失败的重试机制，默认是没有，不过会使用Ribbon的重试 一般情况下，默认值就能满足我们使用，如果要自定义时，只需要创建自定义的@Bean覆盖默认Bean即可。 下面以日志为例来演示如何自定义配置。 2.2.1.配置文件方式 基于配置文件修改feign的日志级别可以针对单个服务： feign: client: config: userservice: # 针对某个微服务的配置 loggerLevel: FULL # 日志级别 也可以针对所有服务： feign: client: config: default: # 这里用default就是全局配置，如果是写服务名称，则是针对某个微服务的配置 loggerLevel: FULL # 日志级别 而日志的级别分为四种： NONE：不记录任何日志信息，这是默认值。 BASIC：仅记录请求的方法，URL以及响应状态码和执行时间 HEADERS：在BASIC的基础上，额外记录了请求和响应的头信息 FULL：记录所有请求和响应的明细，包括头信息、请求体、元数据。 2.2.2.Java代码方式 也可以基于Java代码来修改日志级别，先声明一个类，然后声明一个Logger.Level的对象： public class DefaultFeignConfiguration { @Bean public Logger.Level feignLogLevel(){ return Logger.Level.BASIC; // 日志级别为BASIC } } 如果要全局生效，将其放到启动类的@EnableFeignClients这个注解中： @EnableFeignClients(defaultConfiguration = DefaultFeignConfiguration .class) 如果是局部生效，则把它放到对应的@FeignClient这个注解中： @FeignClient(value = "userservice", configuration = DefaultFeignConfiguration .class) 2.3.Feign使用优化 Feign底层发起http请求，依赖于其它的框架。其底层客户端实现包括： •URLConnection：默认实现，不支持连接池 •Apache HttpClient ：支持连接池 •OKHttp：支持连接池 因此提高Feign的性能主要手段就是使用连接池代替默认的URLConnection。 这里我们用Apache的HttpClient来演示。 1）引入依赖 在order-service的pom文件中引入Apache的HttpClient依赖： <dependency> <groupId>io.github.openfeign</groupId> <artifactId>feign-httpclient</artifactId> </dependency> 2）配置连接池 在order-service的application.yml中添加配置： feign: client: config: default: # default全局的配置 loggerLevel: BASIC # 日志级别，BASIC就是基本的请求和响应信息 httpclient: enabled: true # 开启feign对HttpClient的支持 max-connections: 200 # 最大的连接数 max-connections-per-route: 50 # 每个路径的最大连接数 接下来，在FeignClientFactoryBean中的loadBalance方法中打断点： Debug方式启动order-service服务，可以看到这里的client，底层就是Apache HttpClient： 总结，Feign的优化： 1.日志级别尽量用basic 2.使用HttpClient或OKHttp代替URLConnection ① 引入feign-httpClient依赖 ② 配置文件开启httpClient功能，设置连接池参数 2.4.最佳实践 所谓最近实践，就是使用过程中总结的经验，最好的一种使用方式。 自习观察可以发现，Feign的客户端与服务提供者的controller代码非常相似： feign客户端： UserController： 有没有一种办法简化这种重复的代码编写呢？ 2.4.1.继承方式 一样的代码可以通过继承来共享： 1）定义一个API接口，利用定义方法，并基于SpringMVC注解做声明。 2）Feign客户端和Controller都集成改接口 优点： 简单 实现了代码共享 缺点： 服务提供方、服务消费方紧耦合 参数列表中的注解映射并不会继承，因此Controller中必须再次声明方法、参数列表、注解 2.4.2.抽取方式 将Feign的Client抽取为独立模块，并且把接口有关的POJO、默认的Feign配置都放到这个模块中，提供给所有消费者使用。 例如，将UserClient、User、Feign的默认配置都抽取到一个feign-api包中，所有微服务引用该依赖包，即可直接使用。 2.4.3.实现基于抽取的最佳实践 1）抽取 首先创建一个module，命名为feign-api： 项目结构： 在feign-api中然后引入feign的starter依赖 <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-openfeign</artifactId> </dependency> 然后，order-service中编写的UserClient、User、DefaultFeignConfiguration都复制到feign-api项目中 2）在order-service中使用feign-api 首先，删除order-service中的UserClient、User、DefaultFeignConfiguration等类或接口。 在order-service的pom文件中中引入feign-api的依赖： <dependency> <groupId>cn.itcast.demo</groupId> <artifactId>feign-api</artifactId> <version>1.0</version> </dependency> 修改order-service中的所有与上述三个组件有关的导包部分，改成导入feign-api中的包 3）重启测试 重启后，发现服务报错了： 这是因为UserClient现在在cn.itcast.feign.clients包下， 而order-service的@EnableFeignClients注解是在cn.itcast.order包下，不在同一个包，无法扫描到UserClient。 4）解决扫描包问题 方式一： 指定Feign应该扫描的包： @EnableFeignClients(basePackages = "cn.itcast.feign.clients") 方式二： 指定需要加载的Client接口： @EnableFeignClients(clients = {UserClient.class}) 3.Gateway服务网关 Spring Cloud Gateway 是 Spring Cloud 的一个全新项目，该项目是基于 Spring 5.0，Spring Boot 2.0 和 Project Reactor 等响应式编程和事件流技术开发的网关，它旨在为微服务架构提供一种简单有效的统一的 API 路由管理方式。 3.1.为什么需要网关 Gateway网关是我们服务的守门神，所有微服务的统一入口。 网关的核心功能特性： 请求路由 权限控制 限流 架构图： 权限控制：网关作为微服务入口，需要校验用户是是否有请求资格，如果没有则进行拦截。 路由和负载均衡：一切请求都必须先经过gateway，但网关不处理业务，而是根据某种规则，把请求转发到某个微服务，这个过程叫做路由。当然路由的目标服务有多个时，还需要做负载均衡。 限流：当请求流量过高时，在网关中按照下流的微服务能够接受的速度来放行请求，避免服务压力过大。 在SpringCloud中网关的实现包括两种： gateway zuul Zuul是基于Servlet的实现，属于阻塞式编程。而SpringCloudGateway则是基于Spring5中提供的WebFlux，属于响应式编程的实现，具备更好的性能。 3.2.gateway快速入门 下面，我们就演示下网关的基本路由功能。基本步骤如下： 创建SpringBoot工程gateway，引入网关依赖 编写启动类 编写基础配置和路由规则 启动网关服务进行测试 1）创建gateway服务，引入依赖 创建服务： 引入依赖： <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-gateway</artifactId> </dependency> <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId> </dependency> 2）编写启动类 package cn.itcast.gateway; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class GatewayApplication { public static void main(String[] args) { SpringApplication.run(GatewayApplication.class, args); } } 3）编写基础配置和路由规则 创建application.yml文件，内容如下： server: port: 10010 # 网关端口 spring: application: name: gateway # 服务名称 cloud: nacos: server-addr: localhost:8848 # nacos地址 gateway: routes: # 网关路由配置 - id: user-service # 路由id，自定义，只要唯一即可 # uri: http://127.0.0.1:8081 # 路由的目标地址 http就是固定地址 uri: lb://userservice # 路由的目标地址 lb就是负载均衡，后面跟服务名称 predicates: # 路由断言，也就是判断请求是否符合路由规则的条件 - Path=/user/** # 这个是按照路径匹配，只要以/user/开头就符合要求 我们将符合Path 规则的一切请求，都代理到 uri参数指定的地址。 本例中，我们将 /user/**开头的请求，代理到lb://userservice，lb是负载均衡，根据服务名拉取服务列表，实现负载均衡。 4）重启测试 重启网关，访问http://localhost:10010/user/1时，符合/user/**规则，请求转发到uri：http://userservice/user/1，得到了结果： 5）网关路由的流程图 整个访问的流程如下： 总结： 网关搭建步骤： 创建项目，引入nacos服务发现和gateway依赖 配置application.yml，包括服务基本信息、nacos地址、路由 路由配置包括： 路由id：路由的唯一标示 路由目标（uri）：路由的目标地址，http代表固定地址，lb代表根据服务名负载均衡 路由断言（predicates）：判断路由的规则， 路由过滤器（filters）：对请求或响应做处理 接下来，就重点来学习路由断言和路由过滤器的详细知识 3.3.断言工厂 我们在配置文件中写的断言规则只是字符串，这些字符串会被Predicate Factory读取并处理，转变为路由判断的条件 例如Path=/user/**是按照路径匹配，这个规则是由 org.springframework.cloud.gateway.handler.predicate.PathRoutePredicateFactory类来 处理的，像这样的断言工厂在SpringCloudGateway还有十几个: 名称 说明 示例 After 是某个时间点后的请求 - After=2037-01-20T17:42:47.789-07:00[America/Denver] Before 是某个时间点之前的请求 - Before=2031-04-13T15:14:47.433+08:00[Asia/Shanghai] Between 是某两个时间点之前的请求 - Between=2037-01-20T17:42:47.789-07:00[America/Denver], 2037-01-21T17:42:47.789-07:00[America/Denver] Cookie 请求必须包含某些cookie - Cookie=chocolate, ch.p Header 请求必须包含某些header - Header=X-Request-Id, \d+ Host 请求必须是访问某个host（域名） - Host=.somehost.org,.anotherhost.org Method 请求方式必须是指定方式 - Method=GET,POST Path 请求路径必须符合指定规则 - Path=/red/{segment},/blue/** Query 请求参数必须包含指定参数 - Query=name, Jack或者- Query=name RemoteAddr 请求者的ip必须是指定范围 - RemoteAddr=192.168.1.1/24 Weight 权重处理 我们只需要掌握Path这种路由工程就可以了。 3.4.过滤器工厂 GatewayFilter是网关中提供的一种过滤器，可以对进入网关的请求和微服务返回的响应做处理： 3.4.1.路由过滤器的种类 Spring提供了31种不同的路由过滤器工厂。例如： 名称 说明 AddRequestHeader 给当前请求添加一个请求头 RemoveRequestHeader 移除请求中的一个请求头 AddResponseHeader 给响应结果中添加一个响应头 RemoveResponseHeader 从响应结果中移除有一个响应头 RequestRateLimiter 限制请求的流量 3.4.2.请求头过滤器 下面我们以AddRequestHeader 为例来讲解。 需求：给所有进入userservice的请求添加一个请求头：Truth=itcast is freaking awesome! 只需要修改gateway服务的application.yml文件，添加路由过滤即可： spring: cloud: gateway: routes: - id: user-service uri: lb://userservice predicates: - Path=/user/** filters: # 过滤器 - AddRequestHeader=Truth, Itcast is freaking awesome! # 添加请求头 当前过滤器写在userservice路由下，因此仅仅对访问userservice的请求有效。 3.4.3.默认过滤器 如果要对所有的路由都生效，则可以将过滤器工厂写到default下。格式如下： spring: cloud: gateway: routes: - id: user-service uri: lb://userservice predicates: - Path=/user/** default-filters: # 默认过滤项 - AddRequestHeader=Truth, Itcast is freaking awesome! 3.4.4.总结 过滤器的作用是什么？ ① 对路由的请求或响应做加工处理，比如添加请求头 ② 配置在路由下的过滤器只对当前路由的请求生效 defaultFilters的作用是什么？ ① 对所有路由都生效的过滤器 3.5.全局过滤器 上一节学习的过滤器，网关提供了31种，但每一种过滤器的作用都是固定的。如果我们希望拦截请求，做自己的业务逻辑则没办法实现。 3.5.1.全局过滤器作用 全局过滤器的作用也是处理一切进入网关的请求和微服务响应，与GatewayFilter的作用一样。区别在于GatewayFilter通过配置定义，处理逻辑是固定的；而GlobalFilter的逻辑需要自己写代码实现。 定义方式是实现GlobalFilter接口。 public interface GlobalFilter { /** * 处理当前请求，有必要的话通过{@link GatewayFilterChain}将请求交给下一个过滤器处理 * * @param exchange 请求上下文，里面可以获取Request、Response等信息 * @param chain 用来把请求委托给下一个过滤器 * @return {@code Mono<Void>} 返回标示当前过滤器业务结束 */ Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain); } 在filter中编写自定义逻辑，可以实现下列功能： 登录状态判断 权限校验 请求限流等 3.5.2.自定义全局过滤器 需求：定义全局过滤器，拦截请求，判断请求的参数是否满足下面条件： 参数中是否有authorization， authorization参数值是否为admin 如果同时满足则放行，否则拦截 实现： 在gateway中定义一个过滤器： package cn.itcast.gateway.filters; import org.springframework.cloud.gateway.filter.GatewayFilterChain; import org.springframework.cloud.gateway.filter.GlobalFilter; import org.springframework.core.annotation.Order; import org.springframework.http.HttpStatus; import org.springframework.stereotype.Component; import org.springframework.web.server.ServerWebExchange; import reactor.core.publisher.Mono; @Order(-1) @Component public class AuthorizeFilter implements GlobalFilter { @Override public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) { // 1.获取请求参数 MultiValueMap<String, String> params = exchange.getRequest().getQueryParams(); // 2.获取authorization参数 String auth = params.getFirst("authorization"); // 3.校验 if ("admin".equals(auth)) { // 放行 return chain.filter(exchange); } // 4.拦截 // 4.1.禁止访问，设置状态码 exchange.getResponse().setStatusCode(HttpStatus.FORBIDDEN); // 4.2.结束处理 return exchange.getResponse().setComplete(); } } 3.5.3.过滤器执行顺序 请求进入网关会碰到三类过滤器：当前路由的过滤器、DefaultFilter、GlobalFilter 请求路由后，会将当前路由过滤器和DefaultFilter、GlobalFilter，合并到一个过滤器链（集合）中，排序后依次执行每个过滤器： 排序的规则是什么呢？ 每一个过滤器都必须指定一个int类型的order值，order值越小，优先级越高，执行顺序越靠前。 GlobalFilter通过实现Ordered接口，或者添加@Order注解来指定order值，由我们自己指定 路由过滤器和defaultFilter的order由Spring指定，默认是按照声明顺序从1递增。 当过滤器的order值一样时，会按照 defaultFilter > 路由过滤器 > GlobalFilter的顺序执行。 详细内容，可以查看源码： org.springframework.cloud.gateway.route.RouteDefinitionRouteLocator#getFilters()方法是先加载defaultFilters，然后再加载某个route的filters，然后合并。 org.springframework.cloud.gateway.handler.FilteringWebHandler#handle()方法会加载全局过滤器，与前面的过滤器合并后根据order排序，组织过滤器链 3.6.跨域问题 3.6.1.什么是跨域问题 跨域：域名不一致就是跨域，主要包括： 域名不同： www.taobao.com 和 www.taobao.org 和 www.jd.com 和 miaosha.jd.com 域名相同，端口不同：localhost:8080和localhost8081 跨域问题：浏览器禁止请求的发起者与服务端发生跨域ajax请求，请求被浏览器拦截的问题 解决方案：CORS，这个以前应该学习过，这里不再赘述了。不知道的小伙伴可以查看https://www.ruanyifeng.com/blog/2016/04/cors.html 3.6.2.模拟跨域问题 找到课前资料的页面文件： 放入tomcat或者nginx这样的web服务器中，启动并访问。 可以在浏览器控制台看到下面的错误： 从localhost:8090访问localhost:10010，端口不同，显然是跨域的请求。 3.6.3.解决跨域问题 在gateway服务的application.yml文件中，添加下面的配置： spring: cloud: gateway: # 。。。 globalcors: # 全局的跨域处理 add-to-simple-url-handler-mapping: true # 解决options请求被拦截问题 corsConfigurations: '[/**]': allowedOrigins: # 允许哪些网站的跨域请求 - "http://localhost:8090" allowedMethods: # 允许的跨域ajax的请求方式 - "GET" - "POST" - "DELETE" - "PUT" - "OPTIONS" allowedHeaders: "*" # 允许在请求中携带的头信息 allowCredentials: true # 是否允许携带cookie maxAge: 360000 # 这次跨域检测的有效期 相关知识ppt:SpringCloud.pptx
Nacos集群搭建 1.集群结构图 官方给出的Nacos集群图： 其中包含3个nacos节点，然后一个负载均衡器代理3个Nacos。这里负载均衡器可以使用nginx。 我们计划的集群结构： 三个nacos节点的地址： 节点 ip port nacos1 192.168.150.1 8845 nacos2 192.168.150.1 8846 nacos3 192.168.150.1 8847 2.搭建集群 搭建集群的基本步骤： 搭建数据库，初始化数据库表结构 下载nacos安装包 配置nacos 启动nacos集群 nginx反向代理 2.1.初始化数据库 Nacos默认数据存储在内嵌数据库Derby中，不属于生产可用的数据库。 官方推荐的最佳实践是使用带有主从的高可用数据库集群，主从模式的高可用数据库可以参考传智教育的后续高手课程。 这里我们以单点的数据库为例来讲解。 首先新建一个数据库，命名为nacos，而后导入下面的SQL： CREATE TABLE `config_info` ( `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id', `data_id` varchar(255) NOT NULL COMMENT 'data_id', `group_id` varchar(255) DEFAULT NULL, `content` longtext NOT NULL COMMENT 'content', `md5` varchar(32) DEFAULT NULL COMMENT 'md5', `gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', `gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '修改时间', `src_user` text COMMENT 'source user', `src_ip` varchar(50) DEFAULT NULL COMMENT 'source ip', `app_name` varchar(128) DEFAULT NULL, `tenant_id` varchar(128) DEFAULT '' COMMENT '租户字段', `c_desc` varchar(256) DEFAULT NULL, `c_use` varchar(64) DEFAULT NULL, `effect` varchar(64) DEFAULT NULL, `type` varchar(64) DEFAULT NULL, `c_schema` text, PRIMARY KEY (`id`), UNIQUE KEY `uk_configinfo_datagrouptenant` (`data_id`,`group_id`,`tenant_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='config_info'; /******************************************/ /* 数据库全名 = nacos_config */ /* 表名称 = config_info_aggr */ /******************************************/ CREATE TABLE `config_info_aggr` ( `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id', `data_id` varchar(255) NOT NULL COMMENT 'data_id', `group_id` varchar(255) NOT NULL COMMENT 'group_id', `datum_id` varchar(255) NOT NULL COMMENT 'datum_id', `content` longtext NOT NULL COMMENT '内容', `gmt_modified` datetime NOT NULL COMMENT '修改时间', `app_name` varchar(128) DEFAULT NULL, `tenant_id` varchar(128) DEFAULT '' COMMENT '租户字段', PRIMARY KEY (`id`), UNIQUE KEY `uk_configinfoaggr_datagrouptenantdatum` (`data_id`,`group_id`,`tenant_id`,`datum_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='增加租户字段'; /******************************************/ /* 数据库全名 = nacos_config */ /* 表名称 = config_info_beta */ /******************************************/ CREATE TABLE `config_info_beta` ( `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id', `data_id` varchar(255) NOT NULL COMMENT 'data_id', `group_id` varchar(128) NOT NULL COMMENT 'group_id', `app_name` varchar(128) DEFAULT NULL COMMENT 'app_name', `content` longtext NOT NULL COMMENT 'content', `beta_ips` varchar(1024) DEFAULT NULL COMMENT 'betaIps', `md5` varchar(32) DEFAULT NULL COMMENT 'md5', `gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', `gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '修改时间', `src_user` text COMMENT 'source user', `src_ip` varchar(50) DEFAULT NULL COMMENT 'source ip', `tenant_id` varchar(128) DEFAULT '' COMMENT '租户字段', PRIMARY KEY (`id`), UNIQUE KEY `uk_configinfobeta_datagrouptenant` (`data_id`,`group_id`,`tenant_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='config_info_beta'; /******************************************/ /* 数据库全名 = nacos_config */ /* 表名称 = config_info_tag */ /******************************************/ CREATE TABLE `config_info_tag` ( `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id', `data_id` varchar(255) NOT NULL COMMENT 'data_id', `group_id` varchar(128) NOT NULL COMMENT 'group_id', `tenant_id` varchar(128) DEFAULT '' COMMENT 'tenant_id', `tag_id` varchar(128) NOT NULL COMMENT 'tag_id', `app_name` varchar(128) DEFAULT NULL COMMENT 'app_name', `content` longtext NOT NULL COMMENT 'content', `md5` varchar(32) DEFAULT NULL COMMENT 'md5', `gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', `gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '修改时间', `src_user` text COMMENT 'source user', `src_ip` varchar(50) DEFAULT NULL COMMENT 'source ip', PRIMARY KEY (`id`), UNIQUE KEY `uk_configinfotag_datagrouptenanttag` (`data_id`,`group_id`,`tenant_id`,`tag_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='config_info_tag'; /******************************************/ /* 数据库全名 = nacos_config */ /* 表名称 = config_tags_relation */ /******************************************/ CREATE TABLE `config_tags_relation` ( `id` bigint(20) NOT NULL COMMENT 'id', `tag_name` varchar(128) NOT NULL COMMENT 'tag_name', `tag_type` varchar(64) DEFAULT NULL COMMENT 'tag_type', `data_id` varchar(255) NOT NULL COMMENT 'data_id', `group_id` varchar(128) NOT NULL COMMENT 'group_id', `tenant_id` varchar(128) DEFAULT '' COMMENT 'tenant_id', `nid` bigint(20) NOT NULL AUTO_INCREMENT, PRIMARY KEY (`nid`), UNIQUE KEY `uk_configtagrelation_configidtag` (`id`,`tag_name`,`tag_type`), KEY `idx_tenant_id` (`tenant_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='config_tag_relation'; /******************************************/ /* 数据库全名 = nacos_config */ /* 表名称 = group_capacity */ /******************************************/ CREATE TABLE `group_capacity` ( `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键ID', `group_id` varchar(128) NOT NULL DEFAULT '' COMMENT 'Group ID，空字符表示整个集群', `quota` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '配额，0表示使用默认值', `usage` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '使用量', `max_size` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '单个配置大小上限，单位为字节，0表示使用默认值', `max_aggr_count` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '聚合子配置最大个数，，0表示使用默认值', `max_aggr_size` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '单个聚合数据的子配置大小上限，单位为字节，0表示使用默认值', `max_history_count` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '最大变更历史数量', `gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', `gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '修改时间', PRIMARY KEY (`id`), UNIQUE KEY `uk_group_id` (`group_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='集群、各Group容量信息表'; /******************************************/ /* 数据库全名 = nacos_config */ /* 表名称 = his_config_info */ /******************************************/ CREATE TABLE `his_config_info` ( `id` bigint(64) unsigned NOT NULL, `nid` bigint(20) unsigned NOT NULL AUTO_INCREMENT, `data_id` varchar(255) NOT NULL, `group_id` varchar(128) NOT NULL, `app_name` varchar(128) DEFAULT NULL COMMENT 'app_name', `content` longtext NOT NULL, `md5` varchar(32) DEFAULT NULL, `gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, `gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, `src_user` text, `src_ip` varchar(50) DEFAULT NULL, `op_type` char(10) DEFAULT NULL, `tenant_id` varchar(128) DEFAULT '' COMMENT '租户字段', PRIMARY KEY (`nid`), KEY `idx_gmt_create` (`gmt_create`), KEY `idx_gmt_modified` (`gmt_modified`), KEY `idx_did` (`data_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='多租户改造'; /******************************************/ /* 数据库全名 = nacos_config */ /* 表名称 = tenant_capacity */ /******************************************/ CREATE TABLE `tenant_capacity` ( `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键ID', `tenant_id` varchar(128) NOT NULL DEFAULT '' COMMENT 'Tenant ID', `quota` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '配额，0表示使用默认值', `usage` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '使用量', `max_size` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '单个配置大小上限，单位为字节，0表示使用默认值', `max_aggr_count` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '聚合子配置最大个数', `max_aggr_size` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '单个聚合数据的子配置大小上限，单位为字节，0表示使用默认值', `max_history_count` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '最大变更历史数量', `gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', `gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '修改时间', PRIMARY KEY (`id`), UNIQUE KEY `uk_tenant_id` (`tenant_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='租户容量信息表'; CREATE TABLE `tenant_info` ( `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id', `kp` varchar(128) NOT NULL COMMENT 'kp', `tenant_id` varchar(128) default '' COMMENT 'tenant_id', `tenant_name` varchar(128) default '' COMMENT 'tenant_name', `tenant_desc` varchar(256) DEFAULT NULL COMMENT 'tenant_desc', `create_source` varchar(32) DEFAULT NULL COMMENT 'create_source', `gmt_create` bigint(20) NOT NULL COMMENT '创建时间', `gmt_modified` bigint(20) NOT NULL COMMENT '修改时间', PRIMARY KEY (`id`), UNIQUE KEY `uk_tenant_info_kptenantid` (`kp`,`tenant_id`), KEY `idx_tenant_id` (`tenant_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='tenant_info'; CREATE TABLE `users` ( `username` varchar(50) NOT NULL PRIMARY KEY, `password` varchar(500) NOT NULL, `enabled` boolean NOT NULL ); CREATE TABLE `roles` ( `username` varchar(50) NOT NULL, `role` varchar(50) NOT NULL, UNIQUE INDEX `idx_user_role` (`username` ASC, `role` ASC) USING BTREE ); CREATE TABLE `permissions` ( `role` varchar(50) NOT NULL, `resource` varchar(255) NOT NULL, `action` varchar(8) NOT NULL, UNIQUE INDEX `uk_role_permission` (`role`,`resource`,`action`) USING BTREE ); INSERT INTO users (username, password, enabled) VALUES ('nacos', '$2a$10$EuWPZHzz32dJN7jexM34MOeYirDdFAZm2kuWj7VEOJhhZkDrxfvUu', TRUE); INSERT INTO roles (username, role) VALUES ('nacos', 'ROLE_ADMIN'); 2.2.下载nacos nacos在GitHub上有下载地址：https://github.com/alibaba/nacos/tags，可以选择任意版本下载。 本例中才用1.4.1版本： 2.3.配置Nacos 将这个包解压到任意非中文目录下，如图： 目录说明： bin：启动脚本 conf：配置文件 进入nacos的conf目录，修改配置文件cluster.conf.example，重命名为cluster.conf： 然后添加内容： 127.0.0.1:8845 127.0.0.1.8846 127.0.0.1.8847 然后修改application.properties文件，添加数据库配置 spring.datasource.platform=mysql db.num=1 db.url.0=jdbc:mysql://127.0.0.1:3306/nacos?characterEncoding=utf8&connectTimeout=1000&socketTimeout=3000&autoReconnect=true&useUnicode=true&useSSL=false&serverTimezone=UTC db.user.0=root db.password.0=123 2.4.启动 将nacos文件夹复制三份，分别命名为：nacos1、nacos2、nacos3 然后分别修改三个文件夹中的application.properties， nacos1: server.port=8845 nacos2: server.port=8846 nacos3: server.port=8847 然后分别启动三个nacos节点： startup.cmd 2.5.nginx反向代理 找到课前资料提供的nginx安装包： 解压到任意非中文目录下： 修改conf/nginx.conf文件，配置如下： upstream nacos-cluster { server 127.0.0.1:8845; server 127.0.0.1:8846; server 127.0.0.1:8847; } server { listen 80; server_name localhost; location /nacos { proxy_pass http://nacos-cluster; } } 而后在浏览器访问：http://localhost/nacos即可。 代码中application.yml文件配置如下： spring: cloud: nacos: server-addr: localhost:80 # Nacos地址 2.6.优化 实际部署时，需要给做反向代理的nginx服务器设置一个域名，这样后续如果有服务器迁移nacos的客户端也无需更改配置. Nacos的各个节点应该部署到多个不同服务器，做好容灾和隔离
Nacos安装指南 1.Windows安装 开发阶段采用单机安装即可。 1.1.下载安装包 在Nacos的GitHub页面，提供有下载链接，可以下载编译好的Nacos服务端或者源代码： GitHub主页：https://github.com/alibaba/nacos GitHub的Release下载页：https://github.com/alibaba/nacos/releases 如图： 本课程采用1.4.1.版本的Nacos，课前资料已经准备了安装包： windows版本使用nacos-server-1.4.1.zip包即可。 1.2.解压 将这个包解压到任意非中文目录下，如图： 目录说明： bin：启动脚本 conf：配置文件 1.3.端口配置 Nacos的默认端口是8848，如果你电脑上的其它进程占用了8848端口，请先尝试关闭该进程。 如果无法关闭占用8848端口的进程，也可以进入nacos的conf目录，修改配置文件中的端口： 修改其中的内容： 1.4.启动 启动非常简单，进入bin目录，结构如下： 然后执行命令即可： windows命令： startup.cmd -m standalone 执行后的效果如图： 1.5.访问 在浏览器输入地址：http://127.0.0.1:8848/nacos即可： 默认的账号和密码都是nacos，进入后： 2.Linux安装 Linux或者Mac安装方式与Windows类似。 2.1.安装JDK Nacos依赖于JDK运行，索引Linux上也需要安装JDK才行。 上传jdk安装包： 上传到某个目录，例如：/usr/local/ 然后解压缩： tar -xvf jdk-8u144-linux-x64.tar.gz 然后重命名为java 配置环境变量： export JAVA_HOME=/usr/local/java export PATH=$PATH:$JAVA_HOME/bin 设置环境变量： source /etc/profile 2.2.上传安装包 如图： 也可以直接使用课前资料中的tar.gz： 上传到Linux服务器的某个目录，例如/usr/local/src目录下： 2.3.解压 命令解压缩安装包： tar -xvf nacos-server-1.4.1.tar.gz 然后删除安装包： rm -rf nacos-server-1.4.1.tar.gz 目录中最终样式： 目录内部： 2.4.端口配置 与windows中类似 2.5.启动 在nacos/bin目录中，输入命令启动Nacos： sh startup.sh -m standalone 3.Nacos的依赖 父工程： <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-alibaba-dependencies</artifactId> <version>2.2.5.RELEASE</version> <type>pom</type> <scope>import</scope> </dependency> 客户端： <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId> </dependency>