使用Spring Data GemFire从GemFire读取应用程序对象。使用SpringXD的gemfire-json-server存储的数据

Question

我正在使用SpringXD中的gemfire- json -server模块来使用“Order”对象的json表示来填充GemFire网格。我知道gemfire-json-server模块在GemFire中以Pdx的形式保存数据。我希望在我的应用程序中将GemFire网格的内容读入一个“Order”对象。我得到一个ClassCastException，它写道：

java.lang.ClassCastException: com.gemstone.gemfire.pdx.internal.PdxInstanceImpl cannot be cast to org.apache.geode.demo.cc.model.Order

我正在使用Spring Data GemFire库来读取集群的内容。用于读取Grid内容的代码片段如下：

public interface OrderRepository extends GemfireRepository<Order, String>{
    Order findByTransactionId(String transactionId);
}

如何使用Spring Data GemFire将从GemFire集群读取的数据转换为Order对象？注意:数据最初是使用SpringXD的GemFire -json-gemfire-module存储在服务器中的

Answer 1

仍然在等待GemFire PDX工程团队的反馈，特别是在Region.get(key)方面，但是，有趣的是，如果您使用...

@JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "@type")
public class Order ... {
  ...
}

这行得通！

实际上，我知道GemFire JSONFormatter类(参见here)使用Jackson的API来解组(解/序列化) PDX的JSON数据。

然而，orderRepository.findOne(ID)和ordersRegion.get(key)仍然不能像我期望的那样工作。有关更多详细信息，请参阅下面更新的测试类。

当我有更多的信息时，我会再次报告。

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(classes = GemFireConfiguration.class)
@SuppressWarnings("unused")
public class JsonToPdxToObjectDataAccessIntegrationTest {

  protected static final AtomicLong ID_SEQUENCE = new AtomicLong(0l);

  private Order amazon;
  private Order bestBuy;
  private Order target;
  private Order walmart;

  @Autowired
  private OrderRepository orderRepository;

  @Resource(name = "Orders")
  private com.gemstone.gemfire.cache.Region<Long, Object> orders;

  protected Order createOrder(String name) {
    return createOrder(ID_SEQUENCE.incrementAndGet(), name);
  }

  protected Order createOrder(Long id, String name) {
    return new Order(id, name);
  }

  protected <T> T fromPdx(Object pdxInstance, Class<T> toType) {
    try {
      if (pdxInstance == null) {
        return null;
      }
      else if (toType.isInstance(pdxInstance)) {
        return toType.cast(pdxInstance);
      }
      else if (pdxInstance instanceof PdxInstance) {
        return new ObjectMapper().readValue(JSONFormatter.toJSON(((PdxInstance) pdxInstance)), toType);
      }
      else {
        throw new IllegalArgumentException(String.format("Expected object of type PdxInstance; but was (%1$s)",
          pdxInstance.getClass().getName()));
      }
    }
    catch (IOException e) {
      throw new RuntimeException(String.format("Failed to convert PDX to object of type (%1$s)", toType), e);
    }
  }

  protected void log(Object value) {
    System.out.printf("Object of Type (%1$s) has Value (%2$s)", ObjectUtils.nullSafeClassName(value), value);
  }

  protected Order put(Order order) {
    Object existingOrder = orders.putIfAbsent(order.getTransactionId(), toPdx(order));
    return (existingOrder != null ? fromPdx(existingOrder, Order.class) : order);
  }

  protected PdxInstance toPdx(Object obj) {
    try {
      return JSONFormatter.fromJSON(new ObjectMapper().writeValueAsString(obj));
    }
    catch (JsonProcessingException e) {
      throw new RuntimeException(String.format("Failed to convert object (%1$s) to JSON", obj), e);
    }
  }

  @Before
  public void setup() {
    amazon = put(createOrder("Amazon Order"));
    bestBuy = put(createOrder("BestBuy Order"));
    target = put(createOrder("Target Order"));
    walmart = put(createOrder("Wal-Mart Order"));
  }

  @Test
  public void regionGet() {
    assertThat((Order) orders.get(amazon.getTransactionId()), is(equalTo(amazon)));
  }

  @Test
  public void repositoryFindOneMethod() {
    log(orderRepository.findOne(target.getTransactionId()));
    assertThat(orderRepository.findOne(target.getTransactionId()), is(equalTo(target)));
  }

