java Hazelcast 地图同步

Hazelcast map synchronization

提问by Ananth

I am trying to implement distributed cache using Hazelcast in my application. I am using Hazelcast's IMap. The problem I have is every time I get a value from a map and update the value, I need to do a put(key, value)again. If my value object has 10 properties and I have to update all 10, then I have to call put(key, value)10 times. Something like -

我正在尝试在我的应用程序中使用 Hazelcast 实现分布式缓存。我正在使用 Hazelcast 的IMap. 我遇到的问题是每次从地图中获取值并更新该值时，我都需要put(key, value)再次执行。如果我的值对象有 10 个属性并且我必须更新所有 10 个属性，那么我必须调用put(key, value)10 次。就像是 -

IMap<Integer, Employee> mapEmployees = hz.getMap("employees");
Employee emp1 = mapEmployees.get(100);
emp1.setAge(30);
mapEmployees.put(100, emp1);
emp1.setSex(“F”);
mapEmployees.put(100, emp1);
emp1.setSalary(5000);
mapEmployees.put(100, emp1);

If I don't do this way, some other node which operates on the same Employee object will update it and the final result is that the employee object is not synchronized. Is there any solution to avoid calling put explicitly multiple times? In a ConcurrentHashMap, I don't need to do this because if I change the object, the map also gets updated.

如果我不这样做，在同一个 Employee 对象上操作的其他一些节点将更新它，最终结果是雇员对象没有同步。有什么解决方案可以避免多次显式调用 put 吗？在 a 中ConcurrentHashMap，我不需要这样做，因为如果我更改对象，地图也会更新。

回答by Drew

As of version 3.3 you'll want to use an EntryProcessor:

从 3.3 版开始，您将需要使用 EntryProcessor：

What you really want to do here is build an EntryProcessor<Integer, Employee>and call it using mapEmployees.executeOnKey( 100, new EmployeeUpdateEntryProcessor( new ObjectContainingUpdatedFields( 30, "F", 5000 ) );

你真正想做的是构建一个EntryProcessor<Integer, Employee>并使用它来调用它 mapEmployees.executeOnKey( 100, new EmployeeUpdateEntryProcessor( new ObjectContainingUpdatedFields( 30, "F", 5000 ) );

This way, Hazelcast handles locking the map on the key for that Employee object and allows you to run whatever code is in the EntryProcessor's process()method atomically including updating values in the map.

通过这种方式，Hazelcast 处理锁定该 Employee 对象的键上的映射，并允许您以process()原子方式运行 EntryProcessor方法中的任何代码，包括更新映射中的值。

So you'd implement EntryProcessorwith a custom constructor that takes an object that contains all of the properties you want to update, then in process()you construct the final Employeeobject that will end up in the map and do an entry.setValue(). Don't forget to create a new StreamSerializerfor the EmployeeUpdateEntryProcessorthat can serialize Employeeobjects so that you don't get stuck with java.io serialization.

因此，您将EntryProcessor使用一个自定义构造函数来实现，该构造函数采用一个包含您要更新的所有属性的对象，然后在process()您构造最终Employee将出现在地图中的对象中并执行entry.setValue(). 不要忘了创建一个新StreamSerializer的EmployeeUpdateEntryProcessor是可以序列化Employee的对象，这样你就不会遇到问题与java.io序列化。

Source: http://docs.hazelcast.org/docs/3.5/manual/html/entryprocessor.html

来源：http: //docs.hazelcast.org/docs/3.5/manual/html/entryprocessor.html

回答by Piotr Gwiazda

Probably a transactionis what you need. Or you may want to take a look at distributed lock.

可能交易正是您所需要的。或者您可能想看看分布式锁。

Note that in your solution if this code is ran by two threads changes made by one of them will be overwriten.

请注意，在您的解决方案中，如果此代码由两个线程运行，其中一个线程所做的更改将被覆盖。

回答by fge

Thismay interest you.

这可能会让您感兴趣。

You could do something like this for your Employeeclass (simplified code with one instance variable only):

你可以为你的Employee类做这样的事情（只有一个实例变量的简化代码）：

public final class Employee
    implements Frozen<Builder>
{
    private final int salary;

    private Employee(Builder builder)
    {
        salary = builder.salary;
    }

    public static Builder newBuilder()
    {
        return new Builder();
    }

    @Override
    public Builder thaw()
    {
        return new Builder(this);
    }

    public static final class Builder
        implements Thawed<Employee>
    {
        private int salary;

        private Builder()
        {
        }

        private Builder(Employee employee)
        {
            salary = employee.salary;
        }

        public Builder withSalary(int salary)
        {
            this.salary = salary;
            return this;
        }

        @Override
        public Employee freeze()
        {
            return new Employee(this);
        }
    }
}

This way, to modify your cache, you would:

这样，要修改缓存，您将：

Employee victim = map.get(100);
map.put(100, victim.thaw().withSalary(whatever).freeze());

This is a completely atomic operation.

这是一个完全原子的操作。

回答by Sneg

If there is possibility that another node can update data that your node is working with then using put() will overwrite changes made by another node. Usually it is unwanted behavior, cause it leads to data loss and inconsistent data state.

如果另一个节点有可能更新您的节点正在使用的数据，那么使用 put() 将覆盖另一个节点所做的更改。通常这是不需要的行为，导致数据丢失和数据状态不一致。

Take a look at IMap.replace()method and other ConcurrentMaprelated methods. If replace()is failed then you've faced changes collision. In this case you should give it another attempt:

看看IMap.replace()方法和其他ConcurrentMap相关方法。如果replace()失败，那么您将面临更改冲突。在这种情况下，您应该再试一次：

1. re-read entry from hazelcast
2. update it's fields
3. save to hazelcast with replace

1. 从hazelcast 重读条目
2. 更新它的字段
3. 用替换保存到榛子铸

After some failed attempts you can throw StorageException to the upper level.

在一些失败的尝试之后，您可以将 StorageException 抛出到上层。