- iOS/Objective-C 元类和类别
- objective-c - -1001 错误,当 NSURLSession 通过 httpproxy 和/etc/hosts
- java - 使用网络类获取 url 地址
- ios - 推送通知中不播放声音
我有一个类,我在 updateLiveSockets()
方法中每 30 秒从一个后台线程填充一个映射 liveSocketsByDatacenter
然后我有一个方法 getNextSocket()
将由多个读取器线程调用以获取可用的实时套接字,该套接字使用相同的映射来获取此信息。
public class SocketManager {
private static final Random random = new Random();
private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
private final AtomicReference<Map<Datacenters, List<SocketHolder>>> liveSocketsByDatacenter =
new AtomicReference<>(Collections.unmodifiableMap(new HashMap<>()));
private final ZContext ctx = new ZContext();
// Lazy Loaded Singleton Pattern
private static class Holder {
private static final SocketManager instance = new SocketManager();
}
public static SocketManager getInstance() {
return Holder.instance;
}
private SocketManager() {
connectToZMQSockets();
scheduler.scheduleAtFixedRate(new Runnable() {
public void run() {
updateLiveSockets();
}
}, 30, 30, TimeUnit.SECONDS);
}
// during startup, making a connection and populate once
private void connectToZMQSockets() {
Map<Datacenters, ImmutableList<String>> socketsByDatacenter = Utils.SERVERS;
// The map in which I put all the live sockets
Map<Datacenters, List<SocketHolder>> updatedLiveSocketsByDatacenter = new HashMap<>();
for (Map.Entry<Datacenters, ImmutableList<String>> entry : socketsByDatacenter.entrySet()) {
List<SocketHolder> addedColoSockets = connect(entry.getKey(), entry.getValue(), ZMQ.PUSH);
updatedLiveSocketsByDatacenter.put(entry.getKey(),
Collections.unmodifiableList(addedColoSockets));
}
// Update the map content
this.liveSocketsByDatacenter.set(Collections.unmodifiableMap(updatedLiveSocketsByDatacenter));
}
private List<SocketHolder> connect(Datacenters colo, List<String> addresses, int socketType) {
List<SocketHolder> socketList = new ArrayList<>();
for (String address : addresses) {
try {
Socket client = ctx.createSocket(socketType);
// Set random identity to make tracing easier
String identity = String.format("%04X-%04X", random.nextInt(), random.nextInt());
client.setIdentity(identity.getBytes(ZMQ.CHARSET));
client.setTCPKeepAlive(1);
client.setSendTimeOut(7);
client.setLinger(0);
client.connect(address);
SocketHolder zmq = new SocketHolder(client, ctx, address, true);
socketList.add(zmq);
} catch (Exception ex) {
// log error
}
}
return socketList;
}
// this method will be called by multiple threads to get the next live socket
// is there any concurrency or thread safety issue or race condition here?
public Optional<SocketHolder> getNextSocket() {
// For the sake of consistency make sure to use the same map instance
// in the whole implementation of my method by getting my entries
// from the local variable instead of the member variable
Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter =
this.liveSocketsByDatacenter.get();
Optional<SocketHolder> liveSocket = Optional.absent();
List<Datacenters> dcs = Datacenters.getOrderedDatacenters();
for (Datacenters dc : dcs) {
liveSocket = getLiveSocket(liveSocketsByDatacenter.get(dc));
if (liveSocket.isPresent()) {
break;
}
}
return liveSocket;
}
// is there any concurrency or thread safety issue or race condition here?
private Optional<SocketHolder> getLiveSocketX(final List<SocketHolder> endpoints) {
if (!CollectionUtils.isEmpty(endpoints)) {
// The list of live sockets
List<SocketHolder> liveOnly = new ArrayList<>(endpoints.size());
for (SocketHolder obj : endpoints) {
if (obj.isLive()) {
liveOnly.add(obj);
}
}
if (!liveOnly.isEmpty()) {
// The list is not empty so we shuffle it an return the first element
Collections.shuffle(liveOnly);
return Optional.of(liveOnly.get(0));
}
}
return Optional.absent();
}
// Added the modifier synchronized to prevent concurrent modification
// it is needed because to build the new map we first need to get the
// old one so both must be done atomically to prevent concistency issues
private synchronized void updateLiveSockets() {
Map<Datacenters, ImmutableList<String>> socketsByDatacenter = Utils.SERVERS;
// Initialize my new map with the current map content
Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter =
new HashMap<>(this.liveSocketsByDatacenter.get());
for (Entry<Datacenters, ImmutableList<String>> entry : socketsByDatacenter.entrySet()) {
List<SocketHolder> liveSockets = liveSocketsByDatacenter.get(entry.getKey());
List<SocketHolder> liveUpdatedSockets = new ArrayList<>();
for (SocketHolder liveSocket : liveSockets) { // LINE A
Socket socket = liveSocket.getSocket();
String endpoint = liveSocket.getEndpoint();
Map<byte[], byte[]> holder = populateMap();
Message message = new Message(holder, Partition.COMMAND);
boolean status = SendToSocket.getInstance().execute(message.getAdd(), holder, socket);
boolean isLive = (status) ? true : false;
// is there any problem the way I am using `SocketHolder` class?
SocketHolder zmq = new SocketHolder(socket, liveSocket.getContext(), endpoint, isLive);
liveUpdatedSockets.add(zmq);
}
liveSocketsByDatacenter.put(entry.getKey(),
Collections.unmodifiableList(liveUpdatedSockets));
}
this.liveSocketsByDatacenter.set(Collections.unmodifiableMap(liveSocketsByDatacenter));
}
}
正如你在我的类里面看到的:
updateLiveSockets()
方法中使用所有实时套接字填充 liveSocketsByDatacenter
映射。getNextSocket()
方法给我一个可用的实时套接字,它使用 liveSocketsByDatacenter
映射来获取所需的信息。<我的代码工作正常,没有任何问题,想看看是否有更好或更有效的方法来编写它。我还想就线程安全问题或任何竞争条件(如果有的话)发表意见,但到目前为止我还没有看到任何意见,但我可能是错的。
我最担心的是 updateLiveSockets()
方法和 getLiveSocketX()
方法。我正在迭代 liveSockets
,这是 A 行 SocketHolder
的 List
,然后创建一个新的 SocketHolder
对象并添加到另一个新列表。这里可以吗?
注意: SocketHolder
是一个不可变类。你可以忽略我拥有的 ZeroMQ
东西。
最佳答案
您使用以下同步技术。
updateLiveSockets()
方法是同步的(隐含在此),这将防止两个线程同时切换 map 。getNextSocket()
期间发生切换,则在使用 map 时对 map 进行本地引用以避免混淆。方法。是否像现在这样是线程安全的?
线程安全始终取决于共享可变数据上是否存在适当的同步。在这种情况下,共享的可变数据是数据中心到它们的 SocketHolder 列表的映射。
map 位于 AtomicReference
中的事实,并制作一个本地副本以供使用,在 map 上同步就足够了。您的方法采用 map 的一个版本并使用它,由于 AtomicReference
的性质,切换版本是线程安全的.这也可以通过为 map 制作成员字段来实现 volatile
,因为您所做的只是更新引用(您不对其执行任何先检查后执行的操作)。
作为scheduleAtFixedRate()
保证通过 Runnable
不会与自身同时运行,synchronized
在 updateLiveSockets()
不需要,但是,它也不会造成任何真正的伤害。
所以是的,这个类是线程安全的。
但是,尚不完全清楚 SocketHolder
可以被多个线程同时使用。实际上,此类只是试图尽量减少并发使用 SocketHolder
。 s 通过选择一个随机的活的(虽然不需要洗牌整个数组来选择一个随机索引)。它实际上并没有阻止并发使用。
能否提高效率?
我相信可以。查看 updateLiveSockets()
时方法,它似乎构建了完全相同的 map ,除了 SocketHolder
s 对于 isLive
可能有不同的值旗帜。这使我得出结论,与其切换整个 map ,我只想切换 map 中的每个列表。为了以线程安全的方式更改映射中的条目,我可以只使用 ConcurrentHashMap
.
如果我使用 ConcurrentHashMap
,并且不要切换 map ,而是 map 中的值,我可以摆脱 AtomicReference
.
要更改映射,我可以构建新列表并将其直接放入 map 中。这样效率更高,因为我可以更快地发布数据,创建更少的对象,而我的同步只是建立在现成的组件上,这有利于提高可读性。
这是我的构建(为简洁起见,省略了一些不太相关的部分)
public class SocketManager {
private static final Random random = new Random();
private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
private final Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter = new ConcurrentHashMap<>(); // use ConcurrentHashMap
private final ZContext ctx = new ZContext();
// ...
private SocketManager() {
connectToZMQSockets();
scheduler.scheduleAtFixedRate(this::updateLiveSockets, 30, 30, TimeUnit.SECONDS);
}
// during startup, making a connection and populate once
private void connectToZMQSockets() {
Map<Datacenters, List<String>> socketsByDatacenter = Utils.SERVERS;
for (Map.Entry<Datacenters, List<String>> entry : socketsByDatacenter.entrySet()) {
List<SocketHolder> addedColoSockets = connect(entry.getValue(), ZMQ.PUSH);
liveSocketsByDatacenter.put(entry.getKey(), addedColoSockets); // we can put it straight into the map
}
}
// ...
// this method will be called by multiple threads to get the next live socket
// is there any concurrency or thread safety issue or race condition here?
public Optional<SocketHolder> getNextSocket() {
for (Datacenters dc : Datacenters.getOrderedDatacenters()) {
Optional<SocketHolder> liveSocket = getLiveSocket(liveSocketsByDatacenter.get(dc)); // no more need for a local copy, ConcurrentHashMap, makes sure I get the latest mapped List<SocketHolder>
if (liveSocket.isPresent()) {
return liveSocket;
}
}
return Optional.absent();
}
// is there any concurrency or thread safety issue or race condition here?
private Optional<SocketHolder> getLiveSocket(final List<SocketHolder> listOfEndPoints) {
if (!CollectionUtils.isEmpty(listOfEndPoints)) {
// The list of live sockets
List<SocketHolder> liveOnly = new ArrayList<>(listOfEndPoints.size());
for (SocketHolder obj : listOfEndPoints) {
if (obj.isLive()) {
liveOnly.add(obj);
}
}
if (!liveOnly.isEmpty()) {
// The list is not empty so we shuffle it an return the first element
return Optional.of(liveOnly.get(random.nextInt(liveOnly.size()))); // just pick one
}
}
return Optional.absent();
}
// no need to make this synchronized
private void updateLiveSockets() {
Map<Datacenters, List<String>> socketsByDatacenter = Utils.SERVERS;
for (Map.Entry<Datacenters, List<String>> entry : socketsByDatacenter.entrySet()) {
List<SocketHolder> liveSockets = liveSocketsByDatacenter.get(entry.getKey());
List<SocketHolder> liveUpdatedSockets = new ArrayList<>();
for (SocketHolder liveSocket : liveSockets) { // LINE A
Socket socket = liveSocket.getSocket();
String endpoint = liveSocket.getEndpoint();
Map<byte[], byte[]> holder = populateMap();
Message message = new Message(holder, Partition.COMMAND);
boolean status = SendToSocket.getInstance().execute(message.getAdd(), holder, socket);
boolean isLive = (status) ? true : false;
SocketHolder zmq = new SocketHolder(socket, liveSocket.getContext(), endpoint, isLive);
liveUpdatedSockets.add(zmq);
}
liveSocketsByDatacenter.put(entry.getKey(), Collections.unmodifiableList(liveUpdatedSockets)); // just put it straigth into the map, the mapping will be updated in a thread safe manner.
}
}
}
关于java - 在单个后台线程定期修改 map 的同时读取 map ,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/46997971/
我遇到以下问题。我想读取一个包含数百万行和数百列的大型 csv。我想向下转换列的数据类型。我的方法是读取 csv,然后使用 pd.to_numeric() 对其进行向下转换。我不知道列数及其类型。在读
目前,我从 SQL server (2008) 数据库获取数据。 cyurrent的方法是使用DataTable,然后将其传递并使用。 if (parameters != null)
我有以下问题。我有一个巨大的 csv 文件,想用多处理加载它。对于一个包含 500000 行和 130 列不同数据类型的示例文件,Pandas 需要 19 秒。我试过 dask 因为我想多处理阅读。但
是否有关于用于序列化各种 MFC 数据结构的二进制格式的明确文档?我已经能够在十六进制编辑器中查看我自己的一些类,并使用 Java 的 ByteBuffer 类读取它们(使用自动字节顺序转换等)。 但
我正在使用 Selenium 进行测试,我们用 HTML 文件编写测试用例,并用它们制作测试套件,我们的要求是编写足够健壮的测试用例,以根据测试环境改变自身。 为此,我不希望在 HTML 脚本本身中包
我需要一个 JavaScript 代码来读取存储为 .txt 文件的字典(或者也可以保存为任何其他类型的文件。它也可以在线获得)并将其内容存储在一个变量中。我不能找到一种让 JavaScript 像
我正在尝试遍历包含 SSH 登录和其他日志的日志文本文件。 程序正在返回 SSH 登录的总数。 我的解决方案确实有效,但似乎有点慢(在 200mo 文件上大约需要 3.5 秒)。我想知道是否有任何方法
我正在将大量数据从一个电子表格复制到工作簿中的其他 160 个电子表格。目前,Excel (2013) 遇到错误,因为它没有足够的资源来完成操作。 我的目标是将工作表 4 中 V13:XI1150 范
我正在尝试读取一个有 1147 行的文本文件。下面的代码仅读取第 1050-1147 行。我的目标是读取整个文件并提取位于不同行的特定值以在脚本中使用。一个示例是包含“BlockList: 2”的行中
我正在为游戏编写解释器。用户将其移动输入解释器,程序执行该移动。 现在我想为每个决定实现一个时间限制。玩家不应该能够思考超过 30 秒来写一个移动并按下回车。 call_with_time_limit
以this file例如,我正在尝试读取 data.frame 中的数据。来自 the doc (pdf 文件,表 1),它遵循一些 fortran 约定。我尝试了以下但收效甚微: dir 0' 将
我正在使用 R 阅读 Outlook 附件。我的引用在这里:Download attachment from an outlook email using R 这是我的电子邮件的截图: 这每天都会发送
我不会从表格中读取行来将主题放在列表中 php脚本 $url_obj='http://'.$host.':8069/xmlrpc/object'; $sock=new xmlrpc_client($u
我有一个这样的 csv 文件: id,name,value 1,peter,5 2,peter\,paul,3 我如何读取此文件并告诉 R "\," 不表示新列,仅表示 ","。 我必须添加该文件
我正在尝试读取 ~/Library/Preferences/com.apple.mail.plist (在 Snow Leopard 上)以获取电子邮件地址和其他信息以进入“关于”对话框。我使用以下代
This question already has answers here: How do I use floating-point division in bash? (19个回答) 5个月前关闭
本练习的目标是读取输入文件并将其存储到表中,然后验证输入中的某些字段并输出任何错误记录。我需要读取并存储每个策略组,以便表中一次仅存储 5 条记录,而不是整个文件。 所以我需要读取一个包含 5 条记录
据我了解,LWT 插入始终以 SERIAL 一致性级别完成。如果为 true,这是否意味着读取作为 LWT 插入的行可以安全地以 ANY 的一致性级别读取? 换句话说,我假设 LWT 插入是完全一致的
我看到很多很多通过java脚本读取cookie的函数,但我只想在变量中使用它一次,我是JS新手。 这是我的代码 var TheNumber = (Math.random() + '') * 10000
我正在使用 asp.net 和 C#。我在服务器上部署了一个应用程序[已发布],现在我想查看该网站的代码,据我所知,我可以阅读程序集来查看代码。 请告诉我如何实现它。 提前致谢。 最佳答案 您可以使用
我是一名优秀的程序员,十分优秀!