分布式并发控制

Question

我已经为此工作了几天，找到了几种解决方案，但没有一个非常简单或轻量级。问题基本上是这样的：我们有一个由 10 台机器组成的集群，每台机器都在多线程 ESB 平台上运行相同的软件。我可以相当容易地处理同一台机器上的线程之间的并发问题，但是不同机器上相同数据的并发呢？

本质上，该软件通过网络服务接收将客户数据从一家企业提供给另一家企业的请求。但是，客户可能存在也可能不存在于其他系统上。如果没有，我们通过 Web 服务方法创建它。所以它需要一种测试和设置，但我需要某种信号量来阻止其他机器引起竞争条件。之前我曾遇到过为一个本地客户创建两次远程客户的情况，这并不理想。

我在概念上尝试过的解决方案是：

1. 使用我们的容错共享文件系统创建“锁定”文件，每台机器都将根据客户进行检查

2. 在我们的数据库中使用一个特殊的表，并锁定整个表以便对锁定记录进行“测试和设置”。

3. 使用 Terracotta，这是一种有助于扩展的开源服务器软件，但使用的是中心辐射模型。

4. 使用 EHCache 同步复制我的内存中“锁”。

我无法想象我是唯一遇到过这种问题的人。你是怎么解决的？您是自己做的，还是有最喜欢的第 3 方产品？

Answer 1

您可能要考虑使用Hazelcast 分布式锁。超级精简和简单。

java.util.concurrent.locks.Lock lock = Hazelcast.getLock ("mymonitor");
lock.lock ();
try {
// do your stuff
}finally {
   lock.unlock();
}

Hazelcast - 分布式队列、映射、集合、列表、锁

Answer 2

我们使用的是 Terracotta，所以我想投赞成票。

我一直在关注 Hazelcast，它看起来像是另一种有前途的技术，但由于我没有使用过它，所以无法投票支持它，而且听说它使用基于 P2P 的系统，我真的不相信它可以满足大规模的扩展需求。

但我也听说过 Zookeeper，它来自 Yahoo，正在 Hadoop 保护伞下移动。如果你喜欢冒险尝试一些新技术，这真的很有希望，因为它非常精简和卑鄙，只专注于协调。我喜欢这个愿景和承诺，尽管它可能还是太绿了。

• http://www.terracotta.org
• http://wiki.apache.org/hadoop/ZooKeeper
• http://www.hazelcast.com

Answer 3

Terracotta 更接近于“分层”模型 - 所有客户端应用程序都与 Terracotta 服务器阵列通信（更重要的是，它们不会相互通信）。 Terracotta 服务器阵列能够以集群方式实现规模和可用性（镜像以实现可用性，条带化以实现规模化）。

在任何情况下，正如您可能知道的那样，Terracotta 使您能够通过使用 POJO 同步/等待/通知或使用任何 java.util.concurrent 原语，以与在单个 JVM 中相同的方式表达跨集群的并发性如 ReentrantReadWriteLock、CyclicBarrier、AtomicLong、FutureTask 等。

Terracotta Cookbook 中有很多简单的配方演示了这些原语的用法。

作为示例，我将发布 ReentrantReadWriteLock 示例（注意没有“Terracotta”版本的锁 - 您只需使用普通的 Java ReentrantReadWriteLock）

import java.util.concurrent.locks.*;

public class Main
{
    public static final Main instance = new Main();
    private int counter = 0;
    private ReentrantReadWriteLock rwl = new ReentrantReadWriteLock(true);

    public void read()
    {
        while (true) {
            rwl.readLock().lock();
                try {
                System.out.println("Counter is " + counter);
            } finally {
                rwl.readLock().unlock();
            }
            try { Thread.currentThread().sleep(1000); } catch (InterruptedException ie) {  }
        }
    }

    public void write()
    {
        while (true) {
            rwl.writeLock().lock();
            try {
               counter++;
               System.out.println("Incrementing counter.  Counter is " + counter);
            } finally {
                 rwl.writeLock().unlock();
            }
            try { Thread.currentThread().sleep(3000); } catch (InterruptedException ie) {  }
        }
    }

    public static void main(String[] args)
    {
        if (args.length > 0)  {
            // args --> Writer
            instance.write();
        } else {
            // no args --> Reader
            instance.read();
        }
    }
}

Answer 4

我建议使用Redisson。它实现了 30 多种分布式数据结构和服务，包括java.util.Lock。使用示例：

Config config = new Config();
config.addAddress("some.server.com:8291");
Redisson redisson = Redisson.create(config);

Lock lock = redisson.getLock("anyLock");
lock.lock();
try {
    ...
} finally {
   lock.unlock();
}

redisson.shutdown();

Answer 5

我打算建议使用 memcached 作为一种非常快速的分布式 RAM 存储来保存日志；但似乎 EHCache 是一个类似的项目，但更以 java 为中心。

只要您确定使用原子更新（memcached 支持它们，不了解 EHCache），任何一种都是可行的。这是迄今为止最具可扩展性的解决方案。

作为相关数据点，Google 使用“Chubby”，一种基于 RAM 的快速分布式锁存储作为多个系统的根，其中包括 BigTable。

Answer 6

我在 Coherence 方面做了很多工作，它允许使用多种方法来实现分布式锁。天真的方法是请求在所有参与节点上锁定相同的逻辑对象。在 Coherence 术语中，这是锁定复制缓存上的密钥。这种方法不能很好地扩展，因为网络流量会随着您添加节点而线性增加。更聪明的方法是使用分布式缓存，其中集群中的每个节点自然负责一部分密钥空间，因此将密钥锁定在这样的缓存中总是涉及与最多一个节点的通信。您可以根据这个想法推出自己的方法，或者更好的是，获得 Coherence。它确实是您梦寐以求的可扩展性工具包。

我要补充一点，任何半体面的基于多节点网络的锁定机制都必须相当复杂，才能在发生任何网络故障时正确动作。

Answer 7

不确定我是否理解整个上下文，但听起来您有 1 个单一数据库支持这一点？为什么不使用数据库的锁定：如果创建客户是单个 INSERT，那么这个语句单独可以用作锁定，因为数据库将拒绝违反您的约束之一的第二个 INSERT（例如，客户名称是例如唯一的）。

如果“插入客户”操作不是原子操作并且是一组语句，那么我将引入（或使用）一个初始 INSERT，它创建一些简单的基本记录来识别您的客户（具有必要的 UNIQUEness 约束），然后在同一事务中执行所有其他插入/更新。再次，数据库将负责一致性，任何并发修改都将导致其中一个失败。

Answer 8

我做了一个简单的RMI服务，有两种方法：锁定和释放。这两种方法都需要一个密钥（我的数据模型使用 UUID 作为 pk 所以这也是锁定密钥）。

RMI 是一个很好的解决方案，因为它是集中式的。您不能使用 EJB 执行此操作（特别是在集群中，因为您不知道您的呼叫将在哪台机器上进行）。另外，这很容易。

它对我有用。

Answer 9

如果您可以设置负载平衡，以便始终将单个客户的请求映射到同一台服务器，那么您可以通过本地同步来处理此问题。例如，将您的客户 ID mod 10 找出要使用的 10 个节点中的哪一个。

即使您不想在一般情况下这样做，您的节点也可以针对这种特定类型的请求相互代理。

假设您的用户足够统一（即，如果您有大量用户）以至于您不希望在一个节点过载的情况下会出现热点，那么这应该仍然可以很好地扩展。

Answer 10

您也可以考虑将Cacheonix 用于分布式锁。与这里提到的任何其他内容不同，Cacheonix 支持 ReadWrite 锁，并在需要时将锁升级从读取到写入：

ReadWriteLock rwLock = Cacheonix.getInstance().getCluster().getReadWriteLock();
Lock lock = rwLock.getWriteLock();
try {
  ...
} finally {
  lock.unlock();
}

全面披露：我是一名 Cacheonix 开发人员。

Answer 11

由于你已经连接到一个数据库，在添加另一个infra之前，先看看JdbcSemaphore，它使用起来很简单：

JdbcSemaphore semaphore = new JdbcSemaphore(ds, semName, maxReservations);
boolean acq = semaphore.acquire(acquire, 1, TimeUnit.MINUTES);
if (acq) {
 // do stuff
 semaphore.release();
} else {
  throw new TimeoutException();
}

它是spf4j 库的一部分。

Answer 12

过去，我们会使用网络上的特定“锁定服务器”来处理此问题。呵呵。

您的数据库服务器可能有专门用于执行此类操作的资源。 MS-SQL Server 具有可通过 sp_getapplock/sp_releaseapplock 过程使用的应用程序锁。

Answer 13

我们一直在开发一个开源的分布式同步框架，目前 DistributedReentrantLock 和 DistributedReentrantReadWrite 锁已经实现，但仍处于测试和重构阶段。在我们的架构中，锁密钥被划分在桶中，每个节点对一定数量的桶负责。因此，对于成功的锁定请求，只有一个网络请求。我们还使用 AbstractQueuedSynchronizer 类作为本地锁状态，因此所有失败的锁请求都在本地处理，这大大减少了网络流量。 我们使用 JGroups (http://jgroups.org) 进行组通信，使用 Hessian 进行序列化。

详情请查看http://code.google.com/p/vitrit/。

请将您的宝贵意见发给我。

卡姆兰