微服务通信方案

RPC

RPC实现微服务通信的核心思想

• 全局注册表：把RPC支持的所有方法都注册进去
• 通过将Java对象进行编码 + 方法名传递（TCP/IP 协议）到目标服务器实现微服务通信

RPC的优缺点

• 目前市面上最流行的RPC框架有：gRPC、Thrift、Dubbo
• 速度快，并发性能高
• 实现复杂（相对于Rest而言），需要做的工作与维护上更多（例如：Server 的地址一般存在Zookeeper上，就需要引入和维护ZK）

HTTP （Rest）

认识HTTP

• 标准化的HTTP协议（GET 、POST、PUT、DELETE等），目前主流的微服务通信框架实现都是HTTP
• 简单、标准，需要做的工作和维护工作少；几乎不需要做额外的工作即可与其他微服务集成

Message

认识Mseeage

• 使用消息队列进行分布式系统间的消息通信
• 在数据量大、对消息通信的时效性要求不是很高的场景下、可以考虑使用消息队列来进行微服务之间的通信

使用RestTemplate实现微服务通信

使用RestTemplate的两种方式（思想）

• 在代码（或配置文件中）写死IP和端口号
• 通过注册中心获取服务地址，可以实现负载均衡的效果

使用 SpringCloud Netfilx Ribbon 实现微服务通信及其原理

RibbonConfig

/**
 * @author qingtian
 * @version 1.0
 * @description: 使用 Ribbon 之前的配置，增强 RestTemplate
 * @date 2022/3/15 23:53
 */
@Component
public class RibbonConfig {

    @Bean
    @LoadBalanced
    public RestTemplate restTemplate() {
        return new RestTemplate();
    }
}

SpringCloud Netfilx Ribbon 实现原理

接下来 使用Ribbon的原生api实现负载均衡的功能 如下

/**
     * 使用ribbon 原生api操作
     * @param usernameAndPassword
     * @return
     */
    public JwtToken thinkingInRibbon(UsernameAndPassword usernameAndPassword) {

        String urlFormat = "http://%s/ecommerce-authority-center/authority/token";

        //1. 找到服务提供方的地址和端口号
        List<ServiceInstance> instances = discoveryClient.getInstances(CommonConstant.AUTHORITY_CENTER_SERVICE_ID);

        //构造服务列表
        List<Server> servers = new ArrayList<>(instances.size());
        instances.forEach(i -> {
            servers.add(new Server(i.getHost(),i.getPort()));
            log.info("found target instance : [{}] -> [{}]",i.getHost(),i.getPort());
        });

        //2. 使用负载均衡策略实现远端服务调用
        //构建 Ribbon 负载实例
        BaseLoadBalancer loadBalancer = LoadBalancerBuilder.newBuilder()
                .buildFixedServerListLoadBalancer(servers);

        //3. 设置负载均衡策略
        loadBalancer.setRule(new RetryRule(new RandomRule(),300));

        String result = LoadBalancerCommand.builder().withLoadBalancer(loadBalancer)
                .build().submit(server -> {

                    String targetUrl = String.format(
                            urlFormat,
                            String.format("%s:%s",server.getHost(),server.getPort())
                    );
                    log.info("target requestUrl is : [{}]",targetUrl);
                    HttpHeaders headers = new HttpHeaders();
                    headers.setContentType(MediaType.APPLICATION_JSON);

                    String tokenStr = new RestTemplate().postForObject(
                            targetUrl,
                            new HttpEntity<>(JSON.toJSONString(usernameAndPassword),headers),
                            String.class
                    );
                    return Observable.just(tokenStr);
                    //获取到请求的第一个结果的时候返回回来
                }).toBlocking().first().toString();
        return JSON.parseObject(result,JwtToken.class);
    }

OpenFegin的简单应用

@EnableDiscoveryClient
@SpringBootApplication
@EnableFeignClients
public class NacosClientApplication {

    public static void main(String[] args) {

        SpringApplication.run(NacosClientApplication.class,args);
    }
}

/**
 * @author qingtian
 * @version 1.0
 * @description: TODO
 * @date 2022/3/17 22:48
 */
@FeignClient(contextId = "AuthorityFeignClient", value = "e-commerce-authority-center")
public interface AuthorityFeignClient {

    @RequestMapping(value = "/ecommerce-authority-center/authority/token",
    method = RequestMethod.POST,consumes = "application/json",produces = "application/json")
    JwtToken getTokenByFeign(@RequestBody UsernameAndPassword usernameAndPassword);
}

@PostMapping("/token-by-feign")
    public JwtToken getTokenByFeign(@RequestBody UsernameAndPassword usernameAndPassword) {
        return authorityFeignClient.getTokenByFeign(usernameAndPassword);
    }

如何配置openFeign让它更加好用

SpringCloud OpenFeign 最常用的配置

• OpenFeign开启gzip压缩
• 统一OpenFeign使用配置：日志、重试、请求连接和响应时间限制
• 使用okhttp替换httpclient

在bootstrap.yml中添加配置

