ICMP does not appear to be an option for you... you say you are pinging 'servers', but I think what you mean is that you are pinging 'services' (actual applications running on the server). You connect to the host:port combination. You could even have multiple services running on a single server.
A brief run-down of your code suggests:
- every 10 seconds (or so) you create a bunch of threads.
- each thread tries to connect to a server:port combination.
- if the connect fails, you remove the server
It has already been pointed out to you that the remove process should use some form of locking... using a simple ArrayList will lead to corrupt lists as each of the 100 threads may try to modify the list at the same time.
I suggest a different approach.
I recommend creating a thread for each server you want to monitor. As the server comes available, it adds the server to the 'alive' list. As the server 'drops', it removes it from the alive list.
The simple Socket connection to the server is all you need to worry about when it is alive. The monitor thread can sit there waiting for the socket to die. If it dies, it tries to reconnect. You can choose a different approach for monitoring the TCP state, but, the model of having a separate continuous thread for each monitored server, and the concept of having a maintained collection of 'alive' threads is the important part.
Consider the following class, which has a convenient Main method:
package ping;
import java.io.IOException;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class ServiceMonitor {
private final class Monitor implements Runnable {
private final InetSocketAddress tomonitor;
public Monitor(InetSocketAddress service) {
super();
this.tomonitor = service;
}
@Override
public void run() {
while (true) {
try (Socket remote = new Socket()) {
System.out.println("Trying to Connect to " + tomonitor);
remote.connect(tomonitor, 5*1000);
if (registerRemote(tomonitor, this)) {
try (InputStream is = remote.getInputStream()) {
byte[] buffer = new byte[512];
// This will likely never loop, simply wait for it to close.
while (is.read(buffer) >= 0) {
// still alive ...
}
} catch (IOException ioe) {
// unexpected condition, but this means the server is dead... that's all we need to know.
//ioe.printStackTrace();
} finally {
deregisterRemote(tomonitor, this);
}
}
} catch (SocketTimeoutException ste) {
// simply try again....
} catch (IOException e) {
// could not connect to remote?
try {
Thread.sleep(1000);
} catch (InterruptedException e1) {
return;
}
//e.printStackTrace();
}
}
}
}
private final ConcurrentHashMap<InetSocketAddress, Monitor> alive = new ConcurrentHashMap<>();
private final ThreadGroup tgroup = new ThreadGroup("Server Monitor Group");
private final Lock monlock = new ReentrantLock();
private final Condition idlecondition = monlock.newCondition();
private final Condition busycondition = monlock.newCondition();
public ServiceMonitor() {
}
public void monitorService(InetSocketAddress remote) {
if (!alive.containsKey(remote)) {
Thread mthread = new Thread(tgroup, new Monitor(remote), "Monitor Thread: " + remote.toString());
mthread.setDaemon(true);
mthread.start();
}
}
private boolean registerRemote(InetSocketAddress tomonitor, Monitor monitor) {
if (null == alive.putIfAbsent(tomonitor, monitor)) {
// do any notification if you need to that a server is available....
System.out.println("Register " + tomonitor + " Currently " + alive.size() + " alive");
monlock.lock();
try {
busycondition.signalAll();
} finally {
monlock.unlock();
}
return true;
}
return false;
}
private void deregisterRemote(InetSocketAddress tomonitor, Monitor monitor) {
if (alive.replace(tomonitor, monitor, monitor)) {
// we are currently the holder... good.
alive.remove(tomonitor);
// do any notification if you need to that a server is 'down'.
System.out.println("Deregister " + tomonitor + " Currently " + alive.size() + " alive");
monlock.lock();
try {
if (alive.isEmpty()) {
idlecondition.signalAll();
}
} finally {
monlock.unlock();
}
}
}
public void waitAllDead() {
monlock.lock();
try {
while (!alive.isEmpty()) {
idlecondition.await();
}
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
} finally {
monlock.unlock();
}
}
public void waitBusy() {
monlock.lock();
try {
while (alive.isEmpty()) {
busycondition.await();
}
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
} finally {
monlock.unlock();
}
}
public static void main(String[] args) {
ServiceMonitor monitor = new ServiceMonitor();
for (int i = 0; i < 10; i++) {
monitor.monitorService(new InetSocketAddress("localhost", 40000 + i));
}
monitor.waitBusy();
monitor.waitAllDead();
System.out.println("Complete!");
}
}
This class monitors services simply by their active TCP socket. It has a main method which monitors 10 ports from 40000 through 40009.
Here is a simple dummy server that creates services on those ports.... (and I used it for testing).
package ping;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
public class Service extends Thread {
private final Socket socket;
public Service(int port, Socket socket) {
super("Service for local:" + port + " remote:" + socket.getInetAddress().toString());
this.socket = socket;
}
@Override
public void run() {
try (
OutputStream os = socket.getOutputStream();
InputStream is = socket.getInputStream();
) {
final int delay = (1 + (int)(Math.random() * 5));
Thread.sleep(delay * 1000);
System.out.println(Thread.currentThread().getName() + " Sleeping for " + delay + " seconds.");
os.flush();
} catch (IOException | InterruptedException ioe) {
ioe.printStackTrace();
} finally {
System.out.println("Closing " + Thread.currentThread().getName());
}
}
public static void main(String[] args) throws NumberFormatException, IOException {
int cnt = 10;
for (int i = 0; i < cnt; i++) {
final int port = 40000 + i;
Runnable controller = new Runnable() {
public void run() {
try (
ServerSocket ss = new ServerSocket(port);
) {
int port = ss.getLocalPort();
System.out.println("Listening on port " + port);
Socket sock = null;
Random rand = ThreadLocalRandom.current();
while ((sock = ss.accept()) != null) {
Service service = new Service(port, sock);
sock = null;
service.start();
Thread.sleep(5000 + rand.nextInt(15000));
}
} catch (IOException | InterruptedException ioe) {
ioe.printStackTrace();
}
};
};
Thread thread = new Thread(controller, "Controller " + i);
thread.start();
}
}
}
