9
मैं एक वर्ग है, जिसमें मैं एक नक्शा liveSocketsByDatacenter एक भी पृष्ठभूमि धागे से हर 30 सेकंड updateLiveSockets() विधि के अंदर पॉप्युलेट कर रहा हूँ और फिर मैं एक विधि 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));
}
}
के रूप में आप मेरी कक्षा में देख सकते हैं:
- एक भी पृष्ठभूमि धागा जो हर 30 सेकंड चलाता है से, मैं
updateLiveSockets()विधि में सभी को लाइव सॉकेट साथliveSocketsByDatacenterनक्शा पॉप्युलेट। - और फिर एकाधिक धागे से, मैं
getNextSocket()विधि को मुझे एक लाइव सॉकेट उपलब्ध कराने के लिए कॉल करता हूं जो आवश्यक जानकारी प्राप्त करने के लिएliveSocketsByDatacenterमानचित्र का उपयोग करता है।
मेरे पास कोई कोड बिना किसी समस्या के ठीक काम कर रहा है और यह देखना चाहता था कि यह लिखने के लिए कोई बेहतर या अधिक प्रभावी तरीका है या नहीं। मैं थ्रेड सुरक्षा मुद्दों या किसी भी दौड़ की स्थिति पर कोई राय प्राप्त करना चाहता था, लेकिन अब तक मैंने कोई भी नहीं देखा है लेकिन मैं गलत हो सकता हूं।
मैं अधिकतर updateLiveSockets() विधि और getLiveSocketX() विधि के बारे में चिंतित हूं। मैं liveSockets पर ListSocketHolder लाइन ए पर और फिर SocketHolder ऑब्जेक्ट बना रहा हूं और एक और नई सूची में जोड़ रहा हूं। क्या यह ठीक है?
नोट:SocketHolder एक अपरिवर्तनीय वर्ग है। और आप मेरे पास ZeroMQ सामानों को अनदेखा कर सकते हैं।
यह मुझे लगता है कि 'liveSocketsByDatacenter' ** अपरिवर्तनीय है ** है। यह सब बहुत आसान बनाने का एक तरीका है। –
प्लस आपका तर्क बहुत सारे स्थानों में बहुत गहराई से है। 'वैकल्पिक' का दुरुपयोग मेरी आँखें पानी बनाता है। मैं इसे पूरी तरह से छुटकारा पाउंगा - आप इसे कहीं भी सही तरीके से उपयोग नहीं करते हैं। संकेत: 'Optional.isPresent' को कॉल करना हमेशा एक बुरा विचार है। –
मेरे पास कुछ विधियां हैं जिनमें कुछ भी गलत हो सकता है। सबसे पहले '30% सेकेंड सेकेंड में बैकग्राउंड थ्रेड द्वारा बुलाया गया' अपडेट लाइवेट्स 'है, दूसरा' रीडर थ्रेट 'नामक' getNextSocket 'विधि है जो आंतरिक रूप से 'getLiveSocket' विधि को कॉल करता है, इसलिए इन सभी तीन तरीकों को मेरा विश्वास है कि थ्रेड सुरक्षा मुद्दों के संदर्भ में सही होना चाहिए। क्या आपको लगता है कि वे सभी सही चीजें कर रहे हैं? मैं 'updateLiveSockets' विधि में अधिक डरता हूं। – john