# Feign 的相关配置
feign:
  # feign 开启 gzip 压缩
  compression:
    request:
      enabled: true
      mime-types: text/xml,application/xml,application/json
      min-request-size: 1024
    response:
      enabled: true
  # 禁用默认的 http, 启用 okhttp
  httpclient:
    enabled: false
  okhttp:
    enabled: true
  # OpenFeign 集成 Hystrix
  hystrix:
    enabled: true

OpenFeign配置类

/**
 * @author qingtian
 * @version 1.0
 * @description: Feign配置类
 * @date 2022/3/17 23:16
 */
@Configuration
public class FeignConfig {

    /**
     * 开启 openFeign 日志
     * @return
     */
    @Bean
    public Logger.Level feignLogger() {
        return Logger.Level.FULL;
    }

    /**
     * OpenFeign 开启重试
     * period: 发起当前请求的时间间隔，单位是 ms
     * maxPeriod : 发起请求的最大时间间隔  单位是 ms
     * maxAttempts : 发起请求的最大次数
     * @return
     */
    @Bean
    public Retryer feignRetryer() {
        return new Retryer.Default(100,SECONDS.toMillis(1),5);
    }

    public static final int CONNECT_TIMEOUT_MILLS = 5000;
    public static final int READ_TIMEOUT_MILLS = 5000;

    /**
     * 对请求的链接和响应时间进行限制
     * @return
     */
    @Bean
    public Request.Options options() {
        return new Request.Options(
                CONNECT_TIMEOUT_MILLS, TimeUnit.MILLISECONDS,
                READ_TIMEOUT_MILLS, TimeUnit.MILLISECONDS,
                true
        );
    }
}

使用okhttp来替换自带的httpClient

<!-- feign 替换 JDK 默认的 URLConnection 为 okhttp -->
<dependency>
    <groupId>io.github.openfeign</groupId>
    <artifactId>feign-okhttp</artifactId>
</dependency>

# 禁用默认的 http, 启用 okhttp
  httpclient:
    enabled: false
  okhttp:
    enabled: true

新建OpenFeign配置类

/**
 * @author qingtian
 * @version 1.0
 * @description: okHttp配置
 * @date 2022/3/20 17:56
 */
@Configuration
@ConditionalOnClass(Feign.class)
@AutoConfigureBefore(FeignAutoConfiguration.class) //在feign配置前该配置就要生效
public class FeignHttpConfig {

    @Bean
    public okhttp3.OkHttpClient okHttpClient() {
        return new OkHttpClient().newBuilder()
                .connectTimeout(5, TimeUnit.SECONDS)    //设置连接超时
                .readTimeout(5,TimeUnit.SECONDS)    //设置读超时
                .writeTimeout(5,TimeUnit.SECONDS)   //设置写超时
                .retryOnConnectionFailure(true)     //是否自动重连
                // 配置连接池中的最大空闲线程个数为 10，并保持5分钟
                .connectionPool(new ConnectionPool(10,5L,TimeUnit.MINUTES))
                .build();
    }
}

通过feign的原生API其实现原理

使用feign的原生api

/**
 * @author qingtian
 * @version 1.0
 * @description: 使用feign的原生api
 * @date 2022/3/20 18:08
 */
@Slf4j
@Service
public class UseFeignApi {

    @Autowired
    private DiscoveryClient discoveryClient;

    /**
     * 使用feign的原生api调用远端服务
     * Feign 默认配置初始化，设置自定义配置，生成代理对象
     * @return
     */
    public JwtToken thinkingInFeign(UsernameAndPassword usernameAndPassword) {

        //通过反射拿到 serviceId
        String serviceId = null;
        Annotation[] annotations = AuthorityFeignClient.class.getAnnotations();
        for (Annotation annotation : annotations) {
            if (annotation.annotationType().equals(FeignClient.class)) {
                serviceId = ((FeignClient)annotation).value();
                log.info("get service Id from AuthorityFeignClient : [{}]",serviceId);
                break;
            }
        }

        //如果serviceId不存在，抛出异常
        if (null == serviceId) {
            throw new RuntimeException("can not get serviceId");
        }

        List<ServiceInstance> instances = discoveryClient.getInstances(serviceId);
        if (CollectionUtils.isEmpty(instances)) {
            throw new RuntimeException("can not get target instance from service id" + serviceId);
        }

        //随机选择一个服务实例，模拟负载均衡
        ServiceInstance randomInstance = instances.get(new Random().nextInt(instances.size()));
        log.info("choose service from instance : [{}],[{}],[{}]",serviceId,randomInstance.getHost(),randomInstance.getPort());

        AuthorityFeignClient feignClient = Feign.builder()      //feign 默认初始化配置
                .encoder(new GsonEncoder())                     //设置自定义配置
                .decoder(new GsonDecoder())                     //设置自定义配置
                .logLevel(Logger.Level.FULL)                    //设置自定义配置
                .target(AuthorityFeignClient.class,
                        String.format("http://%s:%s",
                                randomInstance.getHost(),randomInstance.getPort())      //生成feign的代理客户端
                );
        return feignClient.getTokenByFeign(usernameAndPassword);
    }
}

