更新 Pivotal GemFire 中的单个列答案

【问题标题】：Updating individual column in Pivotal GemFire更新 Pivotal GemFire 中的单个列
【发布时间】：2017-05-08 17:22:19
【问题描述】：

据我所知，没有选项可以使用 gemfire 中的查询来更新单个列。要更新单个列，我目前正在获取整个旧对象并修改更改的值并将其存储。如果有人在更新各个列方面实施了任何操作，请分享。

@Region("tracking")
public class Tracking implements Serializable {
public String id;
public String status;
public String program;
}



@Region("tracking")
public interface TrackingQueryRepository extends CrudRepository<Tracking, String> {
}

我是 Delta Propagation 实施的新手。我已阅读用户指南并尝试实施并收到下面给出的异常。你能分享你的想法吗这个。

Another.java – 域类

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.io.Serializable;
import org.springframework.data.annotation.Id;
import org.springframework.data.gemfire.mapping.Region;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.gemstone.gemfire.Delta;
import com.gemstone.gemfire.InvalidDeltaException;


@Region("delta")
public class Another implements Delta, Serializable {

    private static final long serialVersionUID = 1L;

    @Id
    private String anotherId;

    @JsonProperty("anotherProgramId")
    private String anotherProgramId;

    public Another() {
    }

    public Another(String anotherId, String anotherProgramId) {
        this.anotherId = anotherId;
        this.anotherProgramId = anotherProgramId;
    }

    public String getAnotherId() {
        return anotherId;
    }

    public void setAnotherId(String anotherId) {
        this.anotherIdChd = true;
        this.anotherId = anotherId;
    }

    public String getAnotherProgramId() {
        return anotherProgramId;
    }

    public void setAnotherProgramId(String anotherProgramId) {
        this.anotherProgramIdChd = true;
        this.anotherProgramId = anotherProgramId;
    }

    private transient boolean anotherIdChd = false;
    private transient boolean anotherProgramIdChd = false;

    @Override
    public String toString() {
        return "Another [anotherId=" + anotherId + ", anotherProgramId=" + anotherProgramId + "]";
    }

    @Override
    public void fromDelta(DataInput in) throws IOException, InvalidDeltaException {

        if (in.readBoolean()) {
            // Read the change and apply it to the object
            this.anotherId = in.toString();
            System.out.println(" Applied delta to field 'anotherId' = " + this.anotherId);
        }
        if (in.readBoolean()) {
            this.anotherProgramId = in.toString();
            System.out.println(" Applied delta to field 'anotherProgramId' = " + this.anotherProgramId);
        }
    }

    @Override
    public boolean hasDelta() {
        return this.anotherIdChd || this.anotherProgramIdChd;

    }

    @Override
    public void toDelta(DataOutput out) throws IOException {
        System.out.println("Extracting delta from " + this.toString());
        out.writeBoolean(anotherIdChd);
        if (anotherIdChd) {
            // Write just the changes into the data stream

            out.writeUTF(this.anotherId);
            // Once the delta information is written, reset the delta status
            // field
            this.anotherIdChd = false;
            System.out.println(" Extracted delta from field 'anotherId' = " + this.anotherId);
        }
        out.writeBoolean(anotherProgramIdChd);
        if (anotherProgramIdChd) {
            out.writeUTF(this.anotherProgramId);
            this.anotherProgramIdChd = false;
            System.out.println(" Extracted delta from field 'anotherProgramId' = " + this.anotherProgramId);
        }

    }

}

客户端缓存.xml

<pdx>
        <pdx-serializer>
            <class-name>com.gemstone.gemfire.pdx.ReflectionBasedAutoSerializer</class-name>
            <parameter name="classes">
                <string>com\.rs\.main\..+</string>
            </parameter>
        </pdx-serializer>
    </pdx>

Spring XML 命名空间

