8.12 zuul聚合微服务

妖狐艹你老母 2023-07-03 10:43 3阅读 0赞

使用RxJava结合 zuul来实现微服务

## 主方法改动 ##

@SpringBootApplication
    @EnableZuulProxy
    public class ZuulApplication { 
      public static void main(String[] args) { 
        SpringApplication.run(ZuulApplication.class, args);
      }
    
      @Bean
      @LoadBalanced
      public RestTemplate restTemplate() { 
        return new RestTemplate();
      }
    }

## 创建实体类 ##

public class User { 
      private Long id;
      private String username;
      private String name;
      private Integer age;
      private BigDecimal balance;
        //忽略get set
    }

## 创建service ##

@Service
    public class AggregationService { 
      @Autowired
      private RestTemplate restTemplate;
    
      @HystrixCommand(fallbackMethod = "fallback")
      public Observable<User> getUserById(Long id) { 
        // 创建一个被观察者
        return Observable.create(observer -> { 
          // 请求用户微服务的/{id}端点
          User user = restTemplate.getForObject("http://microservice-provider-user/{id}", User.class, id);
          observer.onNext(user);
          observer.onCompleted();
        });
      }
    
      @HystrixCommand(fallbackMethod = "fallback")
      public Observable<User> getMovieUserByUserId(Long id) { 
        return Observable.create(observer -> { 
          // 请求电影微服务的/user/{id}端点
          User movieUser = restTemplate.getForObject("http://microservice-consumer-movie/user/{id}", User.class, id);
          observer.onNext(movieUser);
          observer.onCompleted();
        });
      }
    
      public User fallback(Long id) { 
        User user = new User();
        user.setId(-1L);
        return user;
      }
    }

## 创建controller ##

@RestController
    public class AggregationController { 
      public static final Logger LOGGER = LoggerFactory.getLogger(ZuulApplication.class);
    
      @Autowired
      private AggregationService aggregationService;
    
      @GetMapping("/aggregate/{id}")
      public DeferredResult<HashMap<String, User>> aggregate(@PathVariable Long id) { 
        Observable<HashMap<String, User>> result = this.aggregateObservable(id);
        return this.toDeferredResult(result);
      }
    
        //根据参数，调用已经写好的service
      public Observable<HashMap<String, User>> aggregateObservable(Long id) { 
        // 合并两个或者多个Observables发射出的数据项，根据指定的函数变换它们
        return Observable.zip(
                this.aggregationService.getUserById(id),
                this.aggregationService.getMovieUserByUserId(id),
                (user, movieUser) -> { 
                  HashMap<String, User> map = Maps.newHashMap();
                  map.put("user", user);
                  map.put("movieUser", movieUser);
                  return map;
                }
        );
      }
    //添加一个Observable<HashMap<String, User>> 转换成：DeferredResult<HashMap<String, User>>
      public DeferredResult<HashMap<String, User>> toDeferredResult(Observable<HashMap<String, User>> details) { 
        DeferredResult<HashMap<String, User>> result = new DeferredResult<>();
        // 订阅
        details.subscribe(new Observer<HashMap<String, User>>() { 
          @Override
          public void onCompleted() { 
            LOGGER.info("完成...");
          }
    
          @Override
          public void onError(Throwable throwable) { 
            LOGGER.error("发生错误...", throwable);
          }
    
          @Override
          public void onNext(HashMap<String, User> movieDetails) { 
            result.setResult(movieDetails);
          }
        });
        return result;
      }
    }

*  http://localhost:8040/aggregate/1

{ 
        "movieUser": { 
            "id": 1,
            "username": "account1",
            "name": "张三",
            "age": 20,
            "balance": 100.00
        },
        "user": { 
            "id": 1,
            "username": "account1",
            "name": "张三",
            "age": 20,
            "balance": 100.00
        }
    }