Feign客户端初始化的过程

• Feign客户端初始化包含三个部分

  

三种常用的微服务通信方案

• RPC效率高，可选实现方式多
• REST标准化程度搞，学习、使用成本低
• Message对削峰填谷有重大意义

Rest、Ribbon、OpenFeign一步步的演进

• Rest需要写死服务的ip和端口号（可以从注册中心手动获取），灵活性低

  /**
       * 通过注册中心拿到服务的信息（是所有的实例），再发起调用
        * @param usernameAndPassword
       * @return
       */
      public JwtToken getTokenFromAuthorityServiceWithLoadBalancer(UsernameAndPassword usernameAndPassword){
  
          ServiceInstance serviceInstance = loadBalancerClient.choose(CommonConstant.AUTHORITY_CENTER_SERVICE_ID);
          log.info("Nacos Client Info : [{}],[{}],[{}]",
                  serviceInstance.getServiceId(),serviceInstance.getInstanceId(),
                  JSON.toJSONString(serviceInstance.getMetadata()));
          String requestUrl = String.format(
                  "http://%s:%s/ecommerce-authority-center/authority/token",
                  serviceInstance.getHost(),
                  serviceInstance.getPort()
          );
          log.info("RestTemplate request url and body : [{}],[{}]",requestUrl, JSON.toJSONString(usernameAndPassword));
          HttpHeaders headers = new HttpHeaders();
          headers.setContentType(MediaType.APPLICATION_JSON);
          return new RestTemplate().postForObject(
                  requestUrl,
                  new HttpEntity<>(JSON.toJSONString(usernameAndPassword),headers),
                  JwtToken.class
          );
  
      }
  

• Ribbon提供基于restTemplate的http客户端并且支持服务负载均衡功能

• OPenFeign基于Ribbon，只需要使用注解和接口的配置即可完成对服务提供方的接口绑定

tips: OpenFeign改变Ribbon的负载均衡策略：

1. 使用配置文件更改内置的负载均衡策略

   ### 服务名，这个方式可以为单个服务配置想要的负载均衡策略
   provider02Nacosconfig:
     ribbon:
       NFLoadBalancerRuleClassName: com.netflix.loadbalancer.RandomRule
   

2. 修改JavaConfig类

   @Configuration
   public class FeignConfiguration {
       /**
        * 配置随机的负载均衡策略
        * 特点：对所有的服务都生效
        */
       @Bean
       public IRule loadBalancedRule() {
           return new RandomRule();
       }
   }
   

3. 自定义负载均衡策略

   /**
    * 自定义负载均衡算法
    */
   public class CustomRule implements IRule {
       private ILoadBalancer lb;
       private List<Integer> excludePorts;
   
       public CustomRule() {
       }
   
       public CustomRule(List<Integer> excludePorts) {
           this.excludePorts = excludePorts;
       }
   
       @Override
       public void setLoadBalancer(ILoadBalancer lb) {
           this.lb = lb;
       }
   
       @Override
       public ILoadBalancer getLoadBalancer() {
           return lb;
       }
   
       /**
        * 目标：自定义负载均衡策略：从所有可用的provider中排除掉指定端口号的provider，剩余provider进行随机选择
        * 实现步骤：
        * 1.获取到所有Server
        * 2.从所有Server中排除掉指定端口的Server后，剩余的Server
        * 3.从剩余Server中随机选择一个Server
        */
       @Override
       public Server choose(Object key) {
           // 1.获取到所有Server
           List<Server> servers = lb.getReachableServers();
           // 2.从所有Server中排除掉指定端口的Server后，剩余的Server
           List<Server> availableServers = this.getAvailableServers(servers);
           // 3.从剩余Server中随机选择一个Server
           return this.getAvailableRandomServers(availableServers);
       }
   
       private List<Server> getAvailableServers(List<Server> servers) {
           // 若没有指定要排除的port，则返回所有Server
           if(excludePorts == null || excludePorts.size() == 0) {
               return servers;
           }
           List<Server> aservers = servers.stream()
                   // filter()
                   // noneMatch() 只有当流中所有元素都没有匹配上时，才返回true，只要有一个匹配上了，则返回false
                   .filter(server -> excludePorts.stream().noneMatch(port -> server.getPort() == port))
                   .collect(Collectors.toList());
   
           return aservers;
       }
   
       private Server getAvailableRandomServers(List<Server> availableServers) {
           // 获取一个[0,availableServers.size())的随机数
           int index = new Random().nextInt(availableServers.size());
           return availableServers.get(index);
       }
   }
   

   修改JavaConfig类，使用自定义的负载均衡策略

   @Configuration
   public class FeignConfiguration {
       @Bean
       public IRule loadBalancedRule() {
           List<Integer> list = new ArrayList<>();
           list.add(8081);//排除访问端口
           return new CustomRule(list);
       }
   }