java - 每当两个流上发生写入时如何阻止读取？

Question

我正在尝试实现锁定，通过它我想避免在我进行写入时发生读取。

我的要求是:

• 读取阻塞，直到第一次设置所有两个映射。
• 现在第二次，如果我正在更新 map ，我仍然可以返回所有两个旧 map 值(在所有两个 map 上完成更新之前)或者它应该阻止并返回我所有新的两个 map 值每当对所有两张 map 都进行了更新。

因为我有两个 map - primaryMapping , secondaryMapping 所以它应该返回两个更新 map 的所有新值或者它应该返回 map 的所有旧值.基本上，在更新时我不想返回具有旧值的 primaryMapping 和具有新值的 secondaryMapping 。它应该是一致的，要么返回旧值，要么在更新 map 后返回新值。就我而言， map 更新将在 7 或 8 个月内发生一次(很少)。为此，我正在使用 Countdown Latch，它目前运行良好，没有任何问题。

现在我有两个流程，如下所示:对于每个流程，我都有一个 URL，我们从中获取上述两个 map 的数据。通常，我将为这两个流设置两个映射，因此与设备流相比，我们将对这两个映射的 PROCESS 流和 DEVICE 流设置不同的值。

public enum FlowType {
    PROCESS, DEVICE;
}

我有一个后台线程每 5 分钟运行一次，它从每个 FLOW url 获取数据，并在有更新时填充这两个 map 。当应用程序第一次启动时，它会更新映射，之后，它会在 7-8 个月后更新映射。并且这并不意味着两个流映射会同时发生变化。可能 PROCESS 映射发生了变化，但 DEVICE 映射没有发生变化。

public class DataScheduler {

    private RestTemplate restTemplate = new RestTemplate();
    private static final String PROCESS_URL = "http://process_flow_url/";
    private static final String DEVICE_URL = "http://device_flow_url/";
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

    public void startScheduleTask() {
        scheduler.scheduleAtFixedRate(new Runnable() {
            public void run() {
                try {
                    callService();
                } catch (Exception ex) {
                    // logging exception here using logger
                }
            }
        }, 0, 5, TimeUnit.MINUTES);
    }

    public void callService() throws Exception {
        String url = null;
        Map<FlowType, String> holder = new HashMap<FlowType, String>();
        for (FlowType flow : FlowType.values()) {
            try {
                url = getURL(flow);
                String response = restTemplate.getForObject(url, String.class);
                holder.put(flow, response);
            } catch (RestClientException ex) {
                // logging exception here using logger      
            }
        }
        parseResponse(holder);
    }

    private void parseResponse(Map<FlowType, String> responses) throws Exception {
        Map<FlowType, Mapping> partitionMapper = new HashMap<FlowType, Mapping>();
        boolean update = false;

        for (Map.Entry<FlowType, String> responseEntry : responses.entrySet()) {
            FlowType flow = responseEntry.getKey();
            String response = responseEntry.getValue();

            Map<String, Map<Integer, Integer>> primaryMapping = new HashMap<>();
            Map<String, Map<Integer, Integer>> secondaryMapping = new HashMap<>();

            if (!DataUtils.isEmpty(response)) {
                try (Scanner scanner = new Scanner(response)) {
                    boolean hasProcess = Boolean.parseBoolean(scanner.nextLine().trim().substring(HAS_PROCESS_LEN));
                    if (hasProcess) {
                        update = true;

                        // some code    
                        partitionMapper.put(flow, PartitionHolder.createMapping(
                                            primaryMapping, secondaryMapping));
                    }
                }
            }
        }
        // if there is any update, then only update the mappings, otherwise not.
        if (update) {
            PartitionHolder.setMappings(partitionMapper);
        }
    }
}

正如您在上面看到的，如果任何流映射已更新，它会通过调用 setMappings 方法来更新映射。

下面是我的 PartitionHolder 类:

public class PartitionHolder {

    public static class Mapping {
        public final Map<String, Map<Integer, Integer>> primaryMapping;
        public final Map<String, Map<Integer, Integer>> secondaryMapping;

        public Mapping(Map<String, Map<Integer, Integer>> primaryMapping,
                Map<String, Map<Integer, Integer>> secondaryMapping) {
            this.primaryMapping = primaryMapping;
            this.secondaryMapping = secondaryMapping;
        }

        // getters here
    }

    private static final AtomicReference<Map<FlowType, Mapping>> mappingsHolder = new AtomicReference<Map<FlowType, Mapping>>();
    private static final CountDownLatch hasInitialized = new CountDownLatch(1);

    public static Mapping getFlowMapping(FlowType flowType) {
        try {
            hasInitialized.await();
            return mappingsHolder.get().get(flowType);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
    }

    public static void setMappings(Map<FlowType, Mapping> newMapData) {
        mappingsHolder.set(newMapData);
        hasInitialized.countDown();
    }

    public static Mapping createMapping(
            Map<String, Map<Integer, Integer>> primaryMapping,
            Map<String, Map<Integer, Integer>> secondaryMapping) {
        return new Mapping(primaryMapping, secondaryMapping);
    }
}

现在这就是我在主线程中使用 PartitionHolder 类的方式。在一次调用中，我们将只获得一个流的映射。在下面的代码中，dataKey.getFlowType() 可以是 PROCESS OR DEVICE。

@Override
public DataResponse call() throws Exception {
    Mapping mappings = PartitionHolder.getFlowMapping(dataKey.getFlowType());
    // use mappings object here
}

现在，如您所见，在我上面的代码中，如果任何流映射已更新，我将通过调用 setMappings 方法来更新映射。有两种情况可能发生:

• 当应用程序首次启动时，它可以正常工作，没有任何问题，因为它会更新两个流的映射。
• 现在第二次，假设对于 PROCESS 流，映射已更新，但对于 DEVICE 流映射尚未更新，那么在这种情况下，它将通过调用 setMappings 方法更新整体映射，这将覆盖 mappingsHolder。这种方法的问题是，对于设备流，我将在调用 getFlowMapping 时开始获取空映射，因为它将设备映射覆盖为空映射。对吧？

如何避免第二个问题？除了使用 Map 和 key 作为 FlowType 之外，有没有更好的方法来解决这个问题？我需要在这里使用工厂模式来解决这个问题还是其他更好的方法？

Answer 1

ConcurrentHashMap 是线程安全的并且速度相对较快，因此可以简化问题:

private static final ConcurrentHashMap<FlowType, Mapping> mappingsHolder = 
        new ConcurrentHashMap<FlowType, Mapping>();

public static void setMappings(FlowType flowType, Mapping newMapData) {

    mappingsHolder.put(flowType, newMapData);
    hasInitialized.countDown();
}

public static Mapping getFlowMapping(FlowType flowType) {

    try {
        hasInitialized.await();
        return mappingsHolder.get(flowType);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new IllegalStateException(e);
    }
}

这应该可以解决第二个问题。