<util:properties id="gemfire-props">
<prop key="delta-propagation">true</prop>
</util:properties>
<gfe:client-cache pool-name="serverPool" cache-xml-location="classpath:client-cache.xml" properties-ref="gemfire-props"/>
<gfe:client-region id="delta" pool-name="serverPool" shortcut="PROXY" cloning-enabled="true">

本地 gemfire 实例版本 - pivotal-gemfire-9.0.1

区域创建 创建区域 –name=delta –type=REPLICATE

例外：

2017-05-08 22:17:12.370 ERROR 14696 --- [nio-8080-exec-1] o.a.c.c.C.[.[.[/].[dispatcherServlet]    : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.springframework.dao.DataAccessResourceFailureException: remote server on 10.148.210.249(:loner):53784:e10627eb: com.gemstone.gemfire.pdx.PdxSerializationException: Could not create an instance of a class com.rs.main.Another; nested exception is com.gemstone.gemfire.cache.client.ServerOperationException: remote server on 10.148.210.249(:loner):53784:e10627eb: com.gemstone.gemfire.pdx.PdxSerializationException: Could not create an instance of a class com.rs.main.Another] with root cause

java.lang.ClassNotFoundException: com.rs.main.Another
    at org.apache.geode.internal.ClassPathLoader.forName(ClassPathLoader.java:437) ~[na:na]
    at org.apache.geode.internal.InternalDataSerializer.getCachedClass(InternalDataSerializer.java:4010) ~[na:na]
    at org.apache.geode.pdx.internal.PdxType.getPdxClass(PdxType.java:235) ~[na:na]
    at org.apache.geode.pdx.internal.PdxReaderImpl.basicGetObject(PdxReaderImpl.java:687) ~[na:na]
    at org.apache.geode.pdx.internal.PdxReaderImpl.getObject(PdxReaderImpl.java:682) ~[na:na]
    at org.apache.geode.internal.InternalDataSerializer.readPdxSerializable(InternalDataSerializer.java:3218) ~[na:na]
    at org.apache.geode.internal.InternalDataSerializer.basicReadObject(InternalDataSerializer.java:3005) ~[na:na]
    at org.apache.geode.DataSerializer.readObject(DataSerializer.java:2897) ~[na:na]
    at org.apache.geode.internal.util.BlobHelper.deserializeBlob(BlobHelper.java:90) ~[na:na]
    at org.apache.geode.internal.cache.EntryEventImpl.deserialize(EntryEventImpl.java:1891) ~[na:na]
    at org.apache.geode.internal.cache.EntryEventImpl.deserialize(EntryEventImpl.java:1884) ~[na:na]
    at org.apache.geode.internal.cache.VMCachedDeserializable.getDeserializedValue(VMCachedDeserializable.java:134) ~[na:na]
    at org.apache.geode.internal.cache.EntryEventImpl.processDeltaBytes(EntryEventImpl.java:1687) ~[na:na]
    at org.apache.geode.internal.cache.EntryEventImpl.setNewValueInRegion(EntryEventImpl.java:1558) ~[na:na]
    at org.apache.geode.internal.cache.EntryEventImpl.putExistingEntry(EntryEventImpl.java:1504) ~[na:na]
    at org.apache.geode.internal.cache.AbstractRegionMap.updateEntry(AbstractRegionMap.java:2959) ~[na:na]
    at org.apache.geode.internal.cache.AbstractRegionMap.basicPut(AbstractRegionMap.java:2782) ~[na:na]
    at org.apache.geode.internal.cache.LocalRegion.virtualPut(LocalRegion.java:5750) ~[na:na]
    at org.apache.geode.internal.cache.DistributedRegion.virtualPut(DistributedRegion.java:337) ~[na:na]
    at org.apache.geode.internal.cache.LocalRegionDataView.putEntry(LocalRegionDataView.java:151) ~[na:na]
    at org.apache.geode.internal.cache.LocalRegion.basicUpdate(LocalRegion.java:5730) ~[na:na]
    at org.apache.geode.internal.cache.LocalRegion.basicBridgePut(LocalRegion.java:5374) ~[na:na]
    at org.apache.geode.internal.cache.tier.sockets.command.Put65.cmdExecute(Put65.java:381) ~[na:na]
    at org.apache.geode.internal.cache.tier.sockets.BaseCommand.execute(BaseCommand.java:141) ~[na:na]
    at org.apache.geode.internal.cache.tier.sockets.ServerConnection.doNormalMsg(ServerConnection.java:776) ~[na:na]
    at org.apache.geode.internal.cache.tier.sockets.ServerConnection.doOneMessage(ServerConnection.java:904) ~[na:na]
    at org.apache.geode.internal.cache.tier.sockets.ServerConnection.run(ServerConnection.java:1160) ~[na:na]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) ~[na:1.8.0_121]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ~[na:1.8.0_121]
    at org.apache.geode.internal.cache.tier.sockets.AcceptorImpl$1$1.run(AcceptorImpl.java:519) ~[na:na]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_121]