  @Test
  public void repositoryQueryMethod() {
    assertThat(orderRepository.findByTransactionId(amazon.getTransactionId()), is(equalTo(amazon)));
    assertThat(orderRepository.findByTransactionId(bestBuy.getTransactionId()), is(equalTo(bestBuy)));
    assertThat(orderRepository.findByTransactionId(target.getTransactionId()), is(equalTo(target)));
    assertThat(orderRepository.findByTransactionId(walmart.getTransactionId()), is(equalTo(walmart)));
  }

  @Region("Orders")
  @JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "@type")
  public static class Order implements PdxSerializable {

    protected static final OrderPdxSerializer pdxSerializer = new OrderPdxSerializer();

    @Id
    private Long transactionId;

    private String name;

    public Order() {
    }

    public Order(Long transactionId) {
      this.transactionId = transactionId;
    }

    public Order(Long transactionId, String name) {
      this.transactionId = transactionId;
      this.name = name;
    }

    public String getName() {
      return name;
    }

    public void setName(final String name) {
      this.name = name;
    }

    public Long getTransactionId() {
      return transactionId;
    }

    public void setTransactionId(final Long transactionId) {
      this.transactionId = transactionId;
    }

    @Override
    public void fromData(PdxReader reader) {
      Order order = (Order) pdxSerializer.fromData(Order.class, reader);

      if (order != null) {
        this.transactionId = order.getTransactionId();
        this.name = order.getName();
      }
    }

    @Override
    public void toData(PdxWriter writer) {
      pdxSerializer.toData(this, writer);
    }

    @Override
    public boolean equals(Object obj) {
      if (obj == this) {
        return true;
      }

      if (!(obj instanceof Order)) {
        return false;
      }

      Order that = (Order) obj;

      return ObjectUtils.nullSafeEquals(this.getTransactionId(), that.getTransactionId());
    }

    @Override
    public int hashCode() {
      int hashValue = 17;
      hashValue = 37 * hashValue + ObjectUtils.nullSafeHashCode(getTransactionId());
      return hashValue;
    }

    @Override
    public String toString() {
      return String.format("{ @type = %1$s, id = %2$d, name = %3$s }",
        getClass().getName(), getTransactionId(), getName());
    }
  }

  public static class OrderPdxSerializer implements PdxSerializer {

    @Override
    public Object fromData(Class<?> type, PdxReader in) {
      if (Order.class.equals(type)) {
        return new Order(in.readLong("transactionId"), in.readString("name"));
      }

      return null;
    }

    @Override
    public boolean toData(Object obj, PdxWriter out) {
      if (obj instanceof Order) {
        Order order = (Order) obj;
        out.writeLong("transactionId", order.getTransactionId());
        out.writeString("name", order.getName());
        return true;
      }

      return false;
    }
  }

  public interface OrderRepository extends GemfireRepository<Order, Long> {
    Order findByTransactionId(Long transactionId);
  }

  @Configuration
  protected static class GemFireConfiguration {

    @Bean
    public Properties gemfireProperties() {
      Properties gemfireProperties = new Properties();

      gemfireProperties.setProperty("name", JsonToPdxToObjectDataAccessIntegrationTest.class.getSimpleName());
      gemfireProperties.setProperty("mcast-port", "0");
      gemfireProperties.setProperty("log-level", "warning");

      return gemfireProperties;
    }

    @Bean
    public CacheFactoryBean gemfireCache(Properties gemfireProperties) {
      CacheFactoryBean cacheFactoryBean = new CacheFactoryBean();

      cacheFactoryBean.setProperties(gemfireProperties);
      //cacheFactoryBean.setPdxSerializer(new MappingPdxSerializer());
      cacheFactoryBean.setPdxSerializer(new OrderPdxSerializer());
      cacheFactoryBean.setPdxReadSerialized(false);

      return cacheFactoryBean;
    }

    @Bean(name = "Orders")
    public PartitionedRegionFactoryBean ordersRegion(Cache gemfireCache) {
      PartitionedRegionFactoryBean regionFactoryBean = new PartitionedRegionFactoryBean();

      regionFactoryBean.setCache(gemfireCache);
      regionFactoryBean.setName("Orders");
      regionFactoryBean.setPersistent(false);

      return regionFactoryBean;
    }

    @Bean
    public GemfireRepositoryFactoryBean orderRepository() {
      GemfireRepositoryFactoryBean<OrderRepository, Order, Long> repositoryFactoryBean =
        new GemfireRepositoryFactoryBean<>();

      repositoryFactoryBean.setRepositoryInterface(OrderRepository.class);

      return repositoryFactoryBean;
    }
  }

}

