Hazelcast 地图同步答案

【问题标题】：Hazelcast map synchronizationHazelcast 地图同步
【发布时间】：2013-06-19 04:22:06
【问题描述】：

我正在尝试在我的应用程序中使用 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);

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

【问题讨论】：

Hazelcast 为您提供对象的克隆（因为它以二进制/序列化的形式存储在集群中）。要使更新对其他节点/线程可见，您应该将其放回原处。此外，如果您的 Employee 类不是线程安全的，那么 JDK ConcurrentHashMap 不能保证更新对其他线程的可见性（有时即使您将其放回地图）。因此，无论哪种情况，您都应该使用同步机制。

标签： java hazelcast

【解决方案1】：

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

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

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

因此，您将使用自定义构造函数实现EntryProcessor，该构造函数接受一个包含您要更新的所有属性的对象，然后在process() 中构造最终的Employee 对象，该对象最终将出现在地图中并做一个entry.setValue()。不要忘记为可以序列化Employee 对象的EmployeeUpdateEntryProcessor 创建一个新的StreamSerializer，这样您就不会被java.io 序列化卡住。

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

【讨论】：

【解决方案2】：

您可能需要transaction。或者你可能想看看distributed lock。

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

【讨论】：

【解决方案3】：

This 你可能会感兴趣。

你可以为你的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);
        }
    }
}

这样，要修改缓存，您可以：

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

这是一个完全原子操作。

【讨论】：

感谢您的回复。我会试试这个。
这不是原子操作，因为你有读、修改、写。如果你真的想要一个原子操作，而不是借助锁或事务，你可以使用以下方法：' for(;;){ Person oldPerson = map.get("foo");人 newPerson = new Person(oldPerson); newPerson.incAge(); if(map.replace("foo",oldPerson,newPerson)) 休息； } ' 对于糟糕的布局感到抱歉..

【解决方案4】：

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

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

重读 hazelcast 的条目
更新它的字段
用替换保存到 hazelcast

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

【讨论】：

【解决方案5】：

您应该在地图条目上使用 tryLock：

    long timeout = 60;  // Define your own timeout 
    if (mapEmployees.tryLock(100, timeout, TimeUnits.SECONDS)){
       try {
          Employee emp1 = mapEmployees.get(100);
          emp1.setAge(30);
          emp1.setSex(“F”);
          emp1.setSalary(5000);
          mapEmployees.put(100, emp1);
       } finally {
           mapEmployees.unlock(100);
       }
   }else{
      // do something else like log.warn(...) 
   }

见：https://docs.hazelcast.com/imdg/4.2/data-structures/fencedlock#releasing-locks-with-trylock-timeout

【讨论】：