【问题讨论】：

标签： spring-boot gemfire spring-data-gemfire

【解决方案1】：

嗨（再次）Vigneshwaran-

是的，所以 GemFire 的 Query capabilities（通过 QueryService）严格用于运行查询（即 SELECT 语句）。 GemFire OQL 中没有用于更新和删除的等效项。 GemFire 是一个具有类似Map 的操作（例如get(key)、put(key, value) 等）的键/值存储，您通常在其中使用整个应用程序域对象。但是，无论您的应用程序是对等缓存（即集群的成员）还是缓存客户端，GemFire 都有一些功能可以帮助您。通常，应用程序是缓存客户端并拥有/使用ClientCache，其中集群是独立的，客户端连接到集群很像 RDBMS。

我还要说，虽然 Function 服务很有用，但它不是唯一的选择，实际上可能在代码方面的开销更大。

正如 Wes 上面提到的，使用 PARTITION Region 是非常典型的，尤其是对于“事务性”数据（注意：REPLICATE Regions 更适用于不经常更改的参考数据）。

“功能”可以帮助您，您可以编写Function 来更新应用程序域对象。 “更新”可以在函数的“参数”中传递。要调用函数，您可以使用 GemFire 的 FunctionService 来获取 Execution，使用目标方法之一（例如 [onRegion("tracking")][7]）。

注意：其他定位方法（即onMember(s) 和onServer(s)）分别特定于您的应用程序是“对等”还是“客户端”。例如，如果您的应用程序是客户端，则不能调用onMember(s)，因为它假定您的应用程序是“对等体”。同样，如果您的应用程序是对等体，则不能调用onServer(s)，因为它假定您的应用程序是“客户端”。 onRegion(..) 无论应用程序是对等还是客户端都有效。虽然您可能会想为什么不一直使用onRegion，但根据您的 UC 使用其他形式的定位具有技术优势（例如，考虑服务器组和路由）。总之……

当 Region 为 PARTITION 时，还可以设置 Function 的 [optimizeForWrite()][8]，表示 Function 将更新 Region 数据，因此，当指定 key 时，将被路由到 PARTITION 的主存储桶using the filtering 选项，如上述 Wes 所述。

PARTITION Region 的一致性来自于这样一个事实，即所有更新都首先路由并写入“主节点”（无论哪个服务器接收客户端的更新，这可能是一个甚至不托管该 Region 或数据/密钥的服务器有问题；即不同的分片）。更新主节点后，数据更改将传播（分发）到集群中托管分区/分片数据集的辅助节点的其他节点。这就是 Wes 上面提到的“事务性”一致性。

注意：分区只是对数据进行分片的另一个词，其中数据均匀分布在可用节点的集群中。添加/删除节点时，数据会重新平衡。 PARTITION 也可以有冗余。这些被称为次级。 PARTITION 区域有助于延迟和吞吐量，因为数据被划分（默认为 113 个存储桶），其中每个存储桶具有主副本和可能 1 个或多个副本（辅助，用于冗余；HA），从而提高读取和写入吞吐量。

另外，如果数据必须粘贴，那么你也可以设置Function的HA属性。这将允许在失败的情况下重试。

然而，尽管有所有这些优势，您仍然必须在服务器上的 Function 中处理“如何”更新您的应用程序域对象。您还必须处理“映射”，因为在 GemFire 这样的键/值存储中确实没有与 ORM 等效的东西。当然，这并不难，但也许有更好的方法。

还有另一个功能，称为Delta Propagation。从本质上讲，每当您进行更新时，您总是在获取和更新 GemFire 中的全部价值。

注意：可以以类似投影的方式查询对象的选择字段，但无论如何它不是代理或与实际对象相关。

当您利用 GemFire 的 Serialization 功能时，您可以 leverage Delta Propagation。

在实现“Deltas”时，实际上只有应用程序域对象中的差异被序列化，通过线路发送，无论是在客户端和服务器之间，还是在支持冗余策略的对等方之间。这对您来说是完全无缝的。你得到你的对象（客户端），更新它，然后放它。 GemFire 为您处理发送“delta”的逻辑。

此外，在集群中的服务器上使用client/server topology 和 PARTITION 区域时，您可以启用Single-Hop access，它可以有效地将数据路由到包含“主”存储桶的服务器，从而避免额外的网络跃点，从而会影响您对每次操作的感知延迟。

因此，在 Delta 和 Single-Hop 之间，您最终会得到一个性能相当不错的解决方案，并且仍然可以利用面向对象的方法，如您所期望的那样使用您的应用程序域对象 API。

不过，请注意pitfalls of using Deltas。

无论如何，值得深思。您通常总是有不止一种方法来完成一项任务，但更好的方法并不总是显而易见的，除非您根据您的 UC/目标来衡量和评估所需的效果。

干杯，约翰

【讨论】：

嗨，John，这个应用程序是一个缓存客户端应用程序，并且已经使用了 Gemfire 的序列化功能。我们决定进行 Delta Propagation。您的解释非常准确且非常有帮助。我想实施，然后接受你的回答。非常感谢你。您的帮助对我来说非常宝贵。
我尝试过 Delta Propagation 不适合我，因为 PDX。尝试使用函数 onRegion().withFilter(key).withArgs(columnsAndValuesMap) 来实现。您能否分享为服务器功能（执行方法）编写的代码 sn-ps。提前致谢

【解决方案2】：

您无法使用查询服务更新列。我建议您考虑使用 Function 服务来实现事务一致性。对区域进行分区并使用 .onRegion().withFilter(key).withArgs(columnsAndValuesMap) 调用该函数。

您的函数将读取对象、应用更新并放置。

通过这种方式，您的读取和更新将在服务器上的单个线程中进行，从而确保事务一致性，而不是在客户端读取对象、更改值、执行 put 并希望没有其他人在您身下溜进来。

【讨论】：

Wes - 您能否分享函数代码 sn-p 以在函数中处理此场景。感谢您的帮助。

【解决方案3】：

我们实现相同目标的另一种方法是使用自定义函数，该函数在获取分布式锁后更新值 (BeanUtils)。

这可能会增加性能开销，但可以保证数据完整性。这就是权衡。

请看下面的伪代码

 try{

 //this can be regionName 
 dls = DistributedLockService.getServiceNamed(arbitrary-lock-name) 
 //the key is normally the object @Id
 dls.lock(some-key, waitTimeOut, leaseTimeOut)

 row = region.get(id)
 //Here we copy the desired value (input to function) to the latest value
 BeanUtils.copyProperty(row, key, value);      
 //Insert the modified record to Gemfire - now this becomes equivalent of update <region> set value =  for a specific property.
 region.put(id, row) 

 } finally{

 dls.unlock(some-key);

 }

【讨论】：