Answer 2

因此，如您所知，GemFire (扩展为Apache Geode)以PDX格式(作为PdxInstance)存储JSON。这是因为GemFire可以与许多不同的基于语言的客户端(原生C++/C#，使用Developer REST API的面向web的(JavaScript，Pyhton，Ruby，等等))互操作，并且还能够使用OQL来查询JSON数据。

经过一些实验后，我很惊讶GemFire的表现并不像我所期望的那样。我创建了一个示例，自包含测试类(当然，没有Spring XD )来模拟您的用例……本质上是将JSON数据作为PDX存储在GemFire中，然后尝试使用Repository抽象将数据作为Order应用程序域对象类型读出。

如果使用了来自Spring Data GemFire的存储库抽象和实现，基础架构将尝试基于存储库通用类型参数(在本例中是来自"OrderRepository“定义的"Order”)来访问应用程序域对象。

但是，数据存储在PDX中，那么现在怎么办？

无论如何，Spring提供了MappingPdxSerializer类来使用与GemFire基础设施相同的“映射元数据”将PDX实例转换回应用程序域对象。太酷了，所以我把它插上...

@Bean
public CacheFactoryBean gemfireCache(Properties gemfireProperties) {
  CacheFactoryBean cacheFactoryBean = new CacheFactoryBean();

  cacheFactoryBean.setProperties(gemfireProperties);
  cacheFactoryBean.setPdxSerializer(new MappingPdxSerializer());
  cacheFactoryBean.setPdxReadSerialized(false);

  return cacheFactoryBean;
}

您还会注意到，我将PDX 'read-serialized‘属性(cacheFactoryBean.setPdxReadSerialized(false);)设置为false，以确保数据访问操作返回域对象，而不是PDX实例。

但是，这对查询方法没有影响。事实上，它对以下操作也没有影响...

orderRepository.findOne(amazonOrder.getTransactionId());

ordersRegion.get(amazonOrder.getTransactionId());

这两个调用都返回了一个PdxInstance。请注意，OrderRepository.findOne(..)的实现是基于SimpleGemfireRepository.findOne(key)的，它使用GemfireTemplate.get(key)，它只执行Region.get(key)，因此实际上与(ordersRegion.get(amazonOrder.getTransactionId();)相同。结果不应该是这样，特别是在Region.get()和read-serialized设置为false的情况下。

使用从findByTransactionId(String id)生成的OQL查询(SELECT * FROM /Orders WHERE transactionId = $1)，存储库基础结构对GemFire查询引擎将根据调用者(OrderRepository)的期望(基于泛型类型参数)返回的内容的控制较少，因此运行OQL语句的行为可能与使用get的直接区域访问不同。

接下来，我尝试修改Order类型以实现PdxSerializable，以处理数据访问操作期间的转换(使用get、OQL或其他方式进行直接区域访问)。这没有任何影响。

因此，我尝试为Order对象实现一个自定义的PdxSerializer。这也没有影响。

在这一点上，我唯一能得出的结论是，在Order -> JSON -> PDX和PDX -> Order之间的转换过程中，有些东西丢失了。似乎，GemFire需要PDX所需的额外类型的元数据(类似于PDXFormatter可以识别的JSON数据中的@JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "@type")，尽管我不确定它是否能识别。

请注意，在我的测试类中，我使用Jackson的ObjectMapper将Order序列化为JSON，然后使用GemFire的JSONFormatter将JSON序列化为PDX，我怀疑Spring也在做类似的事情。实际上，Spring XD使用Spring Data GemFire，并且很可能使用JSON Region Auto Proxy支持。这正是SDG的JSONRegionAdvice对象所做的(参见here)。

不管怎样，我有个问题要问GemFire工程团队的其他人。在Spring Data GemFire中还可以做一些事情来确保PDX数据被转换，比如如果数据确实是PdxInstance类型，那么直接使用MappingPdxSerializer来代表调用者自动转换数据。类似于JSON Region Auto Proxying的工作原理，您可以为Orders区域编写AOP拦截器，将PDX自动转换为Order。

不过，我认为这一切都没有必要，因为在这种情况下，GemFire应该做正确的事情。抱歉，我现在没有更好的答案。让我们看看我能发现什么。

干杯，敬请期待！

有关测试代码，请参阅后续文章。