我正在尝试实现锁,通过它我想避免每当我进行写入时发生读取。
我的要求是:
- 读取 block ,直到第一次设置所有两个映射。
- 现在第二次,如果我更新 map ,我仍然可以返回所有两个旧 map 值(在所有两个 map 上完成更新之前),或者它应该阻止并返回所有新的两个 map 值两张 map 均已更新。
由于我有两个 map - primaryMapping
和 secondaryMapping
所以它应该返回两个更新 map 的所有新值,或者应该返回 map 的所有旧值。基本上,在更新时,我不想返回具有旧值的 primaryMapping
和具有新值的 secondaryMapping
。它应该是一致的,要么应该返回旧值,要么应该在更新 map 后返回新值。就我而言, map 每 7 或 8 个月更新一次(很少)。我为此使用倒计时锁存器,到目前为止它工作正常,没有任何问题。
现在我有两个流程,如下所示:对于每个流程,我都有一个 URL,我们可以从该 URL 获取上述两个 map 的数据。一般来说,我将为这两个流提供两个映射,因此与设备流相比,这两个映射的 PROCESS 流将具有不同的值。
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 或 DEVICE。
@Override
public DataResponse call() throws Exception {
Mapping mappings = PartitionHolder.getFlowMapping(dataKey.getFlowType());
// use mappings object here
}
现在,在我上面的代码中,如您所见,如果任何流映射已更新,我将通过调用 setMappings
方法来更新映射。有两种情况可能发生:
- 当应用程序第一次启动时,它可以正常工作,没有任何问题,因为它将更新两个流程的映射。
- 现在第二次,假设对于 PROCESS 流,映射已更新,但对于 DEVICE 流映射尚未更新,那么在这种情况下,它将通过调用
setMappings
方法来更新整体映射,该方法将覆盖mappingsHolder
。这种方法的问题是,对于 DEVICE 流,当我调用 getFlowMapping 时,我将开始获取空映射,因为它将 DEVICE 映射覆盖为空映射。对吗?
如何避免第二个问题?除了使用 Map 和 key 作为 FlowType 之外,还有其他更好的方法来解决这个问题吗?我需要在这里使用工厂模式来解决这个问题还是有其他更好的方法?
最佳答案
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);
}
}
这应该可以解决第二个问题。
关于java - 当两个流上发生写入时如何阻止读取?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/31104800/