java – 并发配对

ConcurrentHashMap<String,Match> map = new ConcurrentHashMap<>();

void addMatch(Match match) {
    boolean success = false;
    while(!success) {
        Match oldMatch = map.remove(match.key);
        if(oldMatch != null) {
            match.setMatch(oldMatch);
            success = true;
       } else if(map.putIfAbsent(match.key,match) == null) {
            success = true;
       }
   }
}

您将继续循环,直到您将匹配添加到地图,或者直到您删除现有匹配并将其配对为止. remove和putIfAbsent都是原子的.

编辑：因为您要将数据卸载到磁盘,您可以使用例如MongoDB到此,用其findAndModify方法.如果具有密钥的对象已经存在,则该命令将删除并返回它,以便您可以将旧对象与新对象配对,并且可能存储与新密钥关联的对;如果具有该键的对象不存在,则该命令将该对象与该键一起存储.这相当于ConcurrentHashMap的行为,除了数据存储在磁盘而不是内存中;您不必担心同时写入两个对象,因为findAndModify逻辑可以防止它们无意中占用相同的密钥.

如果需要将对象序列化为JSON,请使用Jackson.

还有Mongo的替代品,例如DynamoDB,虽然Dynamo只免费提供少量数据.

编辑：鉴于朋友列表不是自反性的,我认为你可以通过组合MongoDB(或其他键值数据库与原子更新)和ConcurrentHashMap来解决这个问题.

> MongoDB中的人员“匹配”或“无与伦比”. (如果我说“从MongoDB中删除一个人”,我的意思是“将该人的状态设置为’匹配’.”)
>添加新人时,首先创建一个ConcurrentHashMap< key,boolean>对于它,可能在全局ConcurrentHashMap< key,ConcurrentHashMap< key,boolean>>中.
>通过新人的朋友迭代：
>如果朋友在MongoDB中,那么使用findAndModify将其原子地设置为“匹配”,然后将新人写入状态为“匹配”的MongoDB,最后将该对添加到MongoDB中的“Pairs”集合中最终用户查询.从全局地图中删除此人的ConcurrentHashMap.
>如果朋友不在MongoDB中,那么检查该朋友是否已写入当前朋友的相关ConcurrentHashMap.它有,然后什么都不做;如果没有,则检查朋友是否有与之关联的ConcurrentHashMap;如果是,则将与当前人的关键字相关联的值设置为“true”. (请注意,由于当前人员无法检查自己的地图并使用一个原子操作修改朋友的地图,因此两个朋友仍然可以写入彼此的哈希地图,但是自我哈希地图检查会降低这种可能性.)
>如果此人尚未匹配,则将其以“不匹配”状态写入MongoDB,从全局映射中删除其ConcurrentHashMap,并创建一个延迟任务,该任务将遍历写入该文件的所有朋友的ID. person的ConcurrentHashMap(即使用ConcurrentHashMap#keySet()).此任务的延迟应该是随机的(例如Thread.sleep(500 * rand.nextInt(30))),这样两个朋友就不会总是同时尝试匹配.如果当前的人没有任何需要重新检查的朋友,则不要为此创建延迟任务.
>当延迟结束时,为此人创建一个新的ConcurrentHashMap,从MongoDB中删除不匹配的人,然后循环回到步骤1.如果该人已经匹配,则不要将其从MongoDB中删除并终止延迟的任务.

在一般情况下,一个人要么与朋友匹配,要么在没有朋友加入系统的同时迭代通过朋友列表(即该人的ConcurrentHashMap将为空)时不匹配.如果同时写朋友：

Friend1和Friend2同时添加.

> Friend1写信给Friend2的ConcurrentHashMap,表示他们错过了对方.
> Friend2写入Friend1的ConcurrentHashMap来表示相同的情况(只有当Friend2在Friend1写入时同时检查Friend1是否写入其地图时才会出现这种情况 – 通常Friend2会检测到Friend1已写入其地图所以它不会写入Friend1的地图).
> Friend1和Friend2都写入MongoDB. Friend1在其后续任务中随机获得5秒延迟,Friend2随机获得15秒延迟.
> Friend1的任务首先触发,并与Friend2匹配.
> Friend2的任务激发第二; Friend2已不在MongoDB中,因此任务会立即终止.

一些打嗝：

> Friend1和Friend2可能都没有将ConcurrentHashMaps与它们相关联,例如：如果Friend2在Friend1检查以查看地图是否在内存中时仍在初始化其哈希映射.这很好,因为Friend2将写入Friend1的哈希映射,因此我们保证最终会尝试匹配 – 至少其中一个将具有哈希映射而另一个迭代,因为哈希映射创建在迭代之前.
>如果两个朋友的任务以某种方式同时解雇,则匹配的第二次迭代可能会失败.在这种情况下,如果朋友在匹配状态的MongoDB中,他们应该从其列表中删除朋友;然后,他们应该将结果列表的联合与写入其ConcurrentHashMap的朋友列表相关联,然后下一次迭代应该将其用作新朋友列表.最终该人将被匹配,否则该人的“重新检查”朋友列表将被清空.
>你应该增加每次后续迭代的任务延迟,以增加两个朋友的任务不会同时运行的可能性(例如Thread.sleep(500 * rand.nextInt(30))在第一次迭代,Thread.sleep (500 * rand.nextInt(60))在第二次迭代中,Thread.sleep(500 * rand.nextInt(90))在第三次迭代等).
>在后续迭代中,您必须在从MongoDB中删除人员之前创建一个人的新ConcurrentHashMap,否则您将进行数据竞争.同样,您必须在迭代其潜在匹配时从MongoDB中删除某个人,否则您可能会无意中将其匹配两次.

编辑：一些代码：

addUnmatchedToMongo(person1)方法将一个“不匹配”的person1写入MongoDB

setToMatched(friend1)使用findAndModify将friend1原子设置为“matched”;如果friend1已匹配或不存在,则该方法将返回false;如果更新成功,则返回true

如果friend1存在且匹配,则isMatched(friend1)返回true,如果它不存在或存在且“不匹配”,则返回false

private ConcurrentHashMap<String,ConcurrentHashMap<String,Person>> globalMap;
private DelayQueue<DelayedRetry> delayQueue;
private ThreadPoolExecutor executor;

executor.execute(new Runnable() {
    public void run() {
        while(true) {
            Runnable runnable = delayQueue.take();
            executor.execute(runnable);
        }
    }
}

public static void findMatch(Person person,Collection<Person> friends) {
    findMatch(person,friends,1);
}

public static void findMatch(Person person,Collection<Person> friends,int delayMultiplier) {
    globalMap.put(person.id,new ConcurrentHashMap<String,Person>());
    for(Person friend : friends) {
        if(**setToMatched(friend)**) {
            // write person to MongoDB in "matched" state
            // write "Pair(person,friend)" to MongoDB so it can be queried by the end user
            globalMap.remove(person.id);
            return;
        } else {
            if(**!isMatched(friend)** && globalMap.get(person.id).get(friend.id) == null) {
                // the existence of "friendMap" indicates another thread is currently  trying to match the friend
                ConcurrentHashMap<String,Person> friendMap = globalMap.get(friend.id);
                if(friendMap != null) {
                    friendMap.put(person.id,person);
                }
            }
        }
    }
    **addUnmatchedToMongo(person)**;
    Collection<Person> retryFriends = globalMap.remove(person.id).values();
    if(retryFriends.size() > 0) {
        delayQueue.add(new DelayedRetry(500 * new Random().nextInt(30 * delayMultiplier),person,retryFriends,delayMultiplier));
    }
}

public class DelayedRetry implements Runnable,Delayed {
    private final long delay;
    private final Person person;
    private final Collection<Person> friends;
    private final int delayMultiplier;

    public DelayedRetry(long delay,Person person,delayMultiplier) {
        this.delay = delay;
        this.person = person;
        this.friends = friends;
        this.delayMultiplier = delayMultiplier;
    }

    public long getDelay(TimeUnit unit) {
        return unit.convert(delay,TimeUnit.MILLISECONDS);
    }

    public void run {
        findMatch(person,delayMultiplier + 1);